使用线程实现Tornado WebSockets非阻塞请求，我的实现是否正确？

Great job getting this working—your approach to handling long-running tasks without blocking other WebSocket connections is on the right path with Tornado! Let’s confirm what’s working, and go over a few small improvements to make your code even more robust.

What’s Working Perfectly #

Your core logic aligns with Tornado’s best practices for non-blocking I/O:

• ThreadPoolExecutor for blocking tasks: You’re offloading the time.sleep(1) (your stand-in for a long calculation) to a thread pool. This keeps Tornado’s main IOLoop free to handle other client connections and incoming messages—exactly what you need for concurrent requests.
• spawn_callback for async coroutines: Using tornado.ioloop.IOLoop.current().spawn_callback() in the open method lets your long-running coroutine run in the background without blocking the connection setup or other operations.
• Non-blocking behavior confirmed: Your frontend logs prove this works! When you send a new message, the server immediately sends back the uppercase response while continuing to stream the TICK values. No blocking—perfect.

Small Tweaks for Better Robustness #

While your code runs well, here are a few adjustments to make it cleaner and more reliable:

1. Fix the Write Message Format #

In on_message_coroutine, this line has a syntax issue:

self.write_message('Message:', message)

write_message expects a single argument. Combine the string and message like this:

self.write_message(f"Message: {message}")

2. Adjust Thread Pool Size #

Your thread pool has a fixed size of 2. If you expect multiple clients running long tasks at the same time, increase max_workers to match your expected concurrency (e.g., ThreadPoolExecutor(max_workers=4)). This prevents tasks from waiting unnecessarily in the queue.

3. Add Error Handling #

Right now, if an error occurs in worker_A, it won’t be caught and could silently break your coroutine. Wrap the blocking task in a try-except block:

def worker_A(websocket, x):
    try:
        time.sleep(1)
        print('TICK', x)
        return x
    except Exception as e:
        print(f"Error in worker_A: {e}")
        return None  # Or handle the error appropriately

4. Clean Up Tasks on Connection Close #

If a client disconnects mid-task, your thread pool will still finish the remaining worker_A calls. Add a flag to stop tasks when the connection closes:

class WebSocketHandler(tornado.websocket.WebSocketHandler):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.connected = True

    def on_close(self):
        self.connected = False
        print("Connection closed")

    # Update worker_A to check the flag
    def worker_A(websocket, x):
        if not websocket.connected:
            return None
        time.sleep(1)
        print('TICK', x)
        return x

5. Remove Unnecessary Code #

The pass in worker_A is redundant—you can safely delete it.

Final Verdict #

Your core implementation is correct for maintaining a persistent, non-blocking WebSocket connection that handles concurrent requests. The tweaks above will just make your code more resilient and easier to maintain as you scale.

内容的提问来源于stack exchange，提问作者Everything's Shiny