标签: cloudamqp

我在Heroku上使用Django/Celery和CloudAMQP插件.

在我的设置文件中,如果我设置BROKER_POOL_LIMIT为除了以外的任何值None,我很快就会超过我的CloudAMQP连接限制6.

我想知道为什么我会有比我指定的连接更多的连接BROKER_POOL_LIMIT以及如何设置BROKER_POOL_LIMIT一个大于None不超过连接限制的值.如果它有用,这是我的配置文件中指定的工作进程类型

worker python manage.py celery worker -B --loglevel=info

我在Heroku上运行Flask应用程序(在免费套餐上),并在使用apply_async安排任务时遇到麻烦.如果我安排了两个以上的任务,我得到一个长堆栈跟踪,但有异常:

AccessRefused(403,U "ACCESS_REFUSED - 访问虚拟主机 'rthtwchf' 交换 'celeryresults' 拒绝了用户rthtwchf'",(40,10), 'Exchange.declare')

奇怪的是前两个任务(在重新启动我的所有进程之前)似乎总是没有问题.

搜索引擎侦探一点点让我/sf/ask/1475033451/这使得它看起来像一个权限问题,但我假设Heroku CloudAMPQ服务已经处理好了.

任何建议表示赞赏!

禁用八卦，混杂和心跳对我的芹菜工人有什么影响？

为了减少发送到CloudAMQP的消息数量，使其不超出免费计划的范围，我决定遵循以下建议。因此，我使用了选项--without-gossip --without-mingle --without-heartbeat。从那时起，我一直在所有芹菜项目中默认使用这些选项，但是我不确定是否有我不知道的副作用。

请注意：

• 我们现在移到Redis代理，并且对发送到代理的邮件数量没有太多限制
• 我们有多个运行多个芹菜工人队列的实例

期待在AppHarbor上构建一个在MVC4上运行的Web应用程序.为了响应性和性能,将通过在消息队列上放置消息来处理稍长的运行任务(通常,生成/发送电子邮件,调整图像大小,支付事务处理等).

在那里的某个地方也会有一个或多个工人,将消息出列并处理因此需要处理的任何内容.中央排队机制是RabbitMQ,特别是通过AppHarbor提供的托管CloudAMQP服务.从理论上讲,这种架构可以通过增加更多工作人员来实现"无限"的可扩展性

现在,为了良好的架构,可测试性等,我想将RabbitMQ置于一个或多个易于模拟的接口之后.在定义这些接口时,我需要考虑一些注意事项.

• 在我获得任何付费用户之前,我仅限于免费的CloudAMQP产品.这意味着最多三个同时连接.
• 鉴于先前的约束,我想尝试将自己限制为每个应用程序一个连接.一个用于MVC应用程序,每个工作一个.而已.
• RabbitMQ中的连接旨在长期生活.尽管如此,许多示例仍然显示了明确打开连接(然后是通道),发送消息然后再次关闭它的代码.
• 对于多线程,可以使用相同的连接,但不能使用相同的通道.据我所知,可以在多线程应用程序中共享连接,然后打开几个通道,例如每个线程一个.
• 我的MVC应用程序通常只是一个发布者.工作者应用程序既可以是消费者也可以是发布者 - 例如,工作人员可以接收处理付款的消息,这将导致其发布消息以向用户调用指示成功或失败的电子邮件消息.
• 对于MVC应用程序,连接通常仅在非常短的时间内使用 - 用于排队消息.
• 对于工人来说,我正在考虑长时间的连接,或多或少永久地打开工人的一生.

好的,太好了,所以这很多东西.在这个项目之前没有任何RabbitMQ的经验,我被一些额外的要点所困扰.

• 界面隔离原理.我应该把它分成两个独立的接口 - 比如IQueueServiceConsumer和IQueueServiceProducer - 还是把它拿得太远了？我发现我总是可以使用SRP等将事物进一步划分为更多的原子单位,而且我不希望鲍勃叔叔狩猎我(因为我在界面名称中比我更多),但我想知道我有多远拿着这个.
• 鉴于我只有一个网络服务器,至少在开始时,是否真的想要在网络应用程序中实现与队列的长期连接？打开和关闭它们给出了一次发生几次的理论机会,这意味着冲突和潜在的信息丢失了.丢失的邮件是不可接受的.
• 在这种情况下,我如何处理连接的生命周期？在我的Ninject模块中打开它(我的界面实际上绑定到RabbitMQ特定的实现),并在应用程序被回收时以某种方式断开它？我怎么能这样做？
• 对于工人来说,生活似乎更容易.在接口上放置一个方法来打开连接,然后将通道提供给线程以获取消息.通过适当调用CloseConnection确保应用程序运行良好.或者实现IDisposable.
• 有可能 - 无论多么小 - RabbitMQ可能在将来被其他东西取代,因此我不希望我的接口过于特定于特定的服务总线(从某种意义上说,这就是我使用它的方式)实现.

在构建类似的应用程序时,是否有其他人已经完成了同样的挑战？您对消息队列的接口的体系结构和实际实现有什么建议吗？我只是对此感到神经质,还​​是我的担忧值得肯定？

如果重要,我的当前消息传递需求由单向消息覆盖,除了在处理完成后确认收到的消息之外,不需要任何响应.

感谢您的任何见解!

由于 Celery 文档建议使用 Flower 进行监控，我试图让它与托管的 RabbitMQ 提供程序 (CloudAMQP) 一起使用

celery flower --broker=amqp://username:password@lemur.cloudamqp.com/vhost --broker_api=https://lemur.cloudamqp.com/api/ --basic_auth=username:password

我可以在 CloudAMQP 管理控制台中看到如下所示的 500 内部服务器错误，这似乎表明密码错误，而我却没有。

The server encountered an error while processing this request:
{error,function_clause,
       [{rabbit_auth_backend_internal,check_password,
                                      [<<"password">>,
                                       <<>>]},
        {rabbit_auth_backend_internal,internal_check_user_login,2},
        {rabbit_access_control,'-check_user_login/2-fun-0-',4},
        {lists,foldl,3},
        {rabbit_mgmt_util,is_authorized,4},
        {webmachine_resource,resource_call,3},
        {webmachine_resource,do,3},
        {webmachine_decision_core,resource_call,1}]}

那么如何使用 cloudamqp 监控 celery 呢？

有没有更好的解决方案（监控/任务队列）？

我有一个Django应用程序,我已经部署到Heroku.这个应用程序使用Celery进行消息排队,我使用RabbitMQ在本地运行它,没有任何意外.

不幸的是,当我将这个宝宝部署到Heroku时,我发现RabbitMQ插件不可用,而且我必须使用CloudAMQP.CloudAMQP和Heroku的文档让我相信我可以使用Celery(即使他们推荐Pika),但是当我尝试部署时,我的调度程序和工作进程都会出现粗糙的连接错误.以下是确切的错误:

2012-07-09T16:46:22+00:00 app[scheduler.1]: [2012-07-09 11:46:22,234: ERROR/Beat] Celerybeat: Connection error: [Errno 111] Connection refused. Trying again in 2.0 seconds...
2012-07-09T16:46:23+00:00 app[worker.1]: [2012-07-09 11:46:23,852: ERROR/MainProcess] Consumer: Connection Error: [Errno 111] Connection refused. Trying again in 2 seconds...

我应该注意我的Heroku配置变量有一个CLOUDAMQP_URL,所以这不应该是一个问题？

如果有人使用CloudAMQP和Django/Heroku,可以给我一些关于如何确保Celery可以与经纪人联系的指导,我将不胜感激.

有没有人成功让MassTransit与AppHarbor和CloudAMQP合作？

我正在忍受它的一段时间.

我让发布者(网站)发送消息,但服务器(后台工作者)似乎没有接收它们.

我担心的一件事是MT需要rabbitmq方案,而CloudAMQP将方案设置为amqp.

我在配置总线时交换了方案(从amqp到rabbitmq)并注意到消息地址中的方案是rabbitmq,这很有意义,因为我更换了它们.但我想知道他们是否必须是服务器的amqp来接他们？

这是一个简单的消息,我已发送,它到达RabbitMQ但服务器没有提取它.

message_id: 08cf2cbc-5b4f-14dd-1231-381f8b520000
delivery_mode:  2
headers:    
Content-Type:   application/vnd.masstransit+json
Payload
614 bytes
Encoding: string
{

  "destinationAddress": "rabbitmq://98eabe2a-aae8-464c-8555-855518dd87d0_apphb.com:*********@lemur.cloudamqp.com/98eabe2a-aae8-464c-8555-855518dd87d0_apphb.com/Messages.Product:ProductCreatedEvent",

  "headers": {},

  "message": {

    "id": "dd6ecfaa-60d2-4cd4-8cd6-a08a00e872fb"

  },

  "messageType": [

    "urn:message:Messages.Product:ProductCreatedEvent"

  ],

  "retryCount": 0,

  "sourceAddress": "rabbitmq://98eabe2a-aae8-464c-8555-855518dd87d0_apphb.com:**********@lemur.cloudamqp.com/98eabe2a-aae8-464c-8555-855518dd87d0_apphb.com/enterprise_web"

}

谢谢,乔

编辑:感谢Carl指出网址中的密码

我在 Heroku 上运行的 Django 站点使用 CloudAMQP 来处理其计划的 Celery 任务。CloudAMQP 注册的消息比我的任务多得多，我不明白为什么。

例如，在过去的几个小时里，我将运行大约 150 个计划任务（两个每分钟运行一次，另一个每五分钟运行一次），但是 CloudAMQP 控制台的消息计数增加了大约 1,300。

我的相关 Django 设置：

BROKER_URL = os.environ.get("CLOUDAMQP_URL", "")
BROKER_POOL_LIMIT = 1
BROKER_HEARTBEAT = None
BROKER_CONNECTION_TIMEOUT = 30
CELERY_ACCEPT_CONTENT = ['json',]
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_RESULT_EXPIRES = 7 * 86400
CELERY_SEND_EVENTS = False
CELERY_EVENT_QUEUE_EXPIRES = 60
CELERY_RESULT_BACKEND = None
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'

我的档案：

web: gunicorn myproject.wsgi --log-file -
main_worker: python manage.py celery worker --beat --without-gossip --without-mingle --without-heartbeat --loglevel=info

查看 Heroku 日志，我只能看到我期望的正在运行的计划任务的数量。

大多数情况下，RabbitMQ 概览图往往如下所示：

我对 …

我们必须为我们的ROR Application实现排队系统.我们已经评估了以下选项:

Amazon SQS:高可用性但性能相对较慢.需要持续的poling.

CloudAmqp:看起来很有希望但对这种支持感到怀疑.

EC2上的RabbitMq设置:需要用户带宽来管理设置,如果设置中出现某些问题,可能会导致停机.

目前,没有任何专门的团队/人员可以全职管理设置,因此在Ec2上实施我们自己的RabbitMq设置可能会在出现问题时导致停机.
我想知道考虑这种情况是我们最好的选择吗？