Python并发编程-RabbitMq消息队列

1 channel.basic_qos(prefetch_count=1)

 1 #!/usr/bin/env python
 2 import pika
 3 import sys
 4  
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(
 6         host=‘localhost‘))
 7 channel = connection.channel()
 8 #队列持久化
 9 channel.queue_declare(queue=‘task_queue‘, durable=True)
10 
11 message = ‘ ‘.join(sys.argv[1:]) or "Hello World!"
12 #消息持久化
   channel.basic_publish(exchange=‘‘,
13                       routing_key=‘task_queue‘,
14                       body=message,
15                       properties=pika.BasicProperties(
16                          delivery_mode = 2, # make message persistent
17                       ))
18 print(" [x] Sent %r" % message)
19 connection.close()

 1 #!/usr/bin/env python
 2 import pika
 3 import time
 4  
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(
 6         host=‘localhost‘))
 7 channel = connection.channel()
 8  
 9 channel.queue_declare(queue=‘task_queue‘, durable=True)
10 print(‘ [*] Waiting for messages. To exit press CTRL+C‘)
11  
12 def callback(ch, method, properties, body):
13     print(" [x] Received %r" % body)
14     time.sleep(body.count(b‘.‘))
15     print(" [x] Done")
16     ch.basic_ack(delivery_tag = method.delivery_tag)
17  
18 channel.basic_qos(prefetch_count=1)
19 channel.basic_consume(callback,
20                       queue=‘task_queue‘)
21  
22 channel.start_consuming()

 1 import pika
 2 
 3 credentials = pika.PlainCredentials(‘nulige‘, ‘123456‘)
 4 # connection = pika.BlockingConnection(pika.ConnectionParameters(host=url_1,
 5 #                                      credentials=credentials, ssl=ssl, port=port))
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(
 7     host=‘192.168.1.118‘,credentials=credentials))
 8 
 9 channel = connection.channel()
10 
11 # 声明queue
12 channel.queue_declare(queue=‘nulige‘,durable=True)
13 
14 # n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
15 channel.basic_publish(exchange=‘‘, 
16                       routing_key=‘nulige‘, #send msg to this queue
17                       body=‘Hello World!23‘,
18                       properties=pika.BasicProperties(
19                           delivery_mode=2,  # make message persistent
20                       )
21                       )
22 
23 
24 print(" [x] Sent ‘Hello World!2‘")
25 connection.close()

 1 import pika
 2 import time
 3 credentials = pika.PlainCredentials(‘nulige‘, ‘123456‘)
 4 connection = pika.BlockingConnection(pika.ConnectionParameters(
 5     host=‘192.168.1.118‘,credentials=credentials))
 6 
 7 channel = connection.channel()
 8 # You may ask why we declare the queue again ? we have already declared it in our previous code.
 9 # We could avoid that if we were sure that the queue already exists. For example if send.py program
10 # was run before. But we‘re not yet sure which program to run first. In such cases it‘s a good
11 # practice to repeat declaring the queue in both programs.
12 channel.queue_declare(queue=‘nulige‘,durable=True)
13 
14 
15 def callback(ch, method, properties, body):
16     print(ch, method, properties)
17 
18     print(" [x] Received %r" % body)
19     time.sleep(1)
20 
21 
22 channel.basic_consume(callback,
23                       queue=‘nulige‘,
24                       #no_ack=True
25                       )
26 channel.basic_qos(prefetch_count=1)
27 print(‘ [*] Waiting for messages. To exit press CTRL+C‘)
28 channel.start_consuming()

 1 import pika
 2 import sys
 3 
 4 credentials = pika.PlainCredentials(‘nulige‘, ‘123456‘)
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(
 6     host=‘192.168.1.118‘,credentials=credentials))
 7 
 8 channel = connection.channel()
 9 channel.exchange_declare(exchange=‘logs‘, type=‘fanout‘) #发送消息类型为fanout,就是给所有人发消息
10 
11 #如果等于空，就输出hello world!
12 message = ‘ ‘.join(sys.argv[1:]) or "info: Hello World!"
13 
14 
15 channel.basic_publish(exchange=‘logs‘,   
16                       routing_key=‘‘,
17                       body=message)
18 
19 print(" [x] Sent %r" % message)
20 connection.close()

 1 import pika
 2 
 3 credentials = pika.PlainCredentials(‘nulige‘, ‘123456‘)
 4 connection = pika.BlockingConnection(pika.ConnectionParameters(
 5     host=‘192.168.1.118‘,credentials=credentials))
 6 
 7 channel = connection.channel()
 8 
 9 channel.exchange_declare(exchange=‘logs‘,type=‘fanout‘)#指定发送类型
10 #必须能过queue来收消息
11 result = channel.queue_declare(exclusive=True)  # 不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
12 
13 queue_name = result.method.queue
14 
15 channel.queue_bind(exchange=‘logs‘,queue=queue_name)  #随机生成的Q，绑定到exchange上面。
16 
17 print(‘ [*] Waiting for logs. To exit press CTRL+C‘)
18 
19 def callback(ch, method, properties, body):
20     print(" [x] %r" % body)
21 
22 
23 channel.basic_consume(callback,
24                       queue=queue_name,
　　　　　　　　　　　　　　　#no_ack=True,
25                       )
26 channel.start_consuming()

 1 import pika
 2 import sys
 3 
 4 credentials = pika.PlainCredentials(‘nulige‘, ‘123456‘)
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(
 6     host=‘192.168.1.118‘,credentials=credentials))
 7 
 8 channel = connection.channel()
 9 
10 channel.exchange_declare(exchange=‘direct_logs‘,type=‘direct‘) #指定类型
11 
12 severity = sys.argv[1] if len(sys.argv) > 1 else ‘info‘  #严重程序，级别；判定条件到底是info，还是空，后面接消息
13 
14 message = ‘ ‘.join(sys.argv[2:]) or ‘Hello World!‘  #消息
15 
16 channel.basic_publish(exchange=‘direct_logs‘,
17                       routing_key=severity, #绑定的是：error  指定关键字（哪些队列绑定了，这个级别，那些队列就可以收到这个消息）
18                       body=message)
19 
20 print(" [x] Sent %r:%r" % (severity, message))
21 connection.close()

 1 import pika
 2 import sys
 3 
 4 credentials = pika.PlainCredentials(‘nulige‘, ‘123456‘)
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(
 6     host=‘192.168.1.118‘,credentials=credentials))
 7 channel = connection.channel()
 8 
 9 channel.exchange_declare(exchange=‘direct_logs‘,type=‘direct‘)
10 result = channel.queue_declare(exclusive=True)
11 queue_name = result.method.queue
12 
13 severities = sys.argv[1:]   #接收那些消息（指info，还是空），没写就报错
14 if not severities:
15     sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0]) #定义了三种接收消息方式info,warning,error
16     sys.exit(1)
17 
18 for severity in severities: #[error  info  warning]，循环severities
19     channel.queue_bind(exchange=‘direct_logs‘,
20                        queue=queue_name,
21                        routing_key=severity)  #循环绑定关键字
22 print(‘ [*] Waiting for logs. To exit press CTRL+C‘)
23 
24 def callback(ch, method, properties, body):
25     print(" [x] %r:%r" % (method.routing_key, body))
26 
27 channel.basic_consume(callback,queue=queue_name,)
28 channel.start_consuming()

 1 首先，设置接收类型为：info、warning、 error 三个中的其中一种或多种类型，再从发送端指定发送给那种类型，后面再接要发送的消息。
 2  
 3 #接收端
 4 D:\python\day42>python3 rabbit_direct_recv.py info error  #指定接收类型为info、erron
 5 
 6  [*] Waiting for logs. To exit press CTRL+C
 7  [x] ‘error‘:b‘err_hpappend‘  #接收到的消息
 8 
 9 D:\python\day42>python3 rabbit_direct_recv.py info warning  #指定接收类型为warning
10 
11  [*] Waiting for logs. To exit press CTRL+C
12  [x] ‘warning‘:b‘nulige‘  #接收到的消息
13 
14 
15 #发送端                                        发送类型   消息
16 D:\python\day42>python3 rabbit_direct_send.py error err_hpappend
17  [x] Sent ‘error‘:‘err_hpappend‘

  18 D:\python\day42>python3 rabbit_direct_send.py warning nulige
  19 [x] Sent ‘warning‘:‘nulige‘

 1 import pika
 2 import sys
 3 
 4 credentials = pika.PlainCredentials(‘nulige‘, ‘123456‘)
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(
 6     host=‘192.168.1.118‘,credentials=credentials))
 7 
 8 channel = connection.channel()
 9 
10 channel.exchange_declare(exchange=‘topic_logs‘,type=‘topic‘) #指定类型
11 
12 routing_key = sys.argv[1] if len(sys.argv) > 1 else ‘anonymous.info‘
13 
14 message = ‘ ‘.join(sys.argv[2:]) or ‘Hello World!‘  #消息
15 
16 channel.basic_publish(exchange=‘topic_logs‘,
17                       routing_key=routing_key,
18                       body=message)
19 print(" [x] Sent %r:%r" % (routing_key, message))
20 connection.close()

 1 import pika
 2 import sys
 3 
 4 credentials = pika.PlainCredentials(‘nulige‘, ‘123456‘)
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(
 6     host=‘192.168.1.118‘,credentials=credentials))
 7 
 8 channel = connection.channel()
 9 channel.exchange_declare(exchange=‘topic_logs‘,type=‘topic‘)
10 
11 result = channel.queue_declare(exclusive=True)
12 queue_name = result.method.queue
13 
14 binding_keys = sys.argv[1:]
15 if not binding_keys:
16     sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
17     sys.exit(1)
18 
19 for binding_key in binding_keys:
20     channel.queue_bind(exchange=‘topic_logs‘,
21                        queue=queue_name,
22                        routing_key=binding_key)
23 
24 print(‘ [*] Waiting for logs. To exit press CTRL+C‘)
25 
26 def callback(ch, method, properties, body):
27     print(" [x] %r:%r" % (method.routing_key, body))
28 
29 channel.basic_consume(callback,queue=queue_name)
30 
31 channel.start_consuming()

 1 #接收端
 2 D:\python\day42>python3 rabbit_topic_recv.py error
 3  [*] Waiting for logs. To exit press CTRL+C
 4  [x] ‘error‘:b‘mysql has error‘
 5 
 6 
 7 D:\python\day42>python3 rabbit_topic_recv.py *.warning mysql.*
 8  [*] Waiting for logs. To exit press CTRL+C
 9  [x] ‘mysql.error‘:b‘mysql has error‘
10 
11 
12 D:\python\day42>python3 rabbit_topic_send.py mysql.info "mysql has error"
13  [x] Sent ‘mysql.info‘:‘mysql has error‘
14 
15 
16 D:\python\day42>python3 rabbit_topic_recv.py *.error.*
17  [*] Waiting for logs. To exit press CTRL+C
18  [x] ‘mysql.error.‘:b‘mysql has error‘
19 
20 
21 #发送端                                指定类型:error      消息内容
22 D:\python\day42>python3 rabbit_topic_send.py error "mysql has error"
23  [x] Sent ‘error‘:‘mysql has error‘
24 
25 
26 D:\python\day42>python3 rabbit_topic_send.py mysql.error "mysql has error"
27  [x] Sent ‘mysql.error‘:‘mysql has error‘
28  [x] ‘mysql.info‘:b‘mysql has error‘
29 
30 
31 D:\python\day42>python3 rabbit_topic_send.py mysql.error. "mysql has error"
32  [x] Sent ‘mysql.error.‘:‘mysql has error‘

 1 import pika
 2 import uuid
 3 
 4 class SSHRpcClient(object):
 5     def __init__(self):
 6         credentials = pika.PlainCredentials(‘nulige‘, ‘123456‘)
 7         self.connection = pika.BlockingConnection(pika.ConnectionParameters(
 8                             host=‘192.168.1.118‘,credentials=credentials))
 9 
10         self.channel = self.connection.channel()
11 
12         result = self.channel.queue_declare(exclusive=True) #客户端的结果必须要返回到这个queue
13         self.callback_queue = result.method.queue
14 
15         self.channel.basic_consume(self.on_response,queue=self.callback_queue) #声明从这个queue里收结果
16 
17     def on_response(self, ch, method, props, body):
18         if self.corr_id == props.correlation_id: #任务标识符
19             self.response = body
20             print(body)
21 
22     # 返回的结果，放在callback_queue中
23     def call(self, n):
24         self.response = None
25         self.corr_id = str(uuid.uuid4()) #唯一标识符
26         self.channel.basic_publish(exchange=‘‘,
27                                    routing_key=‘rpc_queue3‘,  #声明一个Q
28                                    properties=pika.BasicProperties(
29                                        reply_to=self.callback_queue,
30                                        correlation_id=self.corr_id,
31                                    ),
32                                    body=str(n))
33 
34         print("start waiting for cmd result ")
35         count = 0
36         while self.response is None: #如果命令没返回结果
37             print("loop ",count)
38             count +=1
39             self.connection.process_data_events() #以不阻塞的形式去检测有没有新事件
40             #如果没事件，那就什么也不做， 如果有事件，就触发on_response事件
41         return self.response
42 
43 ssh_rpc = SSHRpcClient()
44 
45 print(" [x] sending cmd")
46 response = ssh_rpc.call("ipconfig")
47 
48 print(" [.] Got result ")
49 print(response.decode("gbk"))

 1 import pika
 2 import time
 3 import subprocess
 4 
 5 credentials = pika.PlainCredentials(‘nulige‘, ‘123456‘)
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(
 7     host=‘192.168.1.118‘, credentials=credentials))
 8 
 9 channel = connection.channel()
10 channel.queue_declare(queue=‘rpc_queue3‘)
11 
12 def SSHRPCServer(cmd):
13 
14     print("recv cmd:",cmd)
15     cmd_obj = subprocess.Popen(cmd.decode(),shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
16 
17     result = cmd_obj.stdout.read() or cmd_obj.stderr.read()
18     return result
19 
20 def on_request(ch, method, props, body):
21 
22     print(" [.] fib(%s)" % body)
23     response = SSHRPCServer(body)
24 
25     ch.basic_publish(exchange=‘‘,
26                      routing_key=props.reply_to,
27                      properties=pika.BasicProperties(correlation_id= 28                                                          props.correlation_id),
29                      body=response)
30 
31 channel.basic_consume(on_request, queue=‘rpc_queue3‘)
32 print(" [x] Awaiting RPC requests")
33 channel.start_consuming()