4
\$\begingroup\$

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")
\$\endgroup\$
1
\$\begingroup\$

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.

| improve this answer | |
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.