鸿蒙WebSocket实现双向通信连接python编写的本地服务端

CCC～ぜ

300人浏览 · 2025-07-17 20:48:54

CCC～ぜ · 2025-07-17 20:48:54 发布

本实例通过编写一个简单的python服务端，让一个网页和一个鸿蒙App通过WebSocket协议进行双向通信。让大家直观感受WebSocket的作用。

1）我们要编写text.py、client.html文件，和创建一个 WebSocket鸿蒙项目。

2）首先是text.py，可以直接复制粘贴

import asyncio
import websockets
from datetime import datetime

connected_clients = {}  # 存储所有连接的客户端 ( WebSocket对象)
next_client_id = 1  # 用于生成客户端ID的计数器

async def handler(websocket):
    global next_client_id
    client_id = f"用户{next_client_id}"   # 为客户端生成唯一ID
    next_client_id += 1
    # 将客户端WebSocket对象加入连接存储
    connected_clients[client_id] = websocket
    print(f"客户端 {client_id} 已连接")
    
    try:
        # 向客户端发送欢迎消息，告知其ID
        await websocket.send(f"欢迎！你的客户端 ID 是 {client_id}")
        
        # 持续接收客户端发送的消息
        async for message in websocket:
            # 获取当前时间并格式化
            timestamp = datetime.now().strftime("%H:%M:%S")
            # 构造要广播的消息内容，包含发送者ID、时间和消息本身
            broadcast_message = f"[{timestamp}] {client_id} 发送了: {message}"
            
            # 遍历所有已连接客户端，进行消息转发（排除消息发送者自己）
            for receiver_id, receiver_ws in list(connected_clients.items()):
                if receiver_id != client_id:
                    try:
                        await receiver_ws.send(broadcast_message)
                    except Exception as e:
                        print(f"转发给 {receiver_id} 失败: {e}")
    
    except websockets.exceptions.ConnectionClosedOK:
        # 捕获正常的连接关闭异常，可根据需要添加处理逻辑
        pass
    finally:
        # 客户端断开连接时，从存储中移除
        if client_id in connected_clients:
            del connected_clients[client_id]
            print(f"客户端 {client_id} 已断开")

async def main():
    # 启动WebSocket服务器，监听localhost的8765端口
    async with websockets.serve(handler, "localhost", 8765):
        print("服务器已启动: ws://localhost:8765")
        # 保持服务器运行，等待客户端连接和消息交互
        await asyncio.Future()

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

3）接下来我们写client.html（网页）

<!DOCTYPE html>
<html>
<head>
	<meta charset="UTF-8">
	<title>WebSocket Chat</title>
</head>
<body>
	<script type="text/javascript">
		var socket;
		if (!window.WebSocket) {
			window.WebSocket = window.MozWebSocket;
		}
		if (window.WebSocket) {
			socket = new WebSocket("ws://localhost:8765");
			socket.onmessage = function (event) {
				var ta = document.getElementById('responseText');
				ta.value = ta.value + '\n' + "服务端：" + event.data;
			};
			socket.onopen = function (event) {
				var ta = document.getElementById('responseText');
				ta.value = "连接开启!";
			};
			socket.onclose = function (event) {
				var ta = document.getElementById('responseText');
				ta.value = ta.value + "\n连接被关闭";
			};
		} else {
			alert("你的浏览器不支持 WebSocket！");
		}

		function send(message) {
			if (!window.WebSocket) {
				return;
			}
			if (socket.readyState == WebSocket.OPEN) {
				socket.send(message);
			} else {
				alert("连接没有开启.");
			}
		}
	</script>
	<form onsubmit="return false;">
		<h3>WebSocket 网页聊天室：</h3>
		<p>进入后如果没有连接，请确保WebSocket服务已开启，并刷新界面重试</p>
		<textarea id="responseText" style="width: 500px; height: 200px;"></textarea>
		<br>
		<input type="text" name="message" style="width: 300px" value="Hello?">
		<input type="button" value="发送消息" onclick="send(this.form.message.value)">
		<input type="button" onclick="javascript:document.getElementById('responseText').value=''" value="清空聊天记录">
	</form>
	<br>
	<br>
</body>
</html>

4）接下来在鸿蒙App项目中写一个页面(总体代码)

注意，模拟器连接本地服务器地址要用模拟器访问宿主机的回环地址10.0.2.2

import { webSocket } from '@kit.NetworkKit';
import { BusinessError } from '@kit.BasicServicesKit';

@Entry
@Component
struct TalkPage {

  @State msgInput: string = ''  //输入框信息
  @State chatLog: string[] = []  //信息列表

  // 2. 创建 WebSocket 对象
  private ws = webSocket.createWebSocket()

  // 3. 按钮直接连，只处理成功/失败
  connect() {
    this.ws.connect(`ws://10.0.2.2:8765`, (err) => {
      if (err) {
        this.chatLog.push(`websocket连接失败: ${err.message}`)
      } else {
        this.chatLog.push("websocket连接成功！")
      }
    })
  }

  // 4. 最简发消息函数：直接发字符串，失败提示
  send() {
    if (!this.msgInput.trim()) return  // 空内容不发
    this.ws.send(this.msgInput, (err) => {
      if (err) {
        this.chatLog.push(`发送失败: ${err.message}`)
      } else {
        this.chatLog.push(`我: ${this.msgInput}`) // 自己发的消息也显示
        this.msgInput = ''  // 清空输入框
      }
    })
  }

  // 5. 仅保留最核心的消息接收监听
  aboutToAppear() {
    this.ws.on('message', (_, data) => {
      const msg = typeof data === 'string' ? data : data.toString()  // 不管啥格式，转成字符串直接显示
      this.chatLog.push(`服务器: ${msg}`)
    })
    this.ws.on('close', (err: BusinessError, value: webSocket.CloseResult) => {
      this.chatLog.push("连接已断开" + value.code + ", 原因是 " + value.reason);
    });
    this.ws.on('error', (err: BusinessError) => {
      this.chatLog.push("出错了" + JSON.stringify(err));
    });
  }

  build() {
    Column() {
      // 输入框
      TextInput({ placeholder: '发点啥', text: this.msgInput })
        .onChange(v => this.msgInput = v)
        .width('90%')

      // 按钮行
      Row() {
        Button('请求连接')
          .onClick(() => this.connect())
        Button('发送信息')
          .onClick(() => this.send())
        Button("清空信息")
          .onClick(()=>this.chatLog=[])
      }

      // 信息记录展示
      Scroll() {
        Column() {
          ForEach(this.chatLog, (log:string) => {
            Text(log).width('100%').fontSize(20)
            Blank().height(5)
          })
        }
      }
    }
    .width('100%')
    .height('100%')
    .padding(10)
  }
}

5）接下来我们要运行python文件开启服务器。首先你电脑要安装有python运行环境。我们在终端命令行，cd到python文件的根目录。然后安装websockets库，我这里已经安装就不敲回车了。

pip install websockets

6）注意这种安装方法有可能会因为超时报错。如果出错我们可以换国内镜像源加速安装。

pip install websockets -i https://pypi.tuna.tsinghua.edu.cn/simple

7）安装好了以后我们直接运行文件