当前位置： 首页 > news >正文

不动产网站建设,天津网站建设交易,wordpress最新版本发布文章,网站布局分类文章目录 安装rabbitmq镜像访问容器内部15672端口映射到外面的端口地址RabbitMQ六种模式Hello world模式Work queues模式Publish/Subscribe模式交换机fanout类型 Routing模式Topics模式RPC模式 rabbitmq#xff1a;0-1的学习 学习文档#xff1a;https://www.cnblogs.com… 文章目录 安装rabbitmq镜像访问容器内部15672端口映射到外面的端口地址RabbitMQ六种模式Hello world模式Work queues模式Publish/Subscribe模式交换机fanout类型 Routing模式Topics模式RPC模式 rabbitmq0-1的学习 学习文档https://www.cnblogs.com/guyuyun/p/14970592.html 安装rabbitmq镜像 cd /home/rabbitmqdocker run -d
-v ./data:/var/lib/rabbitmq
-p 5672:5672 -p 5673:15672 –name rabbitmq –restartalways
–hostname myRabbit rabbitmq:3.9.12-management访问容器内部15672端口映射到外面的端口地址 这里是127.0.0.1:5673 默认账号密码账号 guest 密码 guest RabbitMQ六种模式 Hello world模式 Hello world模式是最简单的一种模式一个producer发送message另一个consumer接收message。 producer示例 send.py producer端发送message会涉及最简单的5个步骤具体见代码注释。 import pika# 1. 创建一个到RabbitMQ server的连接如果连接的不是本机

则在pika.ConnectionParameters中传入具体的ip和port即可默认端口开放为5672可以不传

connection pika.BlockingConnection(pika.ConnectionParameters(localhost, port))

2. 创建一个channel

channel connection.channel()

3. 创建队列queue_declare可以使用任意次数

如果指定的queue不存在则会创建一个queue如果已经存在

则不会做其他动作官方推荐每次使用时都可以加上这句

channel.queue_declare(queuehello)

4. 发布消息

channel.basic_publish(exchange, # RabbitMQ中所有的消息都要先通过交换机空字符串表示使用默认的交换机routing_keyhello, # 指定消息要发送到哪个queuebodyHello world!) # 消息的内容

5. 关闭连接

connection.close()consumer示例 receive.py consumer端接收message会涉及最简单的6个步骤具体见代码注释。 import pikadef main():# 1. 创建一个到RabbitMQ server的连接如果连接的不是本机# 则在pika.ConnectionParameters中传入具体的ip和port即可connection pika.BlockingConnection(pika.ConnectionParameters(localhost))# 2. 创建一个channelchannel connection.channel()# 3. 创建队列queue_declare可以使用任意次数# 如果指定的queue不存在则会创建一个queue如果已经存在# 则不会做其他动作官方推荐每次使用时都可以加上这句channel.queue_declare(queuehello)# 4. 定义消息处理程序def callback(ch, method, properties, body):print([x] Received %r % body)# 5. 接收来自指定queue的消息channel.basic_consume(queuehello, # 接收指定queue的消息on_message_callbackcallback, # 接收到消息后的处理程序auto_ackTrue) # 指定为True表示消息接收到后自动给消息发送方回复确认已收到消息print([*] Waiting for message.)# 6. 开始循环等待一直处于等待接收消息的状态channel.start_consuming()if name main:main()Work queues模式 Work queues模式即工作队列模式也称为Task queues模式任务队列模式这个模式的特点在于同一个queue可以允许多个consumer从中获取massageRabbitMQ默认会从queue中依次循环的给不同的consumer发送message。与Hello world模式相比工作队列模式在示例代码中有些许不同 hello world模式中指定了 auto_ackTrue 表示consumer接收到message之后自动发送确认标识告诉RabbitMQ可以从队列中移除该条message了。工作队列模式下使用了默认值即需要手动确认 ch.basic_ack(delivery_tagmethod.delivery_tag) 。hello world模式中只有一个consumer去处理queue中的message工作队列模式中可以有多个consumer去处理queue中的message。工作队列模式中可以使message持久化保证RabbitMQ服务挂掉之后message依然不被丢失。工作队列模式中可以手动标记message已接收并处理完成这一步在编程时千万别忘了否则RabbitMQ会认为该条message没有被处理会一直保留在队列中并适时发送到别的consumer中。 producer示例 new_task.py 注意如果声明queue时参数不一样则建议换一个名称因为RabbitMQ中不允许同名但实际上是不同的两个queue存在比如指定了 durableTrue 参数。 import pikaconnection pika.BlockingConnection(pika.ConnectionParameters(hostlocalhost)) channel connection.channel()# 声明durableTrue可以保证RabbitMQ服务挂掉之后队列中的消息也不丢失原理是因为

RabbitMQ会将queue中的消息保存到磁盘中

channel.queue_declare(queuetask_queue)message Hello World! 555 channel.basic_publish(exchange,routing_keytask_queue,bodymessage,# delivery_mode2可以指定此条消息持久化防止RabbitMQ服务挂掉之后消息丢失# 但是此属性设置并不能百分百保证消息真的被持久化因为RabbitMQ挂掉的时候# 它可能还保存在缓存中没来得及同步到磁盘中# propertiespika.BasicProperties(delivery_mode2) ) print( [x] Sent %r % message) connection.close()consumer示例 worker.py RabbitMQ会将queue中的消息依次发送给不同的consumer所以这里的示例可以用同样的代码多开几个客户端进行测试。 import pika import timeconnection pika.BlockingConnection(pika.ConnectionParameters(hostlocalhost)) channel connection.channel()# 声明durableTrue可以保证RabbitMQ服务挂掉之后队列中的消息也不丢失原理是因为

RabbitMQ会将queue中的消息保存到磁盘中

channel.queue_declare(queuetask_queue) print( [*] Waiting for messages.)def callback(ch, method, properties, body):print( [x] Received %r % body.decode())# 此处以消息中的“.”的数量作为sleep的值是为了模拟不同消息处理的耗时time.sleep(body.count(b.))print( [x] Done)# 手动标记消息已接收并处理完毕RabbitMQ可以从queue中移除该条消息ch.basic_ack(delivery_tagmethod.delivery_tag)# prefetch_count表示接收的消息数量当我接收的消息没有处理完用basic_ack

标记消息已处理完毕之前不会再接收新的消息了

channel.basic_qos(prefetch_count1) channel.basic_consume(queuetask_queue, on_message_callbackcallback)channel.start_consuming()Publish/Subscribe模式 相对于工作/任务模式中的一个message只能发送给一个consumer使用发布订阅模式会将一个message同时发送给多个consumer使用其实就是producer将message广播给所有的consumer。 交换机 这个模式中会引入交换机exchange的概念其实在RabbitMQ中所有的producer都不会直接把message发送到queue中甚至producer都不知道message在发出后有没有发送到queue中事实上producer只能将message发送给exchange由exchange来决定发送到哪个queue中。 exchange的一端用来从producer中接收message另一端用来发送message到queueexchange的类型规定了怎么处理接收到的message发布订阅模式使用到的exchange类型为 fanout 这种exchange类型非常简单就是将接收到的message广播给已知的即绑定到此exchange的所有consumer。 当然如果不想使用特定的exchange可以使用 exchange 表示使用默认的exchange默认的exchange会将消息发送到 routing_key 指定的queue可以参考工作任务队列模式和Hello world模式。 fanout类型 在使用fanout类型的exchange时并不是只有一个queue然后将queue中的message每个consumer都发一份而是会为每个已知绑定的consumer创建一个queue然后广播message到对应queue中fanout类型的exchange会将从生产者接收到的message广播到所有的绑定到自己的queue中这个queue通常是由consumer端指定的专属于consumer自己的、由RabbitMQ随机命名的queue由此consumer广播message后每个consumer都能收到同样的一条message了。 consumer端需要为自己生成一个专属于自己的由RabbitMQ随机命名的queue然后绑定到fanout类型的exchange上由此exchange才知道将message广播给哪些已经绑定到自己的queue。 示例 emit_log.py 用于生成一条日志信息然后广播给所有consumer。 import pikaconnection pika.BlockingConnection(pika.ConnectionParameters(hostlocalhost)) channel connection.channel()# 创建一个指定名称的交换机并指定类型为fanout用于将接收到的消息广播到所有queue中 channel.exchange_declare(exchangelogs, exchange_typefanout)message info: Hello World!

将消息发送给指定的交换机在fanout类型中routing_key表示不用发送到指定queue中

而是将发送到绑定到此交换机的所有queue

channel.basic_publish(exchangelogs, routing_key, bodymessage) print( [x] Sent %r % message) connection.close()示例 receive_logs.py 这个程序可以多运行几个表示有多个consumer需要使用producer发送的消息。 import pikaconnection pika.BlockingConnection(pika.ConnectionParameters(hostlocalhost)) channel connection.channel()# 指定交换机 channel.exchange_declare(exchangelogs, exchange_typefanout)# 使用RabbitMQ给自己生成一个专有的queue result channel.queue_declare(queue, exclusiveTrue) queue_name result.method.queue# 将queue绑定到指定交换机 channel.queue_bind(exchangelogs, queuequeue_name)print( [*] Waiting for logs.)def callback(ch, method, properties, body):print( [x] %r % body)channel.basic_consume(queuequeue_name, on_message_callbackcallback, auto_ackTrue)channel.start_consuming()Routing模式 路由模式中exchange类型为direct与发布订阅模式相似但是不同之处在于发布订阅模式将message不加区分广播给所有的绑定queue但是路由模式中允许queue在绑定exchange时同时指定 routing_key exchange就只会发送message到与 routing_key 匹配的queue中其他的所有message都将被丢弃。当然也允许多个queue指定相同的 routing_key 此时效果就相当于fanout类型的发布订阅模式了。 producer端从代码上看路由模式和订阅模式非常相似唯一不同的是exchange类型为direct且发送message时多了一个routing_key参数exchange会根据routing_key将message发送到对应的queue中。 示例 emit_log_direct.py 发送不同级别的日志消息到queue中不同的consumer根据自己指定的routing_key接收message。 import pikaconnection pika.BlockingConnection(pika.ConnectionParameters(hostlocalhost)) channel connection.channel()# 指定交换机名称和类型 channel.exchange_declare(exchangedirect_logs, exchange_typedirect)# severity info

severity warning

severity error message Hello World!# 与fanout类型的发布订阅模式相比只是多了一个routing_key参数

交换机会根据routing_key将消息发送到对应的queue中

channel.basic_publish(exchangedirect_logs, routing_keyseverity, bodymessage) print( [x] Sent %r:%r % (severity, message)) connection.close()consumer端在路由模式中不同的queue可以指定相同的routing_key同一个queue也可以指定多个routing_key从exchange角度看它知道所有绑定到自己的queue也知道每个queue指定的routing_key发送消息时只需要根据queue的routing_key进行发送即可。 示例 receive_logs_direct.py 这个程序可以多运行几个每个程序指定接收不同日志级别的消息。 import pikaconnection pika.BlockingConnection(pika.ConnectionParameters(hostlocalhost)) channel connection.channel()# 指定交换机名称和类型 channel.exchange_declare(exchangedirect_logs, exchange_typedirect)# 使用RabbitMQ给自己生成一个专属于自己的queue result channel.queue_declare(queue, exclusiveTrue) queue_name result.method.queue# 绑定queue到交换机并指定自己只接受哪些routing_key

可以都接收也可以只接收一种

for severity in [error, warning, info]:

for severity in [error]:channel.queue_bind(exchangedirect_logs, queuequeue_name, routing_keyseverity)print( [*] Waiting for logs. To exit press CTRLC)def callback(ch, method, properties, body):print( [x] %r:%r % (method.routing_key, body))channel.basic_consume(queuequeue_name, on_message_callbackcallback, auto_ackTrue)channel.start_consuming()Topics模式 主题模式的exchange类型为topic相较于路由模式主题模式更加灵活区别就在于它的routing_key可以带通配符 * 匹配一个单词和 # 匹配0个或多个单词每个单词以点号分隔但注意routing_key的总大小不能超过255个字节。 如果一个message同时匹配了多个queue中的routing_key那这几个queue都会收到这个message如果一个message同时匹配了一个queue中的多个routing_key那这个queue也只会接收一次这条message如果一个message没有匹配上任何routing_key那么这个message将被丢弃。 如果routing_key定义为 # 就只有这一个通配符那么这个queue将接收所有message就像exchange类型为fanout的发布订阅模式一样如果routing_key两个通配符都没有使用那么这个queue将会接收固定routing_key的message就像exchange类型为direct的路由模式一样。 producer端从代码上讲producer的代码与路由模式没什么区别只不过在routing_key的传值上需要注意与想要发送到的queue进行匹配。 示例 emit_log_topic.py 还是发送日志消息的示例不过消息类型不再只有级别这一种类型还添加了发送者的信息级别与发送者之间以点号分隔。 import pikaconnection pika.BlockingConnection(pika.ConnectionParameters(hostlocalhost)) channel connection.channel()# 指定交换机名称和类型 channel.exchange_declare(exchangetopic_logs, exchange_typetopic)# 以点号分隔每个单词 routing_key anonymous.error message Hello World! channel.basic_publish(exchangetopic_logs, routing_keyrouting_key, bodymessage) print( [x] Sent %r:%r % (routing_key, message)) connection.close()consumer端consumer根据需要使用星号 * 和井号 # 两个通配符对routing_key进行特定主题的匹配其余部分与路由模式则是一致的。 示例 receive_logs_topic.py : 这个程序可以多运行几个每个程序使用通配符指定不同的主题。 import pikaconnection pika.BlockingConnection(pika.ConnectionParameters(hostlocalhost)) channel connection.channel()# 指定交换机名称和类型 channel.exchange_declare(exchangetopic_logs, exchange_typetopic)# 使用RabbitMQ给自己生成一个专属于自己的queue result channel.queue_declare(queue, exclusiveTrue) queue_name result.method.queue# 可以绑定多个routing_keyrouting_key以点号分隔每个单词

*可匹配一个单词#可以匹配0个或多个单词

for binding_key in [anonymous.]:channel.queue_bind(exchangetopic_logs, queuequeue_name, routing_keybinding_key)print( [] Waiting for logs. To exit press CTRLC)def callback(ch, method, properties, body):print( [x] %r:%r % (method.routing_key, body))channel.basic_consume(queuequeue_name, on_message_callbackcallback, auto_ackTrue)channel.start_consuming()RPC模式 RPC远程调用Remote Procedure Call模式其实就是使用消息队列处理请求的一种方式通常请求接收到后会立即执行且多个请求是并行执行的如果一次性来了太多请求达到了服务端处理请求的瓶颈就会影响性能但是如果使用消息队列的方式最大的一点好处是可以不用立即处理请求而是将请求放入消息队列服务端只需要根据自己的状态从消息队列中获取并处理请求即可。 producer端RPC模式的客户端producer需要使用到两个queue一个用于发送request消息此queue通常在服务端声明和创建一个用于接收response消息。另外需要特别注意的一点是需要为每个request消息指定一个uuidcorrelation_id属性类似请求id用于识别返回的response消息是否属于对应的request。 示例 rpc_client.py import pika import uuidclass FibonacciRpcClient(object):def init(self):self.connection pika.BlockingConnection(pika.ConnectionParameters(hostlocalhost))self.channel self.connection.channel()# 创建一个此客户端专用的queue用于接收服务端发过来的消息result self.channel.queue_declare(queue, exclusiveTrue)self.callback_queue result.method.queueself.channel.basic_consume(queueself.callback_queue,on_message_callbackself.on_response,auto_ackTrue)def on_response(self, ch, method, props, body):# 判断接收到的response是否属于对应requestif self.corr_id props.correlation_id:self.response bodydef call(self, n):self.response Noneself.corr_id str(uuid.uuid4()) # 为该消息指定uuid类似于请求idself.channel.basic_publish(exchange,routing_keyrpc_queue, # 将消息发送到该queuepropertiespika.BasicProperties(reply_toself.callback_queue, # 从该queue中取消息correlation_idself.corr_id, # 为此次消息指定uuid),bodystr(n))while self.response is None:self.connection.process_data_events()return int(self.response)fibonacci_rpc FibonacciRpcClient()print( [x] Requesting fib(30)) response fibonacci_rpc.call(30) print( [.] Got %r % response)consumer端服务端也需要使用到两个queue一个接收request消息通常由服务端创建一个发送response消息通常由客户端创建需要特别注意发送response消息时需要将对应request的uuidcorrelation_id属性赋上。 示例 rpc_server.py import pikaconnection pika.BlockingConnection(pika.ConnectionParameters(hostlocalhost))channel connection.channel()# 指定接收消息的queue channel.queue_declare(queuerpc_queue)def fib(n):if n 0:return 0elif n 1:return 1else: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_keyprops.reply_to, # response发送到该queuepropertiespika.BasicProperties(correlation_idprops.correlation_id), # 使用correlation_id让此response与请求消息对应起来bodystr(response))ch.basic_ack(delivery_tagmethod.delivery_tag)channel.basic_qos(prefetch_count1)

从rpc_queue中取消息然后使用on_request进行处理

channel.basic_consume(queuerpc_queue, on_message_callbackon_request)print( [x] Awaiting RPC requests) channel.start_consuming()