使用带有SQS的Celery时,Errno 111连接被拒绝

Question

我运行celery status或celery purge命令时遇到问题.

 File "/usr/bin/celery", line 11, in <module>
    sys.exit(main())
  File "/usr/lib/python2.7/site-packages/celery/__main__.py", line 30, in main
    main()
  File "/usr/lib/python2.7/site-packages/celery/bin/celery.py", line 81, in main
    cmd.execute_from_commandline(argv)
  File "/usr/lib/python2.7/site-packages/celery/bin/celery.py", line 769, in execute_from_commandline
    super(CeleryCommand, self).execute_from_commandline(argv)))
  File "/usr/lib/python2.7/site-packages/celery/bin/base.py", line 306, in execute_from_commandline
    return self.handle_argv(self.prog_name, argv[1:])
  File "/usr/lib/python2.7/site-packages/celery/bin/celery.py", line 761, in handle_argv
    return self.execute(command, argv)
  File "/usr/lib/python2.7/site-packages/celery/bin/celery.py", line 693, in execute
    ).run_from_argv(self.prog_name, argv[1:], command=argv[0])
  File "/usr/lib/python2.7/site-packages/celery/bin/base.py", line 310, in run_from_argv
    sys.argv if argv is None else argv, command)
  File "/usr/lib/python2.7/site-packages/celery/bin/base.py", line 372, in handle_argv
    return self(*args, **options)
  File "/usr/lib/python2.7/site-packages/celery/bin/base.py", line 269, in __call__
    ret = self.run(*args, **kwargs)
  File "/usr/lib/python2.7/site-packages/celery/bin/celery.py", line 472, in run
    replies = I.run('ping', **kwargs)
  File "/usr/lib/python2.7/site-packages/celery/bin/celery.py", line 324, in run
    return self.do_call_method(args, **kwargs)
  File "/usr/lib/python2.7/site-packages/celery/bin/celery.py", line 346, in do_call_method
    callback=self.say_remote_command_reply)
  File "/usr/lib/python2.7/site-packages/celery/bin/celery.py", line 385, in call
    return getattr(i, method)(*args)
  File "/usr/lib/python2.7/site-packages/celery/app/control.py", line 99, in ping
    return self._request('ping')
  File "/usr/lib/python2.7/site-packages/celery/app/control.py", line 70, in _request
    timeout=self.timeout, reply=True,
  File "/usr/lib/python2.7/site-packages/celery/app/control.py", line 306, in broadcast
    limit, callback, channel=channel,
  File "/usr/lib/python2.7/site-packages/kombu/pidbox.py", line 283, in _broadcast
    chan = channel or self.connection.default_channel
  File "/usr/lib/python2.7/site-packages/kombu/connection.py", line 755, in default_channel
    self.connection
  File "/usr/lib/python2.7/site-packages/kombu/connection.py", line 740, in connection
    self._connection = self._establish_connection()
  File "/usr/lib/python2.7/site-packages/kombu/connection.py", line 695, in _establish_connection
    conn = self.transport.establish_connection()
  File "/usr/lib/python2.7/site-packages/kombu/transport/pyamqp.py", line 112, in establish_connection
    conn = self.Connection(**opts)
  File "/usr/lib/python2.7/site-packages/amqp/connection.py", line 165, in __init__
    self.transport = create_transport(host, connect_timeout, ssl)
  File "/usr/lib/python2.7/site-packages/amqp/transport.py", line 294, in create_transport
    return TCPTransport(host, connect_timeout)
  File "/usr/lib/python2.7/site-packages/amqp/transport.py", line 95, in __init__
    raise socket.error(last_err)
socket.error: [Errno 111] Connection refused

我正在使用SQS BROKER_URL.任务运行正常,但是当我想清除队列(celery purge -f)上的任务时,我得到上述错误.

software -> celery:3.1.11 (Cipater) kombu:3.0.18 py:2.7.5
            billiard:3.3.0.17 py-amqp:1.4.5
platform -> system:Linux arch:64bit, ELF imp:CPython
loader   -> celery.loaders.default.Loader
settings -> transport:amqp results:disabled

我的服务器端口22,80,443,8000端口打开,SQS celery队列中有大量消息,因此芹菜和SQS之间的连接应该没问题.

Answer 1

根据命令状态和清除的文档,您需要为芹菜提供您所指的芹菜应用程序,以便它知道要使用的代理.只需键入$celery purge或输入$celery status芹菜,就不知道你所针对的是什么芹菜应用程序,因此失败了.

因此,请转到您的芹菜应用程序

$cd /path/to/your/celery/app/directory

然后在您的应用程序上调用celery purge.在这个例子中我的目录有celeryapp.py,内容是:

from config import config
from celery import Celery
celery_app = Celery('tasks', 
                    backend=config.celery_backend_uri, 
                    broker=config.celery_broker_uri)
celery_app.conf.update(
    CELERY_IMPORTS=(
        'app.module_a.tasks',   # we're not including our tasks here as
        'app.module_b.tasks',   # our tasks are in other files listed here
    )
)

内容并不像我们的电话那么重要,但它们是为了表明我们的芹菜应用程序内部celeryapp.py所以我可以打电话

$celery -A celeryapp status
worker-name-a@node-name: OK
worker-name-b@node-name: OK

要么

$celery -A celeryapp purge
WARNING: This will remove all tasks from queue: celery.
         There is no undo for this operation! 
(to skip this prompt use the -f option)
Are you sure you want to delete all tasks (yes/NO)? yes
No messages purged from 1 queue

我在这里有一个类似的问题,Sol似乎确认芹菜将输出此错误,如果没有提供应用程序说明

如果你不给它应用程序的位置,它将如何知道要使用的代理运输？