0

Есть 2 потока, взаимодействующих c сервером RabbiMQ (библиотека pika):

В одном запускается Producer, отправляет сообщение и процесс заканчивается.

В другом - Consumer. Запускается первым. Ждёт пока Producer не отправит сообщение. Затем коллбек консьюмера перехватывает сообщение и обрабатывает его (в коллбеке также закрывается соединение -> поток заканчивает свою работу).

Есть вероятность, что до консьюмера сообщение может и не дойти.

Как задать ему время работы?

Варианта 2:

  1. Запустить 3ий поток, который остановит работу консьюмера (=> соответсвенно поток консюмера тоже прекратит свою работу) через какое-то время.

  2. В настройках консьюмера установить какой-то таймаут. Только какой параметр за это отвечает(пробовал heartbeat, blocked_connection_timeout и socket_timeout - не получается).

Интересует реализация второго варианта.

3
  • 1
    если вы параметры уже конкретные обсуждаете, то стоит минимальный полный пример кода привести минимальный воспроизводимый пример. Вы не пробовали асинхронным интерфейсом воспользоваться без всяких потоков? При использовании потоков и блокирующих функций, для timeout нужна кооперация со стороны этих конкретных блокирующих функций и вообще потоки снаружи нехорошо пытаться убить. В случае асинхронного интерфейса, есть общий для избранной библиотеки способ timeout усилить (к примеру, gevent.Timeout, asyncio.wait_for, IOLoop.run_sync, trio.move_on_after, etc).
    – jfs
    1 июн 2018 в 6:45
  • Большое спасибо. Пока остановился на том, чтобы вынести работу паблишера и консьюмера в отдельные скрипты. А их уже запускать с помощью suprocess (там и timeout есть)
    – Sublimer
    1 июн 2018 в 9:03
  • 1
    да, использование отдельного процесса это тяжёлый, но надёжный способ прервать некооперирующую блокирующую функцию, пример c multiprocessing. Вы посмотрите ещё предоставляет ли pika api что-то более специфичное, менее тяжеловесное.
    – jfs
    1 июн 2018 в 9:08

1 ответ 1

0

В данном случае помог метод add_timeout(deadline, callback_method)

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

def callback_timeout():
    channel.close()
    connection.close()

deadline = 10 # в секундах

add_timeout(deadline, callback_method)
...

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge you have read our privacy policy.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.