解决：recv() failed (104: Connection reset by peer) while proxying upgraded connection，问题是连接websocket失败

解决recv() failed (104: Connection reset by peer) while proxying upgraded connection，问题是连接websocket失败

---

这段Python脚本是一个WebSocket服务器代理，

解决 while proxying upgraded connection
1、引入所需的模块：
    requests：用于发送HTTP请求。
	websockets：用于处理WebSocket连接。
	logging：用于记录日志信息。
	json, asyncio, socket：Python标准库中的模块。
	datetime：用于处理日期和时间。
2、定义获取IP地址的函数get_country_from_ip(ip)：通过向百度接口发送请求来获取客户端IP地址对应的地区信息。
3、定义两个异步函数forward_messages和proxy_client：分别用于消息转发和客户端重连机制。
4、定义handle_connection(client, path)函数：用于处理客户端和服务器的连接。在函数内部实现了主要的逻辑，包括IP地址获取、连接主要和备用WebSocket服务器、客户端重连等。
5、配置日志记录器logger：设置日志级别为INFO，指定文件名格式和路径，并创建TimedRotatingFileHandler对象进行日志记录。
6、启动WebSocket服务器并监听本地地址和端口8079：调用websockets.serve()函数启动WebSocket服务，并在事件循环中运行服务器。
1
2
3
4
5
6
7
8
9
10
11
12

我这里是配置了一个是将nginx转成python来做一个中转代理，

服务端重连机制

1、单客服端连接中转服务的时候，中转服务端就直接连接websocket服务端

async def handle_connection(client, path):
    client_ip = client.remote_address[0]
    country = get_country_from_ip(client_ip)
    # 主WebSocket服务器
    main_server = 'ws://localhost:8013/websocket'
    # 备用WebSocket服务器
    backup_server = 'ws://localhost:8014/websocket'
    try:
        client_ip = socket.gethostbyname(client_ip)
        logger.info(f'----------接收到来自{country}的IP地址 {client_ip} 的连接-------时间:{datetime.now()}---')
    except socket.gaierror:
        logger.warning(f'无法解析客户端IP地址: {client_ip}')

    while True:
        try:
            server = await asyncio.wait_for(websockets.connect(main_server), timeout=1)
            logger.info(f'-------已连接到主WebSocket服务器------地区:{country}--IP地址{client_ip}----')
            break
        except (websockets.exceptions.WebSocketException, ConnectionRefusedError, asyncio.TimeoutError):
            logger.warning(f'无法连接到主WebSocket服务器，尝试备用WebSocket服务器...地区:{country}--IP地址{client_ip}----')
            try:
                server = await asyncio.wait_for(websockets.connect(backup_server), timeout=1)
                logger.info(f'-----已连接到备用WebSocket服务器-----地区:{country}--IP地址{client_ip}----')

                break
            except (websockets.exceptions.WebSocketException, ConnectionRefusedError, asyncio.TimeoutError):
                logger.warning(f'无法连接到备用WebSocket服务器，请检查网络连接---地区:{country}--IP地址{client_ip}----')
                await asyncio.sleep(0.001)  # 等待1毫秒后重连
    await proxy_client(client, server, client_ip)
    await server.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

客户端机制

async def proxy_client(client, server, client_ip):
    country = get_country_from_ip(client_ip)
    try:
        await asyncio.gather(
            forward_messages(client, server, client_ip),
            forward_messages(server, client, client_ip)
        )
    except websockets.exceptions.ConnectionClosedError:
        logger.info(f'---客户端正常连接关闭--{client_ip}---，停止接收和转发消息---IP地址:{client_ip}---{country}---时间:{datetime.now()}---')
    except Exception as e:
1
2
3
4
5
6
7
8
9
10

将接收到的数据转发给后端，返回给客户端

async def forward_messages(source, destination, client_ip):
    country = get_country_from_ip(client_ip)
    async for message in source:
        data = json.loads(message)
        userId = data['userId']
        if data['cmd'] == 5013 and data['userId'] == int(userId):
            logger.info(
                f'-------接收到后端发送给-IOS客服端心跳---用户ID:{userId}--IP地址:{client_ip}--地区信息:{country}--时间:{datetime.now()}---------')
        elif data['cmd'] == '5013' and data['userId'] == str(userId):
            logger.info(
                f'-------接收到后端发送给-安卓客服端心跳---用户ID:{userId}--IP地址:{client_ip}--地区信息:{country}--时间:{datetime.now()}---------')
        elif data['cmd'] == 1001 and data['userId'] == int(userId):
            logger.info(f'-------接收到后端发送给-IOS客服端心跳---用户ID:{userId}--IP地址:{client_ip}--时间:{datetime.now()}---------')
        else:
            pass

        await destination.send(message)  # 将消息转发给目标方
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

current_time = datetime.now().strftime("%Y-%m-%d_%H")
# 创建logger对象
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# 打印日志路径
log_out = f'log/game-websocket'

# 创建TimedRotatingFileHandler对象，设置文件名格式为每个小时一个
handler = TimedRotatingFileHandler(log_out, encoding='UTF-8', when='H', interval=1, backupCount=1)
handler.suffix = "-%Y-%m-%d_%H.log"  # 设置文件后缀格式
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
# 关闭处理程序和logger
for h in logger.handlers:
    h.close()
    logger.removeHandler(h)
# 将handler添加到logger对象中
logger.addHandler(handler)

start_server = websockets.serve(handle_connection, '0.0.0.0', 8079)  # 监听本地地址和端口
logger.info(f'中转WebSocket服务器代理已启动......')
# 运行事件循环
asyncio.get_event_loop().run_until_complete(start_server)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24