Question: I am trying to implement in-memory queue like Kafka using Python using concepts like re-entrant locks and threads. I am new to Python threading. I have a consumer which will subscribe to the topic and read a message from it. So far it's working fine but I have doubts related to threading. I am trying to make the consumerRunner process single- threaded. When checking the output I am seeing MainThread and Thread-1. If the function is single-threaded, should it not be displaying the message single time by the thread which is executing this function? Message2 and Message4 has been consumed twice. With the single thread they should be consumed only once. I know main Thread is the default Thread but is it correct to have msg printed twice? Sorry if I am asking silly question. Output: (py3_8) ninjakx@Kritis-MacBook-Pro kafka % python queueDemo.py Msg: message1 has been published to topic: topic1 Msg: message2 has been published to topic: topic1 Msg: message3 has been published to topic: topic2 Msg: message4 has been published to topic: topic1 Msg: message1 has been consumed by consumer: consumer1 at offset: 0 with current thread: MainThread Msg: message3 has been consumed by consumer: consumer1 at offset: 0 with current thread: MainThread Msg: message2 has been consumed by consumer: consumer1 at offset: 1 with current thread: Thread-1 Msg: message2 has been consumed by consumer: consumer1 at offset: 1 with current thread: MainThread Msg: message4 has been consumed by consumer: consumer1 at offset: 2 with current thread: Thread-1 Msg: message4 has been consumed by consumer: consumer1 at offset: 2 with current thread: MainThread  ConsumerImpl.py import zope.interface from ..interface.iConsumer import iConsumer from collections import OrderedDict from mediator.QueueMediatorImpl import QueueMediatorImpl import threading from threading import Thread import time @zope.interface.implementer(iConsumer) class ConsumerImpl: # will keep all the topic it has subscribed to and their offset def __init__(self, consumerName:str): self.__consumerName = consumerName self.__topicList = [] self.__topicVsOffset = OrderedDict() self.__queueMediator = QueueMediatorImpl() self.threadInit() def threadInit(self): thread = Thread(target = self._consumerRunner) thread.start() # thread.join() # print("thread finished...exiting") def __getConsumerName(self): return self.__consumerName def __getQueueMediator(self): return self.__queueMediator def __getSubscribedTopics(self)->list: return self.__topicList def __setTopicOffset(self, topicName:str, offset:int)->int: self.__topicVsOffset[topicName] = offset def __getTopicOffset(self, topicName:str)->int: return self.__topicVsOffset[topicName] def __addToTopicList(self, topicName:str)->None: self.__topicList.append(topicName) def _subToTopic(self, topicName:str): self.__addToTopicList(topicName) self.__topicVsOffset[topicName] = 0 def __consumeMsg(self, msg:str, offset:int): print(f"Msg: {msg} has been consumed by consumer: {self.__getConsumerName()} at offset: {offset} with current thread: {threading.current_thread().name}\n") # pull based mechanism # running on single thread def _consumerRunner(self): while(True): for topicName in self.__getSubscribedTopics(): curOffset = self.__getTopicOffset(topicName) qmd = self.__getQueueMediator() msg = qmd._readMsgIfPresent(topicName, curOffset) if msg is not None: self.__consumeMsg(msg._getMessage(), curOffset) curOffset += 1 # update offset self.__setTopicOffset(topicName, curOffset) try: #sleep for 100 milliseconds #thread sleep # "sleep() makes the calling thread sleep until seconds seconds have elapsed or a signal arrives which is not ignored." time.sleep(0.1) except Exception as e: print(f"Error: {e}")  QueueDemo.py from service.QueueServiceImpl import QueueServiceImpl if __name__ == "__main__": queueService = QueueServiceImpl() producer1 = queueService._createProducer("producer1") producer2 = queueService._createProducer("producer2") producer3 = queueService._createProducer("producer3") producer4 = queueService._createProducer("producer4") consumer1 = queueService._createConsumer("consumer1") consumer2 = queueService._createConsumer("consumer2") consumer3 = queueService._createConsumer("consumer3") producer1._publishToTopic("topic1", "message1") producer1._publishToTopic("topic1", "message2") producer2._publishToTopic("topic2", "message3") producer1._publishToTopic("topic1", "message4") consumer1._subToTopic("topic1") consumer1._subToTopic("topic2") consumer1._consumerRunner() 

Answer: When you create a consumer object you are creating a new thread as part of startup: class ConsumerImpl: ... def __init__(self, consumerName:str): ... self.threadInit() def threadInit(self): thread = Thread(target = self._consumerRunner) thread.start() ...  This will execute _consumerRunner on a new thread. This method spins forever with a while True loop. In the main of your application, you are then calling _consumerRunner directly. if __name__ == "__main__": ... consumer1 = queueService._createConsumer("consumer1") ... consumer1._consumerRunner()  So now you have called _consumerRunner twice. Once directly in the main thread, and once when creating the consumer object in the constructor. This is why you're seeing the duplicate lines: Msg: message2 has been consumed by consumer: consumer1 at offset: 1 with current thread: Thread-1 Msg: message2 has been consumed by consumer: consumer1 at offset: 1 with current thread: MainThread  So you should remove the direct call to consumer1._consumerRunner() from the main.  Some style notes:  Python uses snake_case predominantly. You should update your methods accordingly (e.g. _consumer_runner vs _consumerRunner).  You mark many value as private (in Python, the standard convention is to use a leading _ in the name), however, you are accessing these methods and attributes from outside of the class. If you intend to use these methods external of the class which defines them, leave off the leading underscore, as these methods should be listed as your public contract.  I would strongly avoid using dunder __ method and attribute names. Nothing in Python is truly private, and if you follow standard conventions, you shouldn't be using anything marked private normally with a single underscore anyways. The dunder variables also have special logic which mangles the name external to the class, which can cause a lot of not-so-fun bugs. Only use them if you have MASSIVE classes, and need to strongly guarantee that two layers in your class hierarchy do not conflict on naming choices.  The zope.interface package seems like overkill to me. There are many other ways to keep contracts in check, such as normal abstract classes (using the abc class), and by using static analysis tools such as mypy or pyright to ensure objects are used appropriately. This keeps heavy boilerplate libraries out of your code keeping things more terse.  After you create your threads, you never keep track of them. For proper shutdown of the application you should add logic to collect them and ensure they are closed and joined. A ThreadPoolExecutor, or ExitStack could help with that.  Get into the habit of using the logging module. It is a lot more flexible than calls to print and will provide a uniform look to your logging across the application.