如何使用 pika 的扭曲连接重新连接消费者？

Question

我使用 pika 扭曲连接作为 RabbitMQ 消费者，这是我的代码：

@defer.inlineCallbacks
def run(connection):
    queue_name = 'aaa'
    channel = yield connection.channel()
    queue = yield channel.queue_declare(queue=queue_name, auto_delete=False, exclusive=False)
    yield channel.queue_bind(exchange='amq.direct',queue=queue_name,routing_key=queue_name)
    yield channel.basic_qos(prefetch_count=1)
    queue_object, consumer_tag = yield channel.basic_consume(queue=queue_name,no_ack=False)
    logger.info('[room server]start consume queue %s', queue_name)

    l = task.LoopingCall(read, queue_object)
    l.start(0.1)


@defer.inlineCallbacks
def read(queue_object):
    ch,method,properties,body = yield queue_object.get()
    try:
        data = json.loads(body)
        head_code = data['head_code']
        openid = data['openid']
        message_content = data['message_content']
        conn_id = -1
        try:
            conn_id = data['conn_id']
        except:
            pass
        message_dispatcher(head_code, openid, message_content, conn_id)
        yield ch.basic_ack(delivery_tag=method.delivery_tag)
    except ValueError as e:
        logger.error('[error!]error body %s' % body)
        yield ch.basic_ack(delivery_tag=method.delivery_tag)

credentials = pika.PlainCredentials(config.RABBITMQ_USERNAME, config.RABBITMQ_PASSWD)
parameters = pika.ConnectionParameters(credentials=credentials)
cc = protocol.ClientCreator(reactor, twisted_connection.TwistedProtocolConnection, parameters)

def got_error(failure, d):
    logger.error(failure)
    d = cc.connectTCP(config.RABBITMQ_HOST, config.RABBITMQ_PORT)


def start():
    d = cc.connectTCP(config.RABBITMQ_HOST, config.RABBITMQ_PORT)
    d.addCallback(lambda protocol: protocol.ready)
    d.addCallback(run)
    d.addErrback(got_error, d)

我的问题是连接中断时，重新连接过程不起作用：

如何重新连接？

Answer 1

根据TwistedProtocolConnection docstring，可以提供on_close_callback 函数来处理连接终止。在此函数中，reason_code and reason_text 必须是 args。因此，创建另一个 on_close 回调来处理连接终止及其发生的原因，然后执行必要的逻辑以连接到 RabbitMQ：

def connection_termination(reason_code, reason_text):
    """
    Log the reasons why the connection terminates and then reconnect
    """
    # put your connection code here
    # incrementally space out your reconnections, eg. 2 seconds, if fail, 5 seconds, if fail 10 seconds, etc...

一旦您有了该代码，那么您的 ClientCreator 代码应遵循以下示例：

cc = protocol.ClientCreator(reactor, twisted_connection.TwistedProtocolConnection, parameters, connection_termination)

很遗憾，我目前无法对此进行测试，但它应该可以工作。您已经掌握了大部分逻辑，所以剩下的留给您作为练习；）如果您有任何问题，请发表评论，如果您有解决方案，请向其他人展示最终结果。