使用 nameko 的时候,设置 max_workers 可以实现不同并发的消费
我自己用 eventlet+kombu 写了一个消费者,却不能实现并发消费,只能一个一个排队消费
代码如下:
import random
import eventlet
eventlet.monkey_patch() # noqa (code before rest of imports)
from kombu.mixins import ConsumerMixin
from kombu.messaging import Consumer
from kombu import Connection, Exchange, Queue
from loguru import logger
import time
class MyConsumer(ConsumerMixin):
def __init__(self, connection):
self.connection = connection
def get_consumers(self, Consumer, channel):
print('创建消费者 start')
queue_name = 'evt-ye.events-take--dna_create_service.auth'
exchange_name = 'ye.events'
routing_key = 'take'
exchange = Exchange(exchange_name, type='topic')
queue = Queue(
queue_name, exchange=exchange,
routing_key=routing_key,
queue_arguments={'x-max-priority': 10}
)
# 创建一个消费者,并设置预取消息数量为10
consumer = Consumer(
queues=[queue], callbacks=[self.on_message],
prefetch_count=10
)
print('创建消费者 down')
return [consumer]
def on_message(self, body, message):
logger.debug(f"Received message: {body}")
logger.debug(f'开始消费')
time.sleep(3)
logger.debug(f'结束消费')
message.ack() # 发送ACK确认消息接收
with Connection('amqp://pon:pon@192.168.38.191:5672//') as conn:
consumer = MyConsumer(conn)
consumer.run()
就很烦,怎么办,我就去翻了半天 nameko 的源码代码,发现问题在这里:
site-packages/nameko/messaging.py
注册回调函数
def handle_message(self, provider, body, message):
ident = u"{}.handle_message[{}]".format(
type(provider).__name__, message.delivery_info['routing_key']
)
self.container.spawn_managed_thread(
partial(provider.handle_message, body, message), identifier=ident
)
def get_consumers(self, consumer_cls, channel):
""" Kombu callback to set up consumers.
Called after any (re)connection to the broker.
"""
_log.debug('setting up consumers %s', self)
for provider in self._providers:
callbacks = [partial(self.handle_message, provider)]
consumer = consumer_cls(
queues=[provider.queue],
callbacks=callbacks,
accept=self.accept
)
consumer.qos(prefetch_count=self.prefetch_count)
self._consumers[provider] = consumer
return self._consumers.values()
然后 kombu 在收到消息之后,会调用之前注册的回调函数site-packages/kombu/messaging.py
def receive(self, body, message):
"""Method called when a message is received.
This dispatches to the registered :attr:`callbacks`.
Arguments:
body (Any): The decoded message body.
message (~kombu.Message): The message instance.
Raises:
NotImplementedError: If no consumer callbacks have been
registered.
"""
callbacks = self.callbacks
if not callbacks:
raise NotImplementedError('Consumer does not have any callbacks')
[callback(body, message) for callback in callbacks]
所以,道理很简单,我按照 nameko 的思路,改造了一下我的代码
最后完整的代码如下:
import random
import eventlet
eventlet.monkey_patch() # noqa (code before rest of imports)
from kombu.mixins import ConsumerMixin
from kombu.messaging import Consumer
from kombu import Connection, Exchange, Queue
from loguru import logger
import time
class MyConsumer(ConsumerMixin):
def __init__(self, connection):
self.connection = connection
def get_consumers(self, Consumer, channel):
print('创建消费者 start')
queue_name = 'evt-ye.events-take--dna_create_service.auth'
exchange_name = 'ye.events'
routing_key = 'take'
exchange = Exchange(exchange_name, type='topic')
queue = Queue(
queue_name, exchange=exchange,
routing_key=routing_key,
queue_arguments={'x-max-priority': 10}
)
# 创建一个消费者,并设置预取消息数量为10
consumer = Consumer(
queues=[queue], callbacks=[self.on_message],
prefetch_count=10
)
print('创建消费者 down')
return [consumer]
def on_message(self, body, message):
eventlet.spawn(self._on_message, body, message)
def _on_message(self, body, message):
logger.debug(f"Received message: {body}")
logger.debug(f'开始消费')
time.sleep(3)
logger.debug(f'结束消费')
message.ack() # 发送ACK确认消息接收
with Connection('amqp://pon:pon@192.168.38.191:5672//') as conn:
consumer = MyConsumer(conn)
consumer.run()
修改其实很简单,把回调函数改造一下,收到收到消息之后,立刻起一个协程处理,让协程到后台慢慢处理,保持回调函数永远不阻塞就好了
服务器托管,北京服务器托管,服务器租用 http://www.fwqtg.net
jd.item_search_shop(获得店铺的所有商品) 为了进行电商平台 的API开发,首先我们需要做下面几件事情。 1)开发者注册一个账号 2)然后为每个JD应用注册一个应用程序键(App Key) 。 3)下载JDAPI的SDK并掌握基本的API基础…