使用芹菜从 AMQP 消耗 json DATA答案

【问题标题】：Consume json DATA from AMQP with celery使用芹菜从 AMQP 消耗 json DATA
【发布时间】：2015-03-26 14:16:55
【问题描述】：

我在网络分析方面遇到技术问题。

Netflow 流由 PMACCT 收集，并将此流发送到 rabbitMQ 进行分析。分析将由 python 脚本完成，添加一些字段，如 dns ptr 和其他网络信息。

我使用 pika 编写了一个自定义消费者脚本，但性能很差。我曾尝试使用 python-multiprocessing 来提高性能，但 Celery 的性能会更好（我猜）。

你知道我是否可以使用 Celery 来使用这些数据（由 PMACCT 发送和序列化）？数据格式是带字段的简单 JSON 对象（格式与 Celery 任务格式不匹配）。其他图书馆可以帮助我吗？

最终目标是在多台服务器上对流分析进行负载平衡（数 GB 的流）。

【问题讨论】：

Celery 应该没问题，但如果你需要性能，你应该考虑使用 librabbitmq 和 Celery (github.com/celery/librabbitmq)。其他选择是使用线程安全库，例如我自己的 amqp-storm github.com/eandersson/amqp-storm/blob/master/examples/…
好的@eandersson。 amqp-storm 正是我想要的。我可以将 librabbitmq 与 amqp-storm 一起使用吗？
不，它们都是 AMQP 库。 AMQP-Storm 不如 librabbitmq 快，但使用 AMQP-Storm 更容易，因为它是线程安全的，而 librabbitmq 则不是。
嗯，好吧。我正在尝试使用 AMQP-Storm。让我检查一下。仅供参考，这是什么意思：channel.basic.qos(int)？
这是 RabbitMQ 预取的设置。如果您在一个队列上有多个消费者，我建议您保持较低的价值，例如100.rabbitmq.com/consumer-prefetch.html

标签： python json celery amqp pika

【解决方案1】：

我将此作为 cmets 发布，但我认为我会将其编译为答案。

Celery 应该可以很好地处理 json 数据。

至于性能，我假设您使用的是 Pika BlockingConnection 模块。这是一个非常慢的实现，是crippled with bugs。如果您希望将性能更改为 AMQP 库的 C 实现，例如librabbitmq 应该会大大加快速度，librabbitmq 的一个优势是它可以与Celery 一起使用。

librabbitmq 的缺点是它不是线程安全的。有一些线程安全的替代方案可用；比如我自己的AMQP-Storm，或者来自 pika 维护者的Rabbitpy。

【讨论】：