RabbitMQ 消息队列
rabbitmq是一个在amqp基础上完整的,可复用的企业消息系统。他遵循mozilla public license开源协议。
mq全称为message queue, 消息队列(mq)是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息(针对应用程序的数据)来通信,而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信,而不是通过直接调用彼此来通信,直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。rabbitmq可以,多个程序同时使用rabbitmq ,但是必须队列名称不一样。采用erlang语言,属于爱立信公司开发的。
术语(jargon)
- p(producing):制造和发送信息的一方。
- queue:消息队列。
- c(consuming):接收消息的一方。
1. 安装
ubuntu 上安装
- 添加源、新增公钥(不加会有警告)、更新源,安装:
rabbitmq-server
echo 'deb http://www.rabbitmq.com/debian/ testing main' | sudo tee /etc/apt/sources.list.d/rabbitmq.list wget -o- https://www.rabbitmq.com/rabbitmq-release-signing-key.asc | sudo apt-key add - sudo apt-get update sudo apt-get install rabbitmq-server
- 安装完成后还要配置下:
# 在 rabbitmq 中添加用户 hj@hj:~$ sudo rabbitmqctl add_user username password creating user "hj" # 这为设置成功后的提示,同下 # 将用户设置为管理员(只有管理员才能远程登录) hj@hj:~$ sudo rabbitmqctl set_user_tags username administrator setting tags for user "hj" to [administrator] # 为用户设置读写权限 hj@hj:~$ sudo rabbitmqctl set_permissions -p / username ".*" ".*" ".*" setting permissions for user "username" in vhost "/"
windows 上安装
- 安装
pika
pip3 install -i http://pypi.douban.com/simple/ pika --trusted-host pypi.douban.com
- rabbitmq 是建立在 erlang otp 平台上,所有需要下载 erlang 和 rabbitmq,官网上下载安装
erlang
和rabbitmq
- erlang:http://www.erlang.org/downloads
- rabbitmq:https://www.rabbitmq.com/install-windows.html
- 将 erlang 添加到系统环境变量中
新建一个 erlang_home,值为 erlang 的安装路径(有些安装时会自动添加):
将 erlang_home 添加到 path 中(这里以 win10 平台为例,其他平台可能会不一样):
打开 cmd
以管理员身份证运行,输入 erl
检查 erlang 是否安装成功:
c:\windows\system32>erl eshell v10.3 (abort with ^g) # 版本 1> # 标识符
- rabbitmq 需要开启后台管理插件
rabbitmq management
2. 队列通信
2.1 简单示例
下面我们来使用 rabbitmq
来实现一个简单的消息收发:
- 发送端:一台 windows 机器
- 接收端:一台 ubuntu 虚拟机
消息不能直接发送到队列,而是需要经过 exchange 转发器转发,只有与转发器绑定了的队列,才能收到消息。在这里我们假设不经过 exchange 转发:
- 发送端:
import pika credentials = pika.plaincredentials('username', 'password') connection = pika.blockingconnection(pika.connectionparameters( '192.168.21.xxx', credentials=credentials)) channel = connection.channel() # 建立 rabbit 协议通道 # 声明queue channel.queue_declare(queue='hello') # n rabbitmq a message can never be sent directly to the queue, it always needs to go through an exchange. # 消息不能直接发送到队列,而是需要经过 exchange 转发器转发,只有与转发器绑定了的队列,才能收到消息 channel.basic_publish(exchange='', routing_key='hello', body=b'hello world!') print(" [x] sent 'hello world!'") connection.close()
首先需要输入上面第一章中已经注册的 rabbitmq
账户,然后再连接远程端。
其次再声明了一个队列 queue
,名称为 hello
,在这里 exchange 为空,发送的内容 body
必须是 bytes
类型。
- 接收端:
接收端也必须指定队列名称:
import pika import time credentials = pika.plaincredentials('username', 'password') connection = pika.blockingconnection(pika.connectionparameters( '192.168.21.128', credentials=credentials)) channel = connection.channel() # 建立 rabbit 协议通道 channel.queue_declare(queue='hello') def callback(ch, method, properties, body): print(" [x] received %r" % body) time.sleep(20) print(" [x] msg process done %r" % body) channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=true) print(' [*] waiting for messages. to exit press ctrl+c') channel.start_consuming()
运行结果如下:
2.2 消息持久化
我们已经知道即使消费者死亡,消息(队列)也不会丢失(在禁用 no_ack=true的前提下,现在是 auto_ack=true)
但是如果 rabbitmq
服务器停止,我们的任务一样会丢失,当 rabbitmq
退出或奔溃时,将会忘记队列和消息,除非我们告诉它不要这样,那么我们就要将队列和消息标记为持久。
- 确保
rabbitmq
永远不会丢失我们的队列,需要设置durable=true
:
# 发送端,即消息制造者 channel.queue_declare(queue='task_queue', durable=true)
- 将消息标记为持久性:
# 发送端,即消息制造者 properties=pika.basicproperties( delivery_mode=2, # make message persistent 使消息持久 )
设置好之后,发送端先发送一条消息,接收端先不要启动。使用以下命令关闭启动 rabbitmq
服务,观察队列和消息会不会真正丢失:
# 若命令运行失败,可以尝试使用 管理员模式 sudo # 启动rabbitmq service rabbitmq-server start # 停止rabbitmq service rabbitmq-server stop # 重启rabbitmq service rabbitmq-server restart # 查看当前活动的队列 rabbitmqctl list_queues
2.3 公平分发
所谓公平分发即一个生产者,多个消费者,类似于负载均衡。
下面我将设置一个发送端,两个接收端:
- 发送端:
import pika import time import sys credentials = pika.plaincredentials('username', 'password') connection = pika.blockingconnection(pika.connectionparameters( '192.168.21.128', credentials=credentials)) channel = connection.channel() # 建立 rabbit 协议通道 # 声明queue channel.queue_declare(queue='task_queue', durable=true) message = ' '.join(sys.argv[1:]) or "hello world! %s" % time.time() channel.basic_publish(exchange='', routing_key='task_queue', body=bytes(message, encoding='utf-8'), properties=pika.basicproperties( delivery_mode=2, # make message persistent 使消息持久 ) ) print(" [x] sent %r" % message) connection.close()
- 接收端:
import pika import time credentials = pika.plaincredentials('username', 'password') connection = pika.blockingconnection(pika.connectionparameters( '192.168.21.128', credentials=credentials)) channel = connection.channel() # 建立 rabbit 协议通道 def callback(ch, method, properties, body): print(" [x] received %r" % body) # b'hello world! 1557373639.5839057' time.sleep(20) print(" [x] done") print("method.delivery_tag", method.delivery_tag) # 1 ch.basic_ack(delivery_tag=method.delivery_tag) channel.basic_consume(on_message_callback=callback, queue='task_queue') print(' [*] waiting for messages. to exit press ctrl+c') channel.start_consuming()
另外一个接收端代码一致,在此省略,运行结果如下:
2.4 根据实际情况分发消息
事实上服务器之间接收、处理消息的能力是不一样的,受网络、配置等因素影响,因此公平分发消息就会导致以下问题出现:
- 配置高、网络好的服务器处理消息能力强、快
- 配置一般、网络不好的服务器有可能就会积压很多未处理的消息
为此我们可以在接收端设置 prefetch_count=1
,如果前面还有消息未处理,就告诉发送端不要给我发消息,直至处理完毕前一条消息为止:
channel.basic_qos(prefetch_count=1) # 如果前面有消息没处理完,就不要给我再发消息
3. 订阅(广播)
上面的例子基本上都是一对一发送和接收消息,如果想要将消息发送到所有队列(queue)中,那么就需要用到广播了,而实现广播的一个重要参数就是 exchange
—— 消息转发器。
exchange 在定义时是有类型的,只有符合条件的才能接收消息,大致可分为以下几类:
- fanout(全民广播):凡是绑定 exchange 的队列都可以接收到消息
- direct(组播):以组为单位接收消息,如:发送到某个组,那么这个组里的所有队列都能接收,
routingkey
为关键字/组名 - topic(根据特征收发消息):所有符合
routingkey
绑定的队列都可以接收消息
3.1 fanout 方式
所有绑定 exchange 的 queue 都能接收到消息。
应用场景:视频直播
- 发送端:
import pika import sys credentials = pika.plaincredentials('username', 'password') connection = pika.blockingconnection(pika.connectionparameters( '192.168.21.128', credentials=credentials)) channel = connection.channel() # 建立 rabbit 协议通道 # 指定 exchange 类型、名字 channel.exchange_declare(exchange='logs', exchange_type='fanout') message = ' '.join(sys.argv[1:]) or "info: hello world!" channel.basic_publish(exchange='logs', routing_key='', body=bytes(message, encoding='utf-8')) print(" [x] sent %r" % message) connection.close()
- 接收端:
import pika credentials = pika.plaincredentials('username', 'password') connection = pika.blockingconnection(pika.connectionparameters( '192.168.21.128', credentials=credentials)) channel = connection.channel() # 建立 rabbit 协议通道 channel.exchange_declare(exchange='logs', exchange_type='fanout') # 不指定queue名字, rabbit会随机分配一个名字,exclusive=true会在使用此queue的消费者断开后,自动将queue删除 # 最新源代码需要执行 queue,如果为 '',则 if empty string, the broker will create a unique queue name result = channel.queue_declare('', exclusive=true) queue_name = result.method.queue # result = <method(['channel_number=1', 'frame_type=1', "method=<queue.declareok(['consumer_count=0', 'message_count=0', 'queue=amq.gen-hrrq-pwat9u-32cciokcxa'])>"])> # queue_name = amq.gen-hrrq-pwat9u-32cciokcxa channel.queue_bind(exchange='logs', queue=queue_name) print(' [*] waiting for logs. to exit press ctrl+c') def callback(ch, method, properties, body): print(" [x] %r" % body) channel.basic_consume(on_message_callback=callback, queue=queue_name) channel.start_consuming()
打开两个终端,分别运行:
python3 fanout_send.py t1 python3 fanout_send.py t2
运行结果如下:
3.2 direct 方式
rabbitmq
还可以根据关键字发送接收消息,队列绑定关键字,发送端根据关键字发送到 exchange,exchange 再根据关键字判断发给哪个队列。
- 发送端:
import pika import sys credentials = pika.plaincredentials('username', 'password') connection = pika.blockingconnection(pika.connectionparameters( '192.168.21.128', credentials=credentials)) channel = connection.channel() # 建立 rabbit 协议通道 channel.exchange_declare(exchange='direct_logs', exchange_type='direct') # python3 direct_send.py info severity = sys.argv[1] if len(sys.argv) > 1 else 'info' # 严重程度,级别, info message = ' '.join(sys.argv[2:]) or 'hello world!' # hello world! channel.basic_publish(exchange='direct_logs', routing_key=severity, body=bytes(message, encoding='utf-8')) print(" [x] sent %r:%r" % (severity, message)) # [x] sent 'info' : 'hello world!' connection.close()
- 接收端:
import pika import sys credentials = pika.plaincredentials('username', 'password') connection = pika.blockingconnection(pika.connectionparameters( '192.168.21.128', credentials=credentials)) channel = connection.channel() # 建立 rabbit 协议通道 channel.exchange_declare(exchange='direct_logs', exchange_type='direct') result = channel.queue_declare('', exclusive=true) queue_name = result.method.queue # python3 direct_recv.py info warning error # python3 direct_recv.py info # python3 direct_recv.py error severities = sys.argv[1:] # ['direct_recv.py', 'info', 'warning', 'error']、['direct_recv.py', 'error']、['direct_recv.py', 'info'] if not severities: sys.stderr.write("usage: %s [info] [warning] [error]\n" % sys.argv[0]) sys.exit(1) # 循环绑定关键字 for severity in severities: channel.queue_bind(exchange='direct_logs', queue=queue_name, routing_key=severity) print(' [*] waiting for logs. to exit press ctrl+c') def callback(ch, method, properties, body): print(" [x] %r:%r" % (method.routing_key, body)) channel.basic_consume(on_message_callback=callback, queue=queue_name) channel.start_consuming()
接收端执行打开三个终端,分别执行:
python3 direct_recv.py info warning error python3 direct_recv.py info python3 direct_recv.py error
然后循环关键字,绑定队列(queue),发送端执行相应关键字,接收端这边就能根据关键字接收消息。
运行结果如下:
3.3 topic 方式
- 发送端:
import pika import sys credentials = pika.plaincredentials('username', 'password') connection = pika.blockingconnection(pika.connectionparameters( '192.168.21.128', credentials=credentials)) channel = connection.channel() # 建立 rabbit 协议通道 channel.exchange_declare(exchange='topic_logs', exchange_type='topic') routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info' message = ' '.join(sys.argv[2:]) or 'hello world!' channel.basic_publish(exchange='topic_logs', routing_key=routing_key, body=bytes(message, encoding='utf-8')) print(" [x] sent %r:%r" % (routing_key, message)) connection.close()
- 接收端:
import pika import sys credentials = pika.plaincredentials('username', 'password') connection = pika.blockingconnection(pika.connectionparameters( '192.168.21.128', credentials=credentials)) channel = connection.channel() # 建立 rabbit 协议通道 channel.exchange_declare(exchange='topic_logs', exchange_type='topic') result = channel.queue_declare('', exclusive=true) queue_name = result.method.queue binding_keys = sys.argv[1:] if not binding_keys: sys.stderr.write("usage: %s [binding_key]...\n" % sys.argv[0]) sys.exit(1) for binding_key in binding_keys: channel.queue_bind(exchange='topic_logs', queue=queue_name, routing_key=binding_key) print(' [*] waiting for logs. to exit press ctrl+c') def callback(ch, method, properties, body): print(" [x] %r:%r" % (method.routing_key, body)) channel.basic_consume(on_message_callback=callback, queue=queue_name) channel.start_consuming()
接收端开启四个终端,发送端开启一个:
# 接收端 python3 topic_recv.py *.django.* # 消息两端可以是任意,中间只要是 django 即可 python3 topic_recv.py # # 可以接收任意消息 python3 topic_recv.py mysql.* # 以 mysql 开头,结尾可以是任意 python3 topic_recv.py mysql.error.* # mysql.error 开头,结尾任意 # 发送端 python3 topic_send.py mysql.error.info python3 topic_send.py ss.django.123 python3 topic_send.py mysql.error err happend python3 topic_send.py python.error test
运行结果如下:
总结
-
#
号能匹配任意消息,相当于广播 -
*
号也可以匹配任意,但是必须和其他一起使用
4. rpc(remote procedure call)双向传输
上面收发消息都是单向的,即一个发一个接收,接收的不能够发送。而 rpc 是双向的,既能够发送也能接收。
应用场景:rpc 服务功能
- 发送端:
import pika import uuid class fibonaccirpcclient(object): def __init__(self): credentials = pika.plaincredentials('username', 'password') self.connection = pika.blockingconnection(pika.connectionparameters( '192.168.21.128', credentials=credentials)) channel = self.connection.channel() # 建立 rabbit 协议通道 self.channel = self.connection.channel() result = self.channel.queue_declare('', exclusive=true) self.callback_queue = result.method.queue self.channel.basic_consume(on_message_callback=self.on_response, queue=self.callback_queue, auto_ack=true) #准备接受命令结果 def on_response(self, ch, method, props, body): """"callback方法""" if self.corr_id == props.correlation_id: self.response = body def call(self, n): self.response = none self.corr_id = str(uuid.uuid4()) #唯一标识符 self.channel.basic_publish(exchange='', routing_key='rpc_queue', properties=pika.basicproperties( reply_to=self.callback_queue, correlation_id=self.corr_id, ), body=str(n)) count = 0 while self.response is none: self.connection.process_data_events() #检查队列里有没有新消息,但不会阻塞 count += 1 print("check...", count) return int(self.response) fibonacci_rpc = fibonaccirpcclient() print(" [x] requesting fib(30)") response = fibonacci_rpc.call(5) print(" [.] got %r" % response)
- 接收端:
import pika import time credentials = pika.plaincredentials('username', 'password') connection = pika.blockingconnection(pika.connectionparameters( '192.168.21.128', credentials=credentials)) channel = connection.channel() # 建立 rabbit 协议通道 channel.queue_declare(queue='rpc_queue') def fib(n): if n == 0: return 0 elif n == 1: return 1 else: return fib(n - 1) + fib(n - 2) def on_request(ch, method, props, body): n = int(body) print(" [.] fib(%s)" % n) response = fib(n) ch.basic_publish(exchange='', routing_key=props.reply_to, properties=pika.basicproperties(correlation_id=props.correlation_id), body=str(response)) ch.basic_ack(delivery_tag=method.delivery_tag) channel.basic_qos(prefetch_count=1) channel.basic_consume(on_message_callback=on_request, queue='rpc_queue') print(" [x] awaiting rpc requests") channel.start_consuming()
运行结果如下:
5. 参考:
- rabbitmq基本概念(二):windows下安装
- ubuntu上安装和使用rabbitmq
- rabbitmq在ubuntu 16.04下的安装与配置
- rabbitmq 入门
- python并发编程-rabbitmq消息队列
- windows下 安装 rabbitmq 及操作常用命令
6. 常用命令
#创建用户 rabbitmqctl add_user rabbitadmin 123456 rabbitmqctl set_user_tags rabbitadmin administrator # 给用户授权 rabbitmqctl set_permissions -p / rabbitadmin ".*" ".*" ".*" # 开启插件管理页面 rabbitmq-plugins enable rabbitmq_management rabbitmq-server start # 启动服务 rabbitmq-server stop # 关闭服务 rabbitmq-server restart # 重启服务 rabbitmq-server status # 查看服务状态 ps -ef|grep rabbitmq # 查看端口 rabbitmqctl list_queues # 查看队列消息 ./rabbitmqctl list_users # 查看用户列表命令 rabbitmqctl delete_user username # 删除用户命令 whereis rabbitmq #查看rabbitmq安装目录