Сельдерей неожиданно закрывается после более длительного бездействия

Question

Поэтому я использую RabbitMQ + Celery для создания простой архитектуры RPC. У меня есть один брокер сообщений RabbitMQ и один удаленный рабочий, который управляет дегустацией сельдерея.

Существует третий сервер, который предоставляет тонкий API RESTful. Когда он получает HTTP-запрос, он отправляет задачу удаленному рабочему, ждет ответа и возвращает ответ.

Это отлично работает большую часть времени. Однако я замечаю, что после более длительного бездействия (скажем, 5 минут отсутствия входящих запросов), работник сельдерея ведет себя странно. Первая 3 задачи, полученная после длительного возвращения неактивности этой ошибки:

exchange.declare: connection closed unexpectedly 

После трех ошибочных задач он работает снова. Если в течение более длительного периода времени нет заданий, происходит то же самое. Есть идеи?

Мой сценарий инициализации для сельдерея работника:

# description "Celery worker using sync broker" 

console log 

start on runlevel [2345] 
stop on runlevel [!2345] 

setuid richard 
setgid richard 

script 
chdir /usr/local/myproject/myproject 
exec /usr/local/myproject/venv/bin/celery worker -n celery_worker_deamon.%h -A proj.sync_celery -Q sync_queue -l info --autoscale=10,3 --autoreload --purge 
end script 

respawn 

Мой сельдерей конфигурации:

# Synchronous blocking tasks 
BROKER_URL_SYNC = 'amqp://guest:guest@localhost:5672//' 
# Asynchronous non blocking tasks 
BROKER_URL_ASYNC = 'amqp://guest:guest@localhost:5672//' 

#: Only add pickle to this list if your broker is secured 
#: from unwanted access (see userguide/security.html) 
CELERY_ACCEPT_CONTENT = ['json'] 
CELERY_TASK_SERIALIZER = 'json' 
CELERY_RESULT_SERIALIZER = 'json' 
CELERY_TIMEZONE = 'UTC' 
CELERY_ENABLE_UTC = True 
CELERY_BACKEND = 'amqp' 

# http://docs.celeryproject.org/en/latest/userguide/tasks.html#disable-rate-limits-if-they-re-not-used 
CELERY_DISABLE_RATE_LIMITS = True 

# http://docs.celeryproject.org/en/latest/userguide/routing.html 
CELERY_DEFAULT_QUEUE = 'sync_queue' 
CELERY_DEFAULT_EXCHANGE = "tasks" 
CELERY_DEFAULT_EXCHANGE_TYPE = "topic" 
CELERY_DEFAULT_ROUTING_KEY = "sync_task.default" 
CELERY_QUEUES = { 
    'sync_queue': { 
     'binding_key':'sync_task.#', 
    }, 
    'async_queue': { 
     'binding_key':'async_task.#', 
    }, 
} 

Любые идеи?

EDIT:

Хорошо, теперь, кажется, случается случайно. Я заметил это в RabbitMQ журналах:

=WARNING REPORT==== 6-Jan-2014::17:31:54 === 
closing AMQP connection <0.295.0> (some_ip_address:36842 -> some_ip_address:5672): 
connection_closed_abruptly 

Answer 1

connection_closed_abruptly возникает, когда клиенты отсоединением без надлежащего AMQP протокола выключения:

channel.close (...)

Request a channel close.

This method indicates that the sender wants to close the channel. This may be due to internal conditions (e.g. a forced shut-down) or due to an error handling a specific method, i.e. an exception. When a close is due to an exception, the sender provides the class and method id of the method which caused the exception.

After sending this method, any received methods except Close and Close-OK MUST be discarded. The response to receiving a Close after sending Close must be to send Close-Ok .

channel.close-ok():

Confirm a channel close.

This method confirms a Channel.Close method and tells the recipient that it is safe to release resources for the channel.

A peer that detects a socket closure without having received a Channel. Close-Ok handshake method SHOULD log the error.

Here - проблема.

Вы можете установить пользовательские настройки для BROKER_HEARTBEAT и BROKER_HEARTBEAT_CHECKRATE и проверьте еще раз, например:

BROKER_HEARTBEAT = 10 
BROKER_HEARTBEAT_CHECKRATE = 2.0 

Answer 2

Ваш RabbitMQ сервер или ваш сельдерея работник за балансировки нагрузки любой шанс? Если да, то балансировщик нагрузки закрывает TCP-соединение после некоторого периода бездействия. В этом случае вам придется активировать пульс со стороны клиента (работника). Если вы это сделаете, я бы не рекомендовал использовать чистую Python amqp lib для этого. Вместо этого замените его на librabbitmq.