Есть ли лучший способ обработки сообщений, полученных в очереди от обратного вызова on_message?

Question

У меня в настоящее время есть мой клиентский скрипт paho-MQTT с очередью сообщений и функцией обратного вызова on_message.

    def on_message(self, client, _, message):
        message = message.payload.decode()
        self.messageQueue.put_nowait(message)

мне нужно выполнить некоторую обработку этих полученных сообщений (я получаю около 12 сообщений в очереди каждые 3 секунды)

как мне безопасно обработать эти сообщения?

Answer 1

Если вам нужен надежный механизм распределенной очереди, вы можете реализовать с помощью python сельдерея, где ваш on_message будет действовать как производитель, который ставит задачу на брокера. И несколько рабочих / потребителей будут работать на разных машинах и потреблять задачи от брокера.

Ссылка: http://docs.celeryproject.org/en/latest/getting-started/introduction.html

Надеюсь, это поможет!