Python3 通过 kombu 连接 RabbitMQ 的基本用法

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Python3 通过 kombu 连接 RabbitMQ 的基本用法相关的知识,希望对你有一定的参考价值。

【RabbitMQ 服务器】

# 在 vhosttest 里面有 exchangetest 和 queuetest 通过 rkeytest 绑定
Broker: 192.168.0.xx
virtual host: vhosttest
Exchange: exchangetest 
Queue: queuetest 
Routing key: rkeytest


【Python 环境】

OS: Windows 10
Python: 3.6.3 x64
kombu: 4.1.0


【查看队列状态】

# 通过浏览器查看队列状态
http://192.168.0.xx:15672/api/queues/vhosttest/queuetest  

# 通过命令行查看队列状态curl -u user:password 
http://192.168.0.xx:15672/api/queues/vhosttest/queuetest  |  jq 

# 通过命令行查看队列长度(messages = messages_ready + messages_unacknowledged)
curl -s -u user:password http://192.168.0.xx:15672/api/queues/vhosttest/queuetest  | \    
    jq '.messages'


【send.py】

#encoding: utf-8
#author: walker
#date: 2018-03-09
#summary: 发送方/生产者

import os, sys, time
from kombu import Connection

def Main():
	with Connection('amqp://test:[email protected]:5672/vhosttest') as conn:
		with conn.channel() as channel:
			#producer = Producer(channel)
			producer = channel.Producer()
			
			while True:
				message = time.strftime('%H:%M:%S', time.localtime())
				producer.publish(
						body=message,
						retry=True,
						exchange='exchangetest',
						routing_key='rkeytest'
					)
				print('send message: %s' %  message)     
					
				while True:
				        # 检查队列,以重新得到消息计数
					queue = channel.queue_declare(queue='queuetest', passive=True)     
					messageCount = queue.message_count
					print('messageCount: %d' % messageCount)
					if messageCount < 100:
						break
					time.sleep(1)
					
					
if __name__ == '__main__':
	Main()


【recv.py】

#encoding: utf-8
#author: walker
#date:  2018-03-09
#summary: 接收方/消费者

import os, sys, time
from kombu import Connection, Queue
from kombu.mixins import ConsumerMixin

class C(ConsumerMixin):

	def __init__(self, connection, queueNmae):
		self.connection = connection
		self.queues = [Queue(queueNmae, durable=False)]

	def get_consumers(self, Consumer, channel):
		return [
			Consumer(self.queues, callbacks=[self.on_message]),
		]

	# 接收处理消息的回调函数
	def on_message(self, body, message):
		print("Received %s" % body)
		message.ack()
		
def Main():
	with Connection('amqp://test:[email protected]:5672/vhosttest') as conn:
		C(conn, 'queuetest').run()					
					
if __name__ == '__main__':
	Main()	


【相关阅读】


*** walker ***



以上是关于Python3 通过 kombu 连接 RabbitMQ 的基本用法的主要内容,如果未能解决你的问题,请参考以下文章

Django+Celery+redis kombu.exceptions.EncodeError:Object of type is not JSON serializable报错

[源码分析] 消息队列 Kombu 之 启动过程

[源码解析] 消息队列 Kombu 之 基本架构

引发 ConnectionError(self._error_message(e)) kombu.exceptions.OperationalError:连接到 localhost:6379 的错误

Kombu/Celery 消息传递

用kombu创建rabbitmq vhost