RabbitMQ中的topic exchange将路由键和某模式进行匹配，从而类似于正则匹配的方式去接收喜欢的信息。

topic exchange

如果想使用topic模式，那么可以随意设置routing_key。相反，需要按照一定的要求设定该值。

routing_key在topic模式中应该选择一组拥有特定属性的单词作为该值。

1
2

\* (star) can substitute for exactly one word.
# (hash) can substitute for zero or more words.

例如，如果生产者的routing_key设置为test1.test2.test3，那么消费着中绑定消息队列的routing_key必须可以匹配生产者的routing_key。

1
2
3
4
5
6
7
8
9

#生产者
routing_key = 'test1.test2.test3'
channel.basic_publish(exchange='topic_test', routing_key=routing_key, body=message)

#消费者
routing_key = 'test1.*' #可以
routing_key = '*.test2.*' #可以
routing_key = 'test3' #不可以
channel.queue_bind(exchange='topic_logs', queue=queue_name, routing_key=binding_key)

例子

生产者如下，会依次设置routing_key为A和B，那么需要设置两个消费者的routing_key来分别读取消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

#!/usr/bin/env python
# coding=utf-8
import pika
import sys
import time

connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='topic_test',
type='topic')

message = "test "
for i in range(20):
    for item in ['A', 'B']:
        routing_key = item
        channel.basic_publish(exchange='topic_test',routing_key=routing_key, body=message+item)
        print " [x] Sent %r:%r" % (routing_key, message)
        time.sleep(2)
connection.close()

消费者如下，启动命令分别为:

1
2

python receive.py A
python receive.py B

消费者如下：

#!/usr/bin/env python
# coding=utf-8

import pika
import sys

def callback(ch, method, properties, body):
print " [x] %r:%r" % (method.routing_key, body,)

connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='topic_test',
type='topic')

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

binding_key = sys.argv[1]
print "Usage: %s [binding_key]..." % (sys.argv[1])

channel.queue_bind(exchange='topic_test', queue=queue_name, routing_key=binding_key)
print ' [*] Waiting for logs. To exit press CTRL+C'
channel.basic_consume(callback, queue=queue_name, no_ack=True)

channel.start_consuming()