上一篇博客我们没有介绍完rabbitmq，今天我们接着上一篇的博客继续介绍rabbitmq

这边的博客的内容如下

1、组播，对指定的队列设置关键词，通过关键词来控制消息的分发

2、更加细致的组播

 

先来介绍组播

其实组播和广播只有3个区别

a、组播的exchange的类型是“direct”

b、组播的生产端不需要绑定queue，只需要申明exchange，然后在发布消息的时指定route_key、exchange、message即可

c、消费者者需要绑定队列到exchange上，且也需要指定route_key，只有绑定到和生产者端相同的exchange的队列【队列名称可以不一样】，且生产端的route_key一致就可以收到消息

先看生产端的核心代码

import pikatest_connectio = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))test_channel = test_connectio.channel()test_channel.exchange_declare(exchange="logs",                              exchange_type="direct")#在队列中什么一个exchange，类型为“direct”test_channel.queue_declare(queue="zhang1",durable=True)test_channel.queue_declare(queue="zhang2",durable=True)test_channel.queue_declare(queue="zhang3",durable=True)test_channel.basic_publish(exchange="logs",                           routing_key="error",                           body="this is 组播")#发布消息，将消息发往exchange名称为“logs”中，且指定的routing_key为error的队列中test_channel.close()

　

在看消费者的核心代码

import pikatest_connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))test_channel = test_connection.channel()test_channel.exchange_declare(exchange="logs",                              exchange_type="direct")#在管道中申明一个exhcangetest_channel.queue_declare(queue="zhang2",durable=True)#在管道中申明一个队列test_channel.queue_bind(exchange="logs",                        queue="zhang2",                        routing_key="info")#将队列绑定到exchange上，并指明我这个消费者的routing_key，只有和生产端相同#routing_key的消费者才可以收到生产者的消息，这里可以绑定多个routing_keydef callback(ch,method,properties,body):    print("server:",body)test_channel.basic_consume(callback,                      queue="zhang2",                      no_ack=True)test_channel.start_consuming()

　

 

 

二、这里我们看下更细致的消息控制

更细致的消息控制是什么意思呢？比如我们实现这样一个场景；消费者可以接受*.error的日志，还可以接受apache.info的日志

更细致的消息控制和组播有2个不同

1、更细致的消息控制用到的exchange的类型为topic

2、routing_key的用法用法不一样

下面我们看下生产者的核心代码

import pikaimport systest_connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))test_channel = test_connection.channel()test_channel.exchange_declare(exchange="topic_logs",                              exchange_type="topic")route_key = sys.argv[1] if len(sys.argv) > 1 else "anonymous.info"message = "".join(sys.argv[2:]) or "hello world"test_channel.basic_publish(exchange="topic_logs",                           routing_key=route_key,                           body=message)print("[x] send %r %r" %(route_key,message))test_connection.close()

　　

在看消费者的核心代码

import pikaimport systest_connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))test_channel = test_connection.channel()test_channel.exchange_declare(exchange="topic_logs",                              exchange_type="topic")result = test_channel.queue_declare(exclusive=True)queue_name = result.method.queuebinding_keys = sys.argv[1:]if not binding_keys:    sys.stderr.write("Usage: %s [binding_key]....\n" %sys.argv[0])    sys.exit(0)for binding_key in binding_keys:    test_channel.queue_bind(exchange="topic_logs",                            queue=queue_name,                            routing_key=binding_key)print("waiting for message")def callback(ch,method,properties,body):    print("[x] %s" %(body))test_channel.basic_consume(callback,                           queue=queue_name,                           no_ack=True)test_channel.start_consuming()

 

我们可以做如下测试

在生产者端执行下面的命令，将消息发往mysql.error，消息的内容是xxxxxxxx

python rabbitmq_send.py mysql.error xxxxxxxxx

 

在消费者端执行下面的命令，该消费者接受所有的*.info的消息，和mysql.info的消息

python rabbitmq_receive.py *.info mysql.error