Пика падает без видимой причины

У меня проблема. Мой клиент pika постоянно вылетает с сообщением об ошибке.

Вот что происходит:

  1. RabbitMQ запущен, и производитель уже поместил сообщения в очередь
  2. Я запускаю свой скрипт на python, и он обрабатывает все пакеты, которые буферизируются в очереди
  3. Мой скрипт периодически выдает исключение: ConnectionClosed, однако я нигде ничего не закрываю

Это мой код:

import pika
import traceback

class RPCServer(object):


    def __init__(self, callback, cfg):
        self.cfg = cfg
        self.callback = callback
        self.credentials = None
        self.parameters = None
        self.connection = None
        self.channel = None
        self.counter = 0
        self.initalize_me()


    def initalize_me(self):
        self.credentials = pika.PlainCredentials(self.cfg.USER, self.cfg.PASSWORD)
        self.parameters = pika.ConnectionParameters(host=self.cfg.AMQP_HOST, credentials=self.credentials)
        self.connection = pika.BlockingConnection(self.parameters)
        self.channel = self.connection.channel()
        self.channel.exchange_declare(exchange=self.cfg.RPC_EXCHANGE_NAME, type="direct")
        self.channel.queue_declare(queue=self.cfg.RPC_QUEUE_NAME)
        self.channel.queue_bind(exchange=self.cfg.RPC_EXCHANGE_NAME, queue=self.cfg.RPC_QUEUE_NAME, routing_key=self.cfg.RPC_ROUTING_KEY)
        self.channel.basic_consume(self.rpc_callback, queue=self.cfg.RPC_QUEUE_NAME)
        print "init= " + str(self.cfg.RPC_EXCHANGE_NAME) + " -> " + str(self.cfg.RPC_QUEUE_NAME) + " -> " + str(self.cfg.RPC_ROUTING_KEY)


    def start_rpc_server(self):
        print "Server: Start listening for RPC requests..."
        try:
            self.channel.start_consuming()
        except:
            print "Exception: " + str(traceback.format_exc())
            self.initalize_me()
            self.start_rpc_server()


    def rpc_callback(self, ch, method, props, body):
        self.counter += 1
        if self.counter == 100:
            print "100 package processed..."
            self.counter = 0
        result = self.callback(body)
        properties = pika.BasicProperties(correlation_id=props.correlation_id)
        ch.basic_publish(exchange="", routing_key=props.reply_to, properties=properties, body=result)
        ch.basic_ack(delivery_tag=method.delivery_tag)

И это вывод, когда я запускаю его:

python RunPacer.py 
Initialize Configuration
Start Pacer
100 package processed...
100 package processed...
init= pacing_exchange_debug -> pacing_queue_debug -> pacing_routing_key_debug
Server: Start listening for RPC requests...
Exception: Traceback (most recent call last):
  File "/home/Tom/Pacer/amqp/RPCServer.py", line 46, in start_rpc_server
    self.channel.start_consuming()
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 955, in start_consuming
    self.connection.process_data_events()
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 243, in process_data_events
    raise exceptions.ConnectionClosed()
ConnectionClosed

100 package processed...
100 package processed...
init= pacing_exchange_debug -> pacing_queue_debug -> pacing_routing_key_debug
Server: Start listening for RPC requests...
Exception: Traceback (most recent call last):
  File "/home/Tom/Pacer/amqp/RPCServer.py", line 46, in start_rpc_server
    self.channel.start_consuming()
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 955, in start_consuming
    self.connection.process_data_events()
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 243, in process_data_events
    raise exceptions.ConnectionClosed()
ConnectionClosed

init= pacing_exchange_debug -> pacing_queue_debug -> pacing_routing_key_debug
Server: Start listening for RPC requests...
100 package processed...
100 package processed...
Exception: Traceback (most recent call last):
  File "/home/Tom/Pacer/amqp/RPCServer.py", line 46, in start_rpc_server
    self.channel.start_consuming()
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 955, in start_consuming
    self.connection.process_data_events()
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 243, in process_data_events
    raise exceptions.ConnectionClosed()
ConnectionClosed

Извините, мое описание очень неточное, однако это связано с тем, что я абсолютно не знаю, почему мой скрипт вылетает. Так что любой, действительно любой, совет будет полезен. Спасибо!

РЕДАКТИРОВАТЬ: Добавлены журналы ошибок rabbitmq:

=INFO REPORT==== 4-Dec-2014::12:55:42 ===
accepting AMQP connection <0.8947.0> (183.13.20.123:61598 -> 183.13.20.123:5672)

=ERROR REPORT==== 4-Dec-2014::12:55:42 ===
Error on AMQP connection <0.8947.0> (183.13.20.123:61598 -> 183.13.20.123:5672, vhost: '/', user: 'username', state: running), channel 1:
{amqp_error,unexpected_frame,
            "expected content body, got non content body frame instead",
            'basic.publish'}

=INFO REPORT==== 4-Dec-2014::12:55:43 ===
closing AMQP connection <0.8947.0> (183.13.20.123:61598 -> 183.13.20.123:5672)

Дополнительно: у меня есть Java-программа, работающая в той же очереди (фактически копия скрипта Python в Java), которая работает без каких-либо проблем.

Источник

1 ответ

Это явно ошибка в Pika (см. Здесь: https://github.com/pika/pika/issues/349). Так как эта проблема не была исправлена ​​больше года (отчет об ошибке 12/2013, это сообщение: 12/2014), я больше не буду полагаться на Пику.

Однако отличной (и быстрой) альтернативой (которая приводит к еще меньшему количеству кода) является librabbitmq: https://pypi.python.org/pypi/librabbitmq

Вот пример кода, который реализует вызов RPC с librabbitmq:

# -*- coding: utf-8 -*-

from librabbitmq import Connection
import uuid

class RPCClient(object):

    def __init__(self, cfg):
        self.cfg = cfg
        self.connection = Connection(host=cfg.AMQP_HOST, userid=cfg.USER, password=cfg.PASSWORD)
        self.channel = self.connection.channel()
        result = self.channel.queue_declare(exclusive=True)
        self.callback_queue = result.queue
        self.response = None
        self.corr_id = None
        self.channel.basic_consume(self.callback_queue, callback=self.process_response)


    def process_response(self, msg):
        if self.corr_id == msg.properties['correlation_id']:
            self.response = str(msg.body)


    def rpc_call(self, msg):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        props = {'reply_to' : str(self.callback_queue), 'correlation_id' : str(self.corr_id)}

        self.channel.basic_publish(msg, exchange=self.cfg.RPC_EXCHANGE_NAME,
                                   routing_key=self.cfg.RPC_ROUTING_KEY, **props)
        while self.response is None:
            self.connection.drain_events()
        return str(self.response)

И соответствующий RPCServer:

# -*- coding: utf-8 -*-

from librabbitmq import Connection

class RPCServer(object):

    def __init__(self, callback, cfg):
        self.cfg = cfg
        self.callback = callback
        self.connection = Connection(host=cfg.AMQP_HOST, userid=cfg.USER, password=cfg.PASSWORD)
        self.channel = self.connection.channel()
        self.channel.exchange_declare(cfg.RPC_EXCHANGE_NAME, "direct")
        self.channel.queue_declare(cfg.RPC_QUEUE_NAME)
        self.channel.queue_bind(cfg.RPC_QUEUE_NAME, cfg.RPC_EXCHANGE_NAME, cfg.RPC_ROUTING_KEY)
        self.channel.basic_consume(cfg.RPC_QUEUE_NAME, callback=self.rpc_callback)


    def start_rpc_server(self):
        while True: self.connection.drain_events()


    def rpc_callback(self, msg):
        resp = self.callback(msg.body)
        self.channel.basic_publish(resp, exchange="", routing_key=msg.properties['reply_to'], **msg.properties)
        self.channel.basic_ack(msg.delivery_info['delivery_tag'])
Источник
Другие вопросы по тегам