Установить время ожидания для асинхронного режима Pika ioloop (RabbitMQ)

Мне нужно иметь возможность изящно остановить потребителя (работника), который работает в iloop Pika. Рабочий должен остановиться через 60 секунд. Обрабатываемые в настоящее время сообщения должны быть завершены.

Я попытался поместить connection.close() внутри функции обратного вызова, но это остановило только текущий поток, а не полный цикл ioloop. И это выдавало ужасную ошибку.

См. строку 16 и следующую в моем коде: я использовал (базовый пример о Pika ioloop http://pika.github.com/connecting.html#cps-example:

    from pika.adapters import SelectConnection
    channel = None
    def on_connected(connection):
        connection.channel(on_channel_open)

    def on_channel_open(new_channel):
        global channel
        channel = new_channel
        channel.queue_declare(queue="test", durable=True, exclusive=False, auto_delete=False, callback=on_queue_declared)

    def on_queue_declared(frame):
        channel.basic_consume(handle_delivery, queue='test')

    def handle_delivery(channel, method, header, body):
        print body

        # timer stuff which did NOT work
        global start_time, timeout, connection
        time_diff = time.time()-start_time
        if time_diff > timeout:
            #raise KeyboardInterrupt
            connection.close()

    timeout = 60
    start_time = time.time()

    connection = SelectConnection(parameters, on_connected)

    try:
        connection.ioloop.start()
    except KeyboardInterrupt:
        connection.close()
        connection.ioloop.start()

python rabbitmq pika
person Sebastian    schedule 18.11.2011    source источник

Ответы (2)


arrow_upward
10
arrow_downward

Вы можете прикрепить функцию обратного вызова по тайм-ауту к открытому соединению. Вот дополнительный код для вашего примера.

timeout = 60

def on_timeout():
  global connection
  connection.close()

connection.add_timeout(timeout, on_timeout)
person mike    schedule 18.11.2011

arrow_upward
-4
arrow_downward

Вы можете попробовать использовать:

connection.ioloop.stop()
person Arkadiy Kapustinskiy    schedule 07.12.2011