LINE聊天Bot开发难题：Python Asyncio Websocket与循环冲突求解

我之前也碰到过这种同步循环和asyncio事件循环打架的问题，本质就是主线程只能被一个循环占用，要么跑你的LINE监听while，要么跑websocket的事件循环，没法同时进行。下面给你两个可行的解决思路，从最推荐的全异步方案到兼容现有代码的线程方案都有：

解决方案一：全异步整合（推荐） #

把你的LINE Bot逻辑全部迁移到asyncio事件循环里，用LINE SDK的异步版本（官方line-bot-sdk已经支持asyncio），这样就能让LINE消息监听和websocket服务在同一个事件循环里并行运行。

步骤示例 #

1. 导入异步模块

import asyncio
from linebot.aio import LineBotApi, WebhookHandler
from linebot.models import MessageEvent, TextMessage, TextSendMessage
import websockets
import json

2. 初始化核心组件

# 替换成你的LINE凭证
LINE_CHANNEL_ACCESS_TOKEN = "YOUR_TOKEN"
LINE_CHANNEL_SECRET = "YOUR_SECRET"

line_bot_api = LineBotApi(LINE_CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(LINE_CHANNEL_SECRET)

# 存储活跃的websocket连接（用集合管理多客户端）
active_websockets = set()

3. 实现websocket处理逻辑
   这个函数负责和浏览器端通信，接收处理后的响应并转发给LINE用户：

async def websocket_client_handler(websocket, path):
    active_websockets.add(websocket)
    try:
        async for raw_msg in websocket:
            msg_data = json.loads(raw_msg)
            # 把浏览器返回的响应发给对应的LINE用户
            await line_bot_api.push_message(
                msg_data["user_id"],
                TextSendMessage(text=msg_data["response"])
            )
    finally:
        # 连接断开后移除集合
        active_websockets.remove(websocket)

4. 异步处理LINE消息
   用装饰器注册消息回调，把用户输入转发给浏览器端：

@handler.add(MessageEvent, message=TextMessage)
async def handle_line_text(event):
    user_id = event.source.user_id
    user_input = event.message.text
    
    # 把用户输入发送给所有连接的浏览器客户端
    if active_websockets:
        forward_msg = json.dumps({
            "user_id": user_id,
            "input": user_input
        })
        # 并行发送给所有活跃连接
        await asyncio.gather(*[ws.send(forward_msg) for ws in active_websockets])

5. 启动事件循环，同时运行两个服务
   如果你的LINE Bot用的是webhook模式（推荐的部署方式），可以和FastAPI/Starlette等异步框架整合；如果是用长轮询拉取消息，就把轮询逻辑改成异步任务：

async def line_polling_task():
    """异步轮询LINE消息的任务（如果不用webhook的话）"""
    while True:
        # 用asyncio.to_thread包装同步的消息拉取方法（如果SDK没有异步版本）
        messages = await asyncio.to_thread(line_bot_api.get_messages)
        for msg in messages:
            # 手动触发消息处理
            await handler.handle(msg)
        await asyncio.sleep(1)  # 控制轮询间隔

async def main():
    # 启动websocket服务
    ws_server = await websockets.serve(
        websocket_client_handler, "localhost", 8765
    )
    # 启动LINE轮询任务（如果用webhook就替换成启动异步web服务）
    polling_task = asyncio.create_task(line_polling_task())
    
    # 同时运行两个任务，直到手动终止
    await asyncio.gather(ws_server.wait_closed(), polling_task)

if __name__ == "__main__":
    asyncio.run(main())

这种方案完全没有线程冲突，代码逻辑更统一，是最推荐的方式。

解决方案二：线程分离（兼容现有同步LINE代码） #

如果不想大规模修改现有的LINE同步代码，可以把websocket的事件循环放到单独的线程里运行，主线程继续跑你的LINE监听while循环。但要注意线程安全和跨线程调用asyncio的规则。

步骤示例 #

1. 启动websocket线程

import threading
import asyncio
import websockets
from linebot import LineBotApi, WebhookHandler
import json
import time

LINE_CHANNEL_ACCESS_TOKEN = "YOUR_TOKEN"
LINE_CHANNEL_SECRET = "YOUR_SECRET"

line_bot_api = LineBotApi(LINE_CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(LINE_CHANNEL_SECRET)

active_websockets = set()
# 线程锁保护websocket集合
ws_lock = threading.Lock()

def run_websocket_server():
    # 为线程创建独立的事件循环
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    async def ws_handler(websocket, path):
        with ws_lock:
            active_websockets.add(websocket)
        try:
            async for raw_msg in websocket:
                msg_data = json.loads(raw_msg)
                # 跨线程调用同步LINE API发消息
                line_bot_api.push_message(
                    msg_data["user_id"],
                    TextSendMessage(text=msg_data["response"])
                )
        finally:
            with ws_lock:
                active_websockets.remove(websocket)

    start_server = websockets.serve(ws_handler, "localhost", 8765)
    loop.run_until_complete(start_server)
    loop.run_forever()

# 启动websocket线程（守护线程，主程序退出时自动结束）
threading.Thread(target=run_websocket_server, daemon=True).start()

2. 修改原有LINE循环，跨线程发送消息到websocket

# 原有LINE监听循环
while True:
    messages = line_bot_api.get_messages()
    for msg in messages:
        if isinstance(msg, TextMessage):
            user_id = msg.source.user_id
            user_input = msg.text

            # 构造要发送给浏览器的消息
            forward_msg = json.dumps({
                "user_id": user_id,
                "input": user_input
            })

            # 跨线程调用asyncio发送消息
            def send_to_ws():
                with ws_lock:
                    if active_websockets:
                        return asyncio.gather(*[ws.send(forward_msg) for ws in active_websockets])
            
            # 获取websocket线程的事件循环，用run_coroutine_threadsafe提交任务
            ws_loop = asyncio.get_event_loop()
            asyncio.run_coroutine_threadsafe(send_to_ws(), ws_loop)
    
    time.sleep(1)

这个方案的好处是不用改原有LINE代码，但要额外处理线程安全问题，调试起来可能比全异步方案麻烦一点。

浏览器端重连优化 #

之前你说readyState判断没成功，建议直接实现自动重连逻辑，不管断开原因都自动重试：

function connectWebSocket() {
    const ws = new WebSocket('ws://localhost:8765');

    ws.onopen = () => console.log('WebSocket连接成功');

    ws.onmessage = (event) => {
        const data = JSON.parse(event.data);
        // 这里处理用户输入，生成响应
        const response = processUserInput(data.input);
        // 把响应发回Python服务
        ws.send(JSON.stringify({
            user_id: data.user_id,
            response: response
        }));
    };

    ws.onclose = () => {
        console.log('连接断开，3秒后重试...');
        setTimeout(connectWebSocket, 3000);
    };

    ws.onerror = (err) => {
        console.error('WebSocket错误:', err);
        ws.close();
    };
}

// 初始化连接
connectWebSocket();

这样不管服务重启还是网络波动，浏览器都会自动尝试重连，稳定性会好很多。

内容的提问来源于stack exchange，提问作者Jayson Okhman