526互联

Python_消息队列

发布时间 2023-11-03 18:07:27作者: 辰令

rabbitpy: RabbitMQ Simplified

###[scheme]://[username]:[password]@[host]:[port]/[virtual_host]
  url = 'amqp://guest:guest@localhost:5672/%2F'
  这个%2f是字符斜杠/ uni-encode之后的结果，而/就是默认的hostname，它在HTTP传输的时候必须要转义。

 Port 15672 is for the management panel and is open for connections from a different machine.
    通过 http://serverip:15672 访问 RabbitMQ 的 Web 管理界面，默认用户名密码都是 guest
 Port 5672 is for applications and can only be accessed from the internal network.

部署：

1.安装erlang 语言
   由于rabbitMQ是erlang语言实现的,所以我们需要安装erlang

   sudo apt-get install erlang-nox
   
   查看版本： erl -version

2.安装rabbitMQ
  安装最新版rabbitMQ
   sudo apt-get install rabbitmq-server


3.查看rabbitMQ状态,active(running)表示在线
    sudo systemctl status rabbitmq-server

服务端命令

rabbitmqctl 用来管理RabbitMQ中间件的命令行工具.它通过连接中间件节点来执行所有操作	
     rabbitmqctl 命令，查看队列信息，所包含的信息包括 name，arguments， messages，memory
用户管理
  注意rabbitmqctl 管理RabbitMQ 内部用户数据库. 任何来自其它认证后端的用户对于rabbitmqctl来说是不可见的.	
	 
 sudo  rabbitmqctl list_queues
 rabbitmqctl cluster_status 
  获取 RabbitMQ Server 状态 命令为 rabbitmqctl status  
 
 rabbitmq-plugins list 
   rabbitmq-plugins 是管理RabbitMQ broker插件的命令行	 
 
 　后台管理界面是以插件的形式需要用户手动安装，不像activemq启动后后台管理服务也自动启动可直接访问，
   rabbitmq 后台管理界面访问http://localhost:15672

开启后台管理页面

    要在浏览器中访问 RabbitMQ，需要使用 RabbitMQ 的 Web 界面插件。可以按照以下步骤安装和配置：
    1. 安装 RabbitMQ Web 插件：
    sudo rabbitmq-plugins enable rabbitmq_management
    
    2. 重启 RabbitMQ 服务：
    sudo systemctl restart rabbitmq-server
    
    3. 确认 Web 界面已经启用：
    sudo rabbitmq-plugins list
	在生产环境中，应该限制对 RabbitMQ 管理插件的访问，并使用HTTPS来保护您的数据

生产者

import rabbitpy 

url = 'amqp://guest:guest@192.168.89.38:30595/%2F'

connection = rabbitpy.Connection(url)
channel = connection.channel()
###声明一个exchange，指定channel名称
exchange = rabbitpy.Exchange(channel, 'chapter2-example')
####使用declare方法来发送Exchange.Declare命令
exchange.declare()
####建立队列，指定队列名称
queue = rabbitpy.Queue(channel, 'example')
###发送Queue.Declare命令，会返回(此队列中的消息数，此队列的消费者数)
queue.declare() 
###绑定队列到exchange# 会发送Queue.Bind命令，传入exchange和路由key
queue.bind(exchange, 'example-routing-key')

####下面就可以发送消息了
for message_number in range(10):
    message = rabbitpy.Message(channel, # 指定channel                               
	                        f'Test message {message_number}', # 消息体                             
	                        {'content_type':'text/plain'}, # 消息属性(字典)                             
	                        opinionated=True)

创建Basic.Public方法帧(frame)，消息头帧,和一个消息体帧，然后传输到RabbitMQ

message.publish(exchange, 'example-routing-key')

代码

import rabbitpy,datetime
###[scheme]://[username]:[password]@[host]:[port]/[virtual_host]
url = 'amqp://guest:guest@localhost:5672/%2F'
with rabbitpy.Connection(url) as connection:
    with connection.channel() as channel:
        exchange = rabbitpy.Exchange(channel, 'chapter4-example')
        exchange.declare()
        channel.enable_publisher_confirms() # 开启发布确认
        body = 'This is an important message'
        message = rabbitpy.Message(channel, body,                          
                                   {'content_type': 'text/plain','message_type': 'very important'}
								   )
        if message.publish('chapter4-example','important.message'):
            print('The message was confirmed')
 ### DirectExchange  FanoutExchange  HeadersExchange  TopicExchange
 ### bind  declare delete unbind

消费者

import rabbitpy
with rabbitpy.Connection() as conn:
    with conn.channel() as channel:
        amqp = rabbitpy.AMQP(channel)
        for message in amqp.basic_consume('queue-name'):
            print(message)
#### basic_get  basic_cancel  basic_nack  basic_publish

with conn.channel() as channel:
    queue = rabbitpy.Queue(channel, 'example')
    for message in queue.consume():
        print 'Message: %r' % message
        message.ack()
		
		
import rabbitpy
with rabbitpy.Connection('amqp://guest:guest@localhost:5672/%2f') as conn:
    with conn.channel() as channel:
        queue = rabbitpy.Queue(channel, 'example')
        while len(queue) > 0:
            message = queue.get()
            message.pprint(True)
            message.ack()
            print('There are {} more messages in the queue'.format(len(queue)))

rabbitpy.Queue

#queue = rabbitpy.Queue(channel, 'my-queue')

参考

 https://blog.csdn.net/yjw123456/article/details/118756425	
  https://rabbitpy.readthedocs.io/en/latest/api/channel.html