[Python] How do i comunicate between Threads Using Queues(janus)?? In Python

I'm building a multi-threaded system where:

1. One thread connects to a WebSocket, processes live kline data, and queues symbol-interval keys for processing.
2. Another thread runs multiple async workers that await those keys from the queue and generate signals.

Each thread has its own asyncio event loop, and I use threads to isolate the heavy workflows.

I'm using janus.Queue to bridge async <-> sync code between threads. But every time I try to await queue.async_.get() in the worker thread (different loop than where it was created), I get:

RuntimeError: <janus.Queue object at 0x000002619212E920> is bound to a different event loop


What I tried:

I read that you cannot create a janus.Queue in one thread and use it in another thread's event loop — which matches what I'm seeing.

So I updated my logic like this:

• In the init() of my activator, I just assign self.closing_klines_queue = None and self.entry_signals_queue = None.
• Then inside each thread’s target function, I create a new event loop, assign it with asyncio.set_event_loop(), and create the queue inside that loop.
• After that, I assign the queue back to the object so that other threads can access it through self.entry_signals_queue, self.closing_klines_queue, etc.

This is the wrapper I use: import janus

class SignalQueues:

def __init__(self):
self._queue = janus.Queue()

@property
def sync(self):
return self._queue.sync_q

@property
def async_(self):
return self._queue.async_q


Still stuck:

Even after creating the queues inside the correct thread/loop, when I pass that reference to another thread and try to await queue.async_.get() in its event loop, I still get the RuntimeError.

I understand that each janus.Queue must stay tied to the loop where it's created, but how do I cleanly share queues between threads that each have their own event loops?

Ask:

Can anyone show a clean and safe pattern to share queues across asyncio threads (each with its own loop), using janus or another method? If janus is not appropriate, what’s the alternative to fan out messages from a WebSocket thread to async worker threads?

I’ll post my root ServiceActivator class if needed, but this is the core issue.

Thanks.

MAIN CODE

class ServiceActivator:

def __init__(self):

#Queues and events for inter-thread communication
self.closing_klines_queue = None
self.entry_signals_queue = None

#This event is set when the WebSocket is ready to start processing signals from the queues
self.websocket_ready_event = AsyncCompatibleEvent()

#These events are set when the WebSocket is shutting down, signaling other components to stop processing and cancel tasks
self.websocket_shutdown_event = AsyncCompatibleEvent()
self.worker_shutdown_event = AsyncCompatibleEvent()
self.entry_consumer_shutdown_event = AsyncCompatibleEvent()
self.order_monitor_shutdown_event = AsyncCompatibleEvent()

# Main Event to set all threads to exit gracefully
self._exit_flag = threading.Event()

# Threads for each component
self._websocket_thread = None
self._worker_threads = None
self._entry_thread = None
self._order_thread = None

def start(self):
"""
Starts all service threads and initializes the event loop for each component.
"""
self._websocket_thread = threading.Thread(target=self._run_websocket_loop)
self._websocket_thread.start()

self._worker_threads = threading.Thread(target=self._run_worker_manager_loop)
self._worker_threads.start()

def _run_websocket_loop(self):
"""
Initializes the 3rd-party WebSocket stream and starts listening for kline data.
"""
from domains.signal_service.runtime.thirdparty_stream_startup import ThirdPartyStreamStartup
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
self.closing_klines_queue = SignalQueues()
ws_stream = ThirdPartyStreamStartup(
self.closing_klines_queue,
self.websocket_ready_event,
self.websocket_shutdown_event,
)
loop.run_until_complete(ws_stream.launch_streaming_app())

def _run_worker_manager_loop(self):
"""
Initializes the worker manager that processes signals from the entry queue.
"""
from domains.signal_service.runtime.worker_manager_startup import WorkerManagerStartup
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
self.entry_signals_queue = SignalQueues()
worker_manager = WorkerManagerStartup(
self.entry_signals_queue,
self.closing_klines_queue,
self.websocket_ready_event,
self.worker_shutdown_event,
)
loop.run_until_complete(worker_manager.launch_worker_app())


THIS IS HOW I RUN IT MAINLY

activator = ServiceActivator()
activator.start()

Continue reading...