RabbitMQ pika.exceptions.ConnectionClosed (-1, "ошибка (104, 'Сброс соединения по пиру')") - PullRequest
Новая
       65

RabbitMQ pika.exceptions.ConnectionClosed (-1, "ошибка (104, 'Сброс соединения по пиру')")

0 голосов
/ 24 октября 2018

У меня есть очередь задач в RabbitMQ с несколькими производителями (12) и одним потребителем для тяжелых задач в веб-приложении.Когда я запускаю потребителя, он начинает снимать некоторые сообщения, прежде чем вылетает с этой ошибкой: 

Traceback (most recent call last):
File "jobs.py", line 42, in <module> jobs[job](config)
File "/home/ec2-user/project/queue.py", line 100, in init_queue
channel.start_consuming()
File "/usr/lib/python2.7/site-packages/pika/adapters/blocking_connection.py", line 1822, in start_consuming
self.connection.process_data_events(time_limit=None)
File "/usr/lib/python2.7/site-packages/pika/adapters/blocking_connection.py", line 749, in process_data_events
self._flush_output(common_terminator)
File "/usr/lib/python2.7/site-packages/pika/adapters/blocking_connection.py", line 477, in _flush_output
result.reason_text)
pika.exceptions.ConnectionClosed: (-1, "error(104, 'Connection reset by peer')")

Код производителя: 

message = {'image_url': image_url, 'image_name': image_name, 'notes': notes}

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='tasks_queue')
channel.basic_publish(exchange='', routing_key=queue_name, body=json.dumps(message))

connection.close()

И единственный код потребителя (одинclashing): 

def callback(self, ch, method, properties, body):
    """Callback when receive a message."""
    message = json.loads(body)
    try:
        image = _get_image(message['image_url'])
    except:
        sys.stderr.write('Error getting image in note %s' % note['id'])
   # Crop image with PIL. Not so expensive
   box_path = _crop(image, message['image_name'], box)

   # API call. Long time function
   result = long_api_call(box_path)

   if result is None:
       sys.stderr.write('Error in note %s' % note['id'])
       return
   # update the db
   db.update_record(result)


connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='tasks_queue')
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback_obj.callback, queue='tasks_queue', no_ack=True)
channel.start_consuming()

Как видите, есть 3 дорогих функции для сообщения.Одна задача обрезки, один вызов API и одно обновление базы данных.Без вызова API клиент работает нормально.

Заранее спасибо

1 Ответ

0 голосов
/ 25 октября 2018

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

missed heartbeats from client, timeout: 60s

То, что происходит, - то, что ваш long_api_call блокирует цикл ввода / вывода Пики.Pika - это очень легковесная библиотека, которая не запускает потоки в фоновом режиме, поэтому вы должны кодировать таким образом, чтобы не блокировать цикл ввода / вывода Пики дольше, чем интервал сердцебиения.RabbitMQ считает, что ваш клиент умер или не отвечает и принудительно закрывает соединение.

Пожалуйста, посмотрите мой ответ здесь , который ссылается на этот пример кода , показывающий, как правильно выполнитьдлительное задание в отдельном потоке.Вы по-прежнему можете использовать no_ack=True, просто пропустите вызов ack_message.

ПРИМЕЧАНИЕ: команда RabbitMQ контролирует почтовое сообщение rabbitmq-userslist и только иногда отвечает на вопросы по StackOverflow.

...