CeleryRabbitMQ分布式队列测试消息

Question

当我在终端上运行celery -A tasks worker时，我一直得到：ERROR/MainProcess] consumer: Cannot connect to amqp://ec2celeryuser。在

基本上，我要做的是让celery/rabbitmq在（2）个ec2实例中正常工作。将tasks.py中的一个愚蠢的任务传递给rabbitmq进行处理。在

1. 实例1-住宅rabbitMQ

目前运行RabbitMQ fine。如果我运行sudo rabbitmqctl status，它会输出：

Status of node 'rabbit@ip-xx-xxx-xxx-xx' ...
[{pid,786},

2。实例2-种植芹菜

我尝试在instance 2上对实例1运行celery，使用以下in终端：

芹菜-一个任务工人

我有一个文件celeryconfig.py：

^{pr2}$

我有一个文件client.py：

from tasks import add

result = add.delay(4, 4) # call task
result_sum = result.get(timeout=5) # wait to get result for a maximum of 5 seconds

我有一个文件tasks.py：

from celery import Celery

app = Celery('tasks', broker='amqp://ec2celeryuser:mypasshere@xx.xxx.xx.xx:5672/celeryserver1/')

@app.task
def add(x, y):
    return x + y

我已经正确地设置了一个vhost，一个用户ec2celeryuser，并授予该用户以下权限：

sudo rabbitmqctl set_permissions -p /celeryserver1 ec2celeryuser ".*" ".*" ".*"

如果我这样做：sudo rabbitmqctl list_users在RabbitMQ（实例1）上，它显示：

ec2celeryuser   []
guest   [administrator

我试过两个用户名和他们的密码，但没有改变。在

我一直在跟踪Celery Guide，和一个tutorial，运气不太好。在

我做错什么了？很明显有一个连接问题，但我做错了什么？在

谢谢你！在

Answer 1

所以我在这里看到了一些错误。首先，tasks.py中rabbitMQ的代理URL似乎不正确。它应该读像下面这样的东西。在

app = Celery('tasks', broker='amqp://ec2celeryuser:ec2celerypassword@xx.xxx.xx.xx/celeryserver1/')

此外，您可能希望在运行工作进程时指定希望celery提供服务的应用程序。您可以从tasks.py所在的目录运行celery -A tasks worker。在

另一件事是您在client.py中调用任务的代码似乎不正确。从celery文档中，可以按如下方式调用该任务：

from tasks import add

result = add.delay(4, 4) # call task
result_sum = result.get(timeout=5) # wait to get result for a maximum of 5 seconds

解决这些问题也许能解决你的问题，或者至少能让你更亲近。在

Answer 2

感谢用户natdempk帮助我修复队列的配置语法。在

问题是在rabbitmq中创建一个vhost，如下所示：

sudo rabbitmqctl add_vhost /celeryserver1

当它应该是：

^{pr2}$

然后，我不得不重置我的用户ec2celeryuser的权限，如下所示：

sudo rabbitmqctl set_permissions -p celeryserver1 ec2celeryuser ".*" ".*" ".*"

我意识到这是个问题：我访问了/var/log/rabbitmq/<last log file.log>

看到了：

=INFO REPORT==== 30-Apr-2014::12:45:58 ===
accepted TCP connection on [::]:5672 from xx.xxx.xxx.xxx:45964

=INFO REPORT==== 30-Apr-2014::12:45:58 ===
starting TCP connection <x.xxx.x> from from xx.xxx.xxx.xxx:45964

=ERROR REPORT==== 30-Apr-2014::12:46:01 ===
exception on TCP connection <x.xxx.x> from from xx.xxx.xxx.xxx:45964
{channel0_error,opening,
                {amqp_error,access_refused,
                            "access to vhost 'celeryserver1/' refused for user 'ec2celeryuser'",
                            'connection.open'}}

自从修理了主机，我现在愉快地看到：

[2014-04-30 13:08:10,101: WARNING/MainProcess] celery@ip-xx-xxx-xx-xxx ready.