Python消息队列(RabbitMQ)

RabbitMQ 即一个消息队列,主要是用来实现应用程序的异步和解耦,同时也能起到消息缓冲,消息分发的作用。可维护多个队列,可实现消息的一对一和广播等方式发送

RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编写,支持多种客户端,如:Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等,支持AJAX。用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗。

准备工作

  1. rabbitmq安装
  2. 配置端口
firewall-cmd --zone=public --add-port=15672/tcp --permanent
firewall-cmd --zone=public --add-port=5672/tcp --permanent
systemctl restart firewalld.service # 重启防火墙
 
  •  
  1. 创建用户:
rabbitmqctl add_user ruroot rproot
 
  •  
  1. 修改角色为管理员:
rabbitmqctl set_user_tags ruroot administrator
 
  •  
  1. 设置权限:
rabbitmqctl set_permissions -p / ruroot ".*" ".*" ".*"
 
  • 1

测试结果:
在这里插入图片描述

Python简单操控

安装

pip3 install pika
 
  • 1

发送数据:

如果生成多个的话,实现效果是轮询发送,一个一个循环发送数据,如同“皇帝轮流做…”

import pika

#建立连接
userx=pika.PlainCredentials("ruroot","rproot")
conn=pika.BlockingConnection(pika.ConnectionParameters("192.168.153.128",5672,'/',credentials=userx))

#开辟管道
channelx=conn.channel()

#声明队列,参数为队列名
channelx.queue_declare(queue="shenshupian")

#发送数据,发送一条,如果要发送多条则复制此段
channelx.basic_publish(exchange="",
                       routing_key="shenshupian",# 队列名
                       body="hello world" # 发送的数据
                       )
print("--------发送数据完成-----------")

#关闭连接
conn.close()
 
  •  

取出数据:

import pika

#建立连接
userx=pika.PlainCredentials("ruroot","rproot")
conn=pika.BlockingConnection(pika.ConnectionParameters("192.168.153.128",5672,'/',credentials=userx))

#开辟管道
channelx=conn.channel()

#声明队列,参数为队列名
channelx.queue_declare(queue="shenshupian")

#消息处理函数,执行完成才说明接收完成,此时才可以接收下一条,串行
def dongcallbackfun(v1,v2,v3,bodyx):
    print("得到的数据为:",bodyx)

#接收准备
channelx.basic_consume(queue="shenshupian", #队列名
                       on_message_callback=dongcallbackfun, #收到消息的回调函数
                       auto_ack=True #是否发送消息确认
                       )
print("-------- 开始接收数据 -----------")

#开始接收消息
channelx.start_consuming()
 

发送端是否设置数据保存时间:

默认服务器(rabbitmq-server)重启后消息队列和消息数据均会全部消失

消息队列的永久保存,开启后将仅仅实现服务器重启后消息队列依然在,但数据还是会丢失,如果要保存数据,请参考接下来

#声明队列,参数为队列名
#实现队列永久保存,durable=True
channelx.queue_declare(queue="test1",durable=True)
 

数据的永久保存(一直等待被取,即使服务器重启),将要永久保存的发送数据添加属性properties

#发送数据
channelx.basic_publish(exchange="",
                       routing_key="test1",# 队列名
                       body="dongxiaodongtodata333335", # 发送的数据
                       properties=pika.BasicProperties(
                           delivery_mode=2, #实现消息永久保存
                       )
                       )
 

发送端实现能者多劳

在发送端发送数据前,添加下面一句,此句添加一次即可,可以实现自动判断多接收端的处理速度,实现接收端处理快则多派发任务,处理慢则少派发任务

channelx.basic_qos(prefetch_count=1)
 

接收端是否接收确认:

接收端开启消息确认(值为False),接收端则会在接收回调函数结束时手动发送确认消息到数据发送者,如果接收端在回调函数处理未完成时就挂掉了,那么发送端将会立即把当前数据转交到下一个接收端进行数据处理

#消息处理函数,执行完成才说明接收完成,此时才可以接收下一条,串行
def dongcallbackfun(v1,v2,v3,bodyx):
    print("得到的数据为:",bodyx)

#接收准备
channelx.basic_consume(queue="shenshupian", #队列名
                       on_message_callback=dongcallbackfun, #收到消息的回调函数
                       auto_ack=True #是否发送消息确认
                       )
 

非阻塞版数据接收:

启用会立即返回结果,如果有数据则进入回调函数,无数据则进行下一条,可以配合while使用

conn.process_data_events() #使用连接对象进行数据接收判断
print("无数据")
 

实现消息的订阅和发布:

发布:

import pika

#建立连接
userx=pika.PlainCredentials("ruroot","rproot")
conn=pika.BlockingConnection(pika.ConnectionParameters("192.168.153.128",5672,'/',credentials=userx))

#开辟管道
channelx=conn.channel()

#声明发布和订阅通道,如果可以确认通道存在则可以去掉该句
channelx.exchange_declare(exchange="shupian",exchange_type="fanout")

#发送数据
channelx.basic_publish(exchange="shupian",#确定发布主题为:shupian
                       routing_key="",
                       body="shupian 666", # 发送的数据
                       )

print("--------发送数据完成-----------")

#关闭连接
conn.close()


 # 开辟管道
        channelx = conn.channel()
        # 声明队列,参数为队列名
        arguments = {
            'x-message-ttl': 10000
        }
        channelx.queue_declare(queue="robotHeartbeatQueue", durable=True, arguments=arguments)
        channelx.exchange_declare(exchange="robotHeartbeatExchange",
                                  durable=True,
                                  auto_delete=False)
        channelx.queue_bind(queue="robotHeartbeatQueue", exchange="robotHeartbeatExchange", routing_key="robotHeartbeatRoutingKey")

        # 发送数据,发送一条,如果要发送多条则复制此段
        channelx.basic_publish(exchange="robotHeartbeatExchange",
                               routing_key="robotHeartbeatRoutingKey",
                               body=msg  # 发送的数据
                               )

 print('关闭mq连接')
    conn.close()

 

 
 

订阅:

import pika
#建立连接
userx=pika.PlainCredentials("ruroot","rproot")
conn=pika.BlockingConnection(pika.ConnectionParameters("192.168.153.128",5672,'/',credentials=userx))

#开辟管道
channelx=conn.channel()

#声明发布和订阅通道,如果可以确认通道存在则可以去掉该句
channelx.exchange_declare(exchange="shupian",exchange_type="fanout")

#声明队列,生成一个随机的且不存在的队列,该队列会在连接断开后自动销毁
resqueue=channelx.queue_declare('', exclusive=True)
#得到随机生成的队列名
queuenamex=resqueue.method.queue

#将队列和发布数据绑定,确定订阅主题为:shupian
channelx.queue_bind(exchange="shupian",queue=queuenamex)

#消息处理函数,执行完成才说明接收完成,此时才可以接收下一条,串行
def dongcallbackfun(channlx,methodx,v3,bodyx):
    print("得到的数据为:",bodyx)

#接收准备
channelx.basic_consume(on_message_callback=dongcallbackfun, #收到消息的回调函数
                       queue=queuenamex, #队列名
                       auto_ack=True
                       )

print("-------- 开始接收数据 -----------")

#开始接收消息
channelx.start_consuming()
 
  •  

通过管道实现进一步的消息订阅和发布:

发布:

import pika

#建立连接
userx=pika.PlainCredentials("ruroot","rproot")
conn=pika.BlockingConnection(pika.ConnectionParameters("192.168.153.128",5672,'/',credentials=userx))

#开辟管道
channelx=conn.channel()

#声明发布和订阅通道,如果可以确认通道存在则可以去掉该句
channelx.exchange_declare(exchange="shupian2",exchange_type="direct")

#发送数据
channelx.basic_publish(exchange="shupian2",#确定发布主题为:shupian2
                       routing_key="test1", #确定发布的队列(发布的主题):test1
                       body="nihao 666", # 确定发送的数据
                       )

#发送数据
channelx.basic_publish(exchange="shupian2",#确定发布主题为:shupian2
                       routing_key="test2", #确定发布的队列(发布的主题):test2
                       body="nihao 555", # 确定发送的数据
                       )

print("--------发送数据完成-----------")

#关闭连接
conn.close()
 
  •  

订阅:

import pika
#建立连接
userx=pika.PlainCredentials("ruroot","rproot")
conn=pika.BlockingConnection(pika.ConnectionParameters("192.168.153.128",5672,'/',credentials=userx))

#开辟管道
channelx=conn.channel()

#声明发布和订阅通道,如果可以确认通道存在则可以去掉该句
channelx.exchange_declare(exchange="shupian2",exchange_type="direct")

#声明队列,生成一个随机的且不存在的队列,该队列会在连接断开后自动销毁
resqueue=channelx.queue_declare('', exclusive=True)
#得到随机生成的队列名
queuenamex=resqueue.method.queue

#将队列和发布数据绑定,确定订阅主题为:test1 和 test2
channelx.queue_bind(exchange="shupian2",queue=queuenamex,routing_key="test1")
channelx.queue_bind(exchange="shupian2",queue=queuenamex,routing_key="test2")

#消息处理函数,执行完成才说明接收完成,此时才可以接收下一条,串行
def dongcallbackfun(channlx,methodx,v3,bodyx):
    print("队列名(订阅的主题名)为:%r  得到的数据为:%r  "%(methodx.routing_key,bodyx))

#接收准备
channelx.basic_consume(on_message_callback=dongcallbackfun, #收到消息的回调函数
                       queue=queuenamex, #队列名
                       auto_ack=True
                       )

print("-------- 开始接收数据 -----------")

#开始接收消息
channelx.start_consuming()











 
 

 

 
 
版权声明:本文为xiondun原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://www.cnblogs.com/xiondun/p/16387909.html