这几天研究了一下rabbitmq的工作模式，只有了解了工作模式，你才能更好的对应需求，网上有很多文档，我就结合代码搬运一下。
一.生产-消费者模式(queue，非常简单的模式，一发一收)
生产者
#!/usr/bin/env python
import pika
parameters = pika.ConnectionParameters('localhost')
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
channel.queue_declare(queue='hello')
number = 0
while number < 5:
channel.basic_publish(exchange='', routing_key='hello', body="hello world: {}".format(number))
print(" [x] Sent {}".format(number))
number += 1
connection.close()
消费者
#!/usr/bin/env python
import pika
import datetime
parameters = pika.ConnectionParameters('localhost')
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
channel.queue_declare(queue='hello')
def callback(ch, method, properties, body): # 定义一个回调函数，用来接收生产者发送的消息
print("{} [消费者] recv {}".format(datetime.datetime.now(), body))
channel.basic_consume(callback, queue='hello', no_ack=True)
print('{} [消费者] waiting for msg .'.format(datetime.datetime.now()))
channel.start_consuming() # 开始循环取消息
二.工作队列模式(routing key + queue,分发，多个worker处理不同的任务)
生产者
#!/usr/bin/env python
import pika
import sys
parameters = pika.ConnectionParameters(host='localhost')
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
for i in range(10):
message = 'Hello World: {}'.format(i)
channel.basic_publish(exchange='',
routing_key='task_queue',
body=message,
properties=pika.BasicProperties(delivery_mode=2))
print(" [x] Sent %r " % message)
connection.close()
消费者
#!/usr/bin/env python
import pika
import time
parameters = pika.ConnectionParameters(host='localhost')
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
print(' [*] Warting for messages. To exit press CTRL+C')
def call_back(ch, method, properties, body):
print(" [x] Received %r" % body)
time.sleep(body.count(b'.'))
print(" [x] Done")
ch.basic_ack(delivery_tag=method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(call_back, queue='task_queue')
channel.start_consuming()
生产者打印输出：
Task：
[x] Sent 'Hello World: 0'
[x] Sent 'Hello World: 1'
[x] Sent 'Hello World: 2'
[x] Sent 'Hello World: 3'
[x] Sent 'Hello World: 4'
[x] Sent 'Hello World: 5'
[x] Sent 'Hello World: 6'
[x] Sent 'Hello World: 7'
[x] Sent 'Hello World: 8'
[x] Sent 'Hello World: 9'
woerk1输出：
[*] Warting for messages. To exit press CTRL+C
[x] Received b'Hello World: 0'
[x] Done
[x] Received b'Hello World: 2'
[x] Done
[x] Received b'Hello World: 4'
[x] Done
[x] Received b'Hello World: 6'
[x] Done
[x] Received b'Hello World: 7'
[x] Done
[x] Received b'Hello World: 9'
[x] Done
worker2输出：
[*] Warting for messages. To exit press CTRL+C
[x] Received b'Hello World: 1'
[x] Done
[x] Received b'Hello World: 3'
[x] Done
[x] Received b'Hello World: 5'
[x] Done
[x] Received b'Hello World: 8'
[x] Done
三.fanout：所有绑定到exchange的queue都可以接收消息
消息发布端：
# -- coding:utf-8 --
author = "MuT6 Sch01aR"
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
channel = connection.channel()
channel.exchange_declare(exchange='fanout_logs', exchange_type='fanout')
message = 'Hello World!'
channel.basic_publish(exchange='fanout_logs',
routing_key='',
body=message)
print('发送消息：', message)
connection.close()
消息订阅端:
# -- coding:utf-8 --
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
channel = connection.channel()
channel.exchange_declare(exchange='fanout_logs', exchange_type='fanout')
result = channel.queue_declare(exclusive=True) # 不指定queue名字,rabbitmq会随机分配一个名字
# exclusive=True会在使用此queue的消息订阅端断开后,自动将queue删除
queue_name = result.method.queue
print('当前queue名称：', queue_name)
channel.queue_bind(exchange='fanout_logs', queue=queue_name)
def callback(ch, method, properties, body):
print(ch, method, properties)
print('收到数据：', body)
channel.basic_consume(callback,
queue=queue_name,
no_ack=True,
)
print('开始等待消息')
channel.start_consuming()
消息发布端需要在消息订阅端运行之后运行，不然消息订阅端收不到消息
开启3个消息订阅端和一个消息发布端
消息发布端发布消息
3个消息订阅端会同时接收到消息
direct：指定的queue才能收到消息
# -- coding:utf-8 --
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
severity = sys.argv[1] if len(sys.argv) > 1 else 'info' # 关键字
message = 'Hello World!'
channel.basic_publish(exchange='direct_logs',
routing_key=severity,
body=message)
print('级别 [%s] 发送数据 [%s]' % (severity, message))
connection.close()
消息订阅端
# -- coding:utf-8 --
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
severities = sys.argv[1:] # 关键字
for severity in severities:
channel.queue_bind(exchange='direct_logs',
queue=queue_name,
routing_key=severity)
def callback(ch, method, properties, body):
print(ch, method, properties)
print('收到来自%s的消息：%s' % (method.routing_key, body))
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
print('开始等待消息')
channel.start_consuming()
开启3个消息订阅端，分别引用关键字info、warning，warning、error，error、info，开启一个消息发布端，引用关键字wanrning
只有引用关键字warning的消息订阅端才收到消息
没有引用关键字warning的就收不到消息
topic：符合条件的queue才能收到消息
消息订阅端指定条件，符合条件的queue才能被消息订阅端收到消息
会匹配任意个任意字符串，*会匹配一个任意字符串
单个#会匹配全部字符串，单个*无法匹配
消息发布端
# -- coding:utf-8 --
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = 'Hello World!'
channel.basic_publish(exchange='topic_logs',
routing_key=severity,
body=message)
print('给 [%s] 发送消息 [%s]' % (severity, message))
connection.close()
消息订阅端
# -- coding:utf-8 --
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
keys = sys.argv[1:]
for key in keys:
channel.queue_bind(exchange='topic_logs',
queue=queue_name,
routing_key=key)
def callback(ch, method, properties, body):
print(ch, method, properties)
print('来自 [%s] 的信息 [%s]' % (method.routing_key, body))
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
print('开始接收消息')
channel.start_consuming()
开启3个消息订阅端和一个消息发布端
3个消息订阅端，一个*.exe匹配以.exe结尾的，一个#匹配全部字符串，一个python.exe只匹配字符串为python.exe的
消息发布端引用关键字qq.exe
image.png
很显然，匹配以.exe结尾和匹配全部字符串的消息订阅端将会收到消息，只匹配字符串为python.exe的不会收到消息
image.png
image.png
image.png
这里也有一个很棒的博客地址，大家也可以看看。
https://www.cnblogs.com/pangguoping/p/5720134.html
