One consumer and two producers using threads I have based this code on Producer-Consumer Threading on Real Python.

The idea is that I have 2 producers:


One producer  will give result every 0.4 to 0.7 seconds, depending on REST call.
The second producer will be a web socket that will forward filtered information to consumer.


And one consumer that needs to act on results from producers.

This code is working fine, but I'd like a second opinion on what I can improve from design/architecture side.

import concurrent.futures
import logging
import queue
import random
import threading
import time


def producer_1(queue_1, event_1):
    """Pretend we're getting a number from the network."""
    while True:
        time.sleep(1)
        message = random.randint(1, 101)
        logging.info("Producer_1 got message: %s", message)
        queue_1.put(message)
        event_1.set()

    logging.info("Producer_1 Exiting")


def producer_2(queue_2, event_2):
    """Pretend we're getting a number from the network."""
    while True:
        time.sleep(0.5)
        message = random.randint(1, 101)
        logging.info("Producer_2 got message: %s", message)
        queue_2.put(message)
        event_2.set()

    logging.info("Producer_2 Exiting")


def consumer(queue_1, queue_2, event_1):
    """Pretend we're saving a number in the database."""

    while True:
        event_1.wait()
        logging.info("Consumer START")

        if not queue_1.empty():
            message = queue_1.get()
            logging.info("Consumer storing message_1: %s (size=%d)", message, queue_1.qsize())

            if queue_1.empty():
                event_1.clear()
                logging.info("Queue_1 EMPTY")

        if not queue_2.empty():
            message = queue_2.get()
            logging.info("Consumer storing message_2: %s (size=%d)", message, queue_2.qsize())

            if queue_2.empty():
                event_1.clear()
                logging.info("Queue_2 EMPTY")

        logging.info("Consumer END")


    logging.info("Consumer Exiting")


if __name__ == "__main__":
    format = "%(asctime)s: %(message)s"
    logging.basicConfig(format=format, level=logging.INFO,
                        datefmt="%H:%M:%S")

    pipeline_1 = queue.Queue(maxsize=10)
    event_1 = threading.Event()

    pipeline_2 = queue.Queue(maxsize=10)

    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        #executor.submit(consumer, pipeline_1, event_1, pipeline_2, event_2)
        executor.submit(consumer, pipeline_1, pipeline_2, event_1)
        executor.submit(producer_1, pipeline_1, event_1)
        executor.submit(producer_2, pipeline_2, event_1)

        logging.info("Main: about to set event")
Logging after an infinite loop

The only way that this:

while True:
    time.sleep(1)
    message = random.randint(1, 101)
    logging.info("Producer_1 got message: %s", message)
    queue_1.put(message)
    event_1.set()


is going to terminate is on an exception. That in itself isn't a bad thing - stop iteration exceptions are a common pattern in Python. However, it means that this line:

logging.info("Producer_1 Exiting")


can by definition never execute. You might want to try:

try:
    while True:
        time.sleep(1)
        message = random.randint(1, 101)
        logging.info("Producer_1 got message: %s", message)
        queue_1.put(message)
        event_1.set()
finally:
    logging.info("Producer_1 Exiting")


main method

Putting code in this block:

if __name__ == "__main__":


doesn't remove it from global scope. For that, you need to add a main function.