RabbitMQ 消息队列 学习笔记（2）

 1 #7512
 2 '''
 3 exchange用法，
 4 fanout广播,广播是实时的
 5 
 6 '''
 7 import pika,sys
 8 
 9 connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
10 channel = connection.channel()
11 # channel.queue_declare('msg')
12 
13 
14 #fanout广播
15 channel.exchange_declare(exchange='logs',
16                          exchange_type='fanout')
17 
18 message = ' '.join(sys.argv[1:]) or "info :哈哈"
19 # message ="哈哈"
20 channel.basic_publish(exchange='logs',
21                       routing_key= '',
22                       body=message,
23                       )
24 
25 print(" [x] Sent 'Hello World!'")
26 connection.close()

 1 #7512
 2 '''
 3 广播接收端
 4 
 5 '''
 6 import pika,time
 7 
 8 connection  = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
 9 channel = connection.channel()
10 
11 channel.exchange_declare(exchange='logs',
12                          exchange_type='fanout')
13 
14 
15 # exclusive排他,唯一的,不指定queue名字,rabbit会随机分配一个名字
16 #exclusive=True会在使用此queue的消费者断开后,自动将queue删除
17 result = channel.queue_declare(queue='',
18                                exclusive= True)
19 
20 print("result",result,"xxx",result.method.queue)
21 #取queue的名字
22 queue_name= result.method.queueresult.method.queue
23 
24 #绑定交换器exchange
25 channel.queue_bind(exchange='logs',
26                    queue=queue_name)
27 
28 
29 def recv(ch ,methrod ,proporties ,body):
30     print("recv msg : [%s]"%body.decode())
31     # time.sleep(20)
32     # ch.basic_ack(delivery_tag = methrod.delivery_tag)
33 
34 
35 # channel.basic_qos(prefetch_count=1)#最多能处理队列中的1条消息
36 channel.basic_consume(queue='',
37                       on_message_callback=recv ,
38                       auto_ack= True)
39 
40 print(' [*] Waiting for messages. To exit press CTRL+C')
41 channel.start_consuming()

 1 #7512
 2 
 3 import pika
 4 
 5 connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
 6 channel = connection.channel()
 7 
 8 channel.exchange_declare(exchange='direct_logs',
 9                          exchange_type="direct")
10 
11 severity = input(">>:")  or "info"
12 massage = input(">>:")  or "哈哈"
13 
14 channel.basic_publish(exchange="direct_logs",
15                       routing_key=severity,
16                       body = massage)
17 
18 print("[x] Send to %r : %r "%(severity,massage))
19 connection.close()

 1 #7512
 2 import pika,sys
 3 
 4 connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
 5 channel = connection.channel()
 6 
 7 channel.exchange_declare(exchange='direct_logs',
 8                          exchange_type="direct")
 9 
10 result = channel.queue_declare(queue="",
11                       exclusive=True)
12 
13 queue_name = result.method.queue
14 
15 serverities = sys.argv[1:]
16 if not serverities:
17     sys.exit()
18 print(serverities)
19 
20 for i in serverities:
21     channel.queue_bind(exchange='direct_logs',
22                  queue=queue_name,
23                  routing_key= i )
24 
25 
26 def callback(ch, method, properties, body):
27     print("ch:",ch, "\nmethod:",method,"\nproperties" ,properties)
28     print(" [x] Received %r" % body.decode())
29     ch.basic_ack(delivery_tag=method.delivery_tag)#手动回应
30 
31 channel.basic_consume(queue=queue_name,
32                       on_message_callback=callback,
33                       auto_ack=False)
34 
35 channel.start_consuming()

 1 #7512
 2 
 3 import pika,sys
 4 
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
 6 
 7 channel = connection.channel()
 8 channel.exchange_declare(exchange="topic_logs",
 9                          exchange_type="topic")
10 
11 key = sys.argv[1:]  if len(sys.argv) > 1 else input(">>:")
12 
13 message =  input(">>:")
14 channel.basic_publish(exchange='topic_logs',
15                       routing_key=key,
16                       body=message
17                       )
18 
19 print("[x] sent [%s] : a msg is [%s]"%(key,message))
20 channel.start_consuming()

 1 #7512
 2 '''
 3 #收所有
 4 '''
 5 import pika,sys
 6 
 7 connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
 8 channel = connection.channel()
 9 
10 channel.exchange_declare(exchange="topic_logs",
11                          exchange_type="topic")
12 
13 result = channel.queue_declare(queue='',
14                       exclusive=True)
15 
16 queue_name = result.method.queue
17 
18 massage = sys.argv[1:] or input(">>:").split()
19 for msg in massage:
20     channel.queue_bind(queue=queue_name,
21                        exchange="topic_logs",
22                        routing_key= msg)
23 
24 def callback(ch, method, properties, body):
25     print(" [x] Received %r" % body.decode())
26     ch.basic_ack(delivery_tag=method.delivery_tag)#手动回应
27 
28 channel.basic_consume(queue=queue_name,
29                       on_message_callback=callback,
30                       auto_ack=False)
31 
32 print("[x]  wait the msg...")
33 channel.start_consuming()