我想使用用java nio编写的彗星服务器来发送实时更新.接收信息时,我希望它扫描数据,并通过rabbitmq将任务发送到工作线程.理想情况下,我希望芹菜服务器位于兔子的另一端,管理将处理这些任务的工作线程池.
然而,根据我的理解,芹菜通过坐在rabbitmq的两端来工作,它通过嵌入到消费者和生产者的代码中而基本上接管了生产者和消费者的角色.有没有办法像我上面描述的那样设置芹菜?谢谢
是的,cource!
您可以添加Custom Message Consumers到芹菜应用程序.
请参阅芹菜文档中的Extensions and Bootsteps.
以下是上述链接中示例代码的一部分:
from celery import Celery
from celery import bootsteps
from kombu import Consumer, Exchange, Queue
my_queue = Queue('custom', Exchange('custom'), 'routing_key')
app = Celery(broker='amqp://')
class MyConsumerStep(bootsteps.ConsumerStep):
def get_consumers(self, channel):
return [Consumer(channel,
queues=[my_queue],
callbacks=[self.handle_message],
accept=['json'])]
def handle_message(self, body, message):
print('Received message: {0!r}'.format(body))
message.ack()
app.steps['consumer'].add(MyConsumerStep)
Run Code Online (Sandbox Code Playgroud)
测试一下:
python -m芹菜 - 主要工作者
| 归档时间: |
|
| 查看次数: |
2637 次 |
| 最近记录: |