Я использую сельдерей 4.3.0 и rabbitmq 3.6.6. Существует также Haproxy для балансировки rabbitmq, но сейчас он использует только один узел rabbitmq.
Время ожидания клиент / сервер в Haproxy 24 часа.
Иногда я получаю сообщение об ошибке, и состояние выполнения задачи не возвращается, в то время как задачи продолжают выполняться:
Received 0x00 while expecting 0xce
Traceback (most recent call last):
File "/var/www/serv/app/mod_axmsg/dispatch.py", line 376, in execute
result_value = func_meta.func(*task.args_list, **task.kwargs_dict)
File "/var/www/serv/app/mod_rmc/libs/health.py", line 91, in update_all_health
health.update_all_health()
File "/var/www/serv/app/mod_core/libs/health.py", line 515, in update_all_health
result.join()
File "/usr/local/serv/lib/python3.5/site-packages/celery/result.py", line 765, in join
interval=interval, no_ack=no_ack, on_interval=on_interval,
File "/usr/local/serv/lib/python3.5/site-packages/celery/result.py", line 226, in get
on_message=on_message,
File "/usr/local/serv/lib/python3.5/site-packages/celery/backends/asynchronous.py", line 188, in wait_for_pending
for _ in self._wait_for_pending(result, **kwargs):
File "/usr/local/serv/lib/python3.5/site-packages/celery/backends/asynchronous.py", line 255, in _wait_for_pending
on_interval=on_interval):
File "/usr/local/serv/lib/python3.5/site-packages/celery/backends/asynchronous.py", line 56, in drain_events_until
yield self.wait_for(p, wait, timeout=1)
File "/usr/local/serv/lib/python3.5/site-packages/celery/backends/asynchronous.py", line 65, in wait_for
wait(timeout=timeout)
File "/usr/local/serv/lib/python3.5/site-packages/celery/backends/rpc.py", line 63, in drain_events
return self._connection.drain_events(timeout=timeout)
File "/usr/local/serv/lib/python3.5/site-packages/kombu/connection.py", line 323, in drain_events
return self.transport.drain_events(self.connection, **kwargs)
File "/usr/local/serv/lib/python3.5/site-packages/kombu/transport/pyamqp.py", line 103, in drain_events
return connection.drain_events(**kwargs)
File "/usr/local/serv/lib/python3.5/site-packages/amqp/connection.py", line 505, in drain_events
while not self.blocking_read(timeout):
File "/usr/local/serv/lib/python3.5/site-packages/amqp/connection.py", line 510, in blocking_read
frame = self.transport.read_frame()
File "/usr/local/serv/lib/python3.5/site-packages/amqp/transport.py", line 280, in read_frame
'Received {0:#04x} while expecting 0xce'.format(ch))
amqp.exceptions.UnexpectedFrame: Received 0x00 while expecting 0xce
Мой код: celery.py
import logging
from app import db
from celery import Celery
LOG = logging.getLogger(__name__)
def make_celery(flask_app):
broker = flask_app.config['AMQP_URL']
celery = Celery('mod_celery', broker=broker, backend="rpc://")
celery.conf.update(flask_app.config)
class ContextTask(celery.Task):
def __call__(self, *args, **kwargs):
with flask_app.app_context():
return self.run(*args, **kwargs)
celery.Task = ContextTask
return celery
def in_tx(task_function):
def tx_wrapper(*args):
try:
task_function(*args)
db.session.commit()
except Exception as ex:
msg = 'Could not execute task {}. Reason: {}.' \
.format(task_function.__name__, str(ex))
LOG.error(msg)
db.session.rollback()
return tx_wrapper
И файл tasks.py
from app import celery
from .celery import in_tx
@celery.task(name='app.mod_celery.update_client_health_task')
@in_tx
def update_client_health_task(client_id):
from app.mod_core.libs.health import update_client_health
update_client_health(client_id)
В чем может быть проблема? Пожалуйста, не предлагайте мне redis.
Я не разработчик, и я не написал код, но мне действительно нужно разобраться. Я новичок в python
Спасибо.