前言
在移动应用开发中，我们通常扮演'客户端'的角色，去连接远程的 WebSocket 服务。但有时，我们需要在设备本身运行一个微型服务器，例如用于局域网内的设备发现、P2P 文件传输信令，或者在调试模式下作为数据广播源。
shelf_web_socket 是基于 Dart 标准 Web 服务器框架 shelf 的 WebSocket 处理器。它能让你在 Flutter 应用（包括 OpenHarmony）中轻松启动一个能够处理 WebSocket 连接的 HTTP 服务。
一、核心概念
- Shelf: Dart 的 Web 服务器中间件管道框架（类似 Express.js）。
- Handler: 处理请求并返回响应的函数。
- WebSocketChannel: 下层封装，让 WebSocket 操作像 Stream 一样简单。
建立连接 ws://ip:port
协议升级请求 (Upgrade Request)
升级成功 (Success)
数据流/接收端 (Stream/Sink)
WebSocket 客户端\n(App/浏览器)
鸿蒙应用\n(shelf_web_socket)
处理器 (shelf handler)
通信通道 (WebSocketChannel)
二、集成与基础用法
2.1 添加依赖
dependencies:
shelf: ^1.4.0
shelf_web_socket: ^3.0.0
web_socket_channel: ^3.0.0
2.2 启动基础服务
import 'package:shelf/shelf_io.dart' as shelf_io;
import 'package:shelf_web_socket/shelf_web_socket.dart';
import 'package:web_socket_channel/web_socket_channel.dart';
void main() async {
// 定义 WebSocket 处理器
var handler = webSocketHandler((WebSocketChannel webSocket) {
webSocket.stream.listen((message) {
print('收到消息：$message');
webSocket.sink.add('服务端已收到：$message');
});
});
// 启动服务，监听所有 IP (0.0.0.0)
var server = await shelf_io.serve(handler, '0.0.0.0', 8080);
print('WebSocket 服务已启动：ws://${server.address.host}:${server.port}');
}
三、进阶场景与示例
3.1 示例一：广播消息
实现一个聊天室功能，将一个客户端发来的消息广播给所有连接者。
import 'package:shelf_web_socket/shelf_web_socket.dart';
import 'package:web_socket_channel/web_socket_channel.dart';
final List<WebSocketChannel> _clients = [];
var broadcastHandler = webSocketHandler((WebSocketChannel webSocket) {
_clients.add(webSocket);
print('新客户端连接，当前在线：${_clients.length}');
webSocket.stream.listen((message) {
// 广播给其他客户端
for (var client in _clients) {
if (client != webSocket) {
client.sink.add(message);
}
}
}, onDone: () {
_clients.remove(webSocket);
print('客户端断开，剩余：${_clients.length}');
});
});
3.2 示例二：鉴权校验
在建立连接前检查 Header 中的 Token。
import 'package:shelf/shelf.dart';
import 'package:shelf_web_socket/shelf_web_socket.dart';
Handler authMiddleware(Handler innerHandler) {
return (Request request) {
if (request.headers['Authorization'] != 'Bearer my_secret_token') {
return Response.forbidden('未授权访问');
}
return innerHandler(request);
};
}
// 使用 Pipeline 组装
// var handler = Pipeline().addMiddleware(authMiddleware).addHandler(wsHandler);
3.3 示例三：处理二进制数据
WebSocket 不仅能传字符串，还能传字节流（如图片片段）。
import 'dart:typed_data';
var binaryHandler = webSocketHandler((WebSocketChannel webSocket) {
webSocket.stream.listen((message) {
if (message is List<int>) {
print('收到二进制数据，长度：${message.length}');
// 处理二进制逻辑...
} else {
print('收到文本：$message');
}
});
});
四、OpenHarmony 平台适配
4.1 网络权限
作为服务端，你需要监听端口，这同样需要 Internet 权限。
"requestPermissions": [
{"name": "ohos.permission.INTERNET"}
]
4.2 后台运行限制
移动操作系统通常限制应用在后台运行服务。如果你的 WebSocket 服务需要长期运行，建议在前台 Service 中启动，或仅在 App 前台可见时运行。
五、完整实战示例：局域网即时画板服务端
本示例将在 OpenHarmony 设备上启动一个 WebSocket 服务。任何连接到该服务的客户端（可以是另一个 App 或浏览器）发送的坐标点，都会被广播给其他人，实现多端协同绘图。
5.1 示例代码
import 'dart:async';
import 'dart:io';
import 'package:flutter/material.dart';
import 'package:shelf/shelf_io.dart' as shelf_io;
import 'package:shelf_web_socket/shelf_web_socket.dart';
import 'package:web_socket_channel/web_socket_channel.dart';
void main() {
runApp(const MaterialApp(home: ServerPage()));
}
class ServerPage extends StatefulWidget {
const ServerPage({super.key});
@override
State<ServerPage> createState() => _ServerPageState();
}
class _ServerPageState extends State<ServerPage> {
HttpServer? _server;
final List<WebSocketChannel> _sockets = [];
final List<String> _logs = [];
String _ipInfo = '获取中...';
@override
void initState() {
super.initState();
_getIpAddress();
}
Future<void> _getIpAddress() async {
try {
final interfaces = await NetworkInterface.list(type: InternetAddressType.IPv4);
final ip = interfaces.first.addresses.first.address;
setState(() {
_ipInfo = ip;
});
} catch (e) {
setState(() {
_ipInfo = '无法获取 IP';
});
}
}
// 启动服务
Future<void> _startServer() async {
var handler = webSocketHandler((WebSocketChannel webSocket) {
_sockets.add(webSocket);
_log('新连接接入');
webSocket.stream.listen((message) {
_log('广播数据：$message');
// 广播坐标数据
for (var socket in _sockets) {
if (socket != webSocket) {
socket.sink.add(message);
}
}
}, onDone: () {
_sockets.remove(webSocket);
_log('连接断开');
});
});
try {
_server = await shelf_io.serve(handler, InternetAddress.anyIPv4, 8080);
_log('服务启动于 ws://$_ipInfo:8080');
} catch (e) {
_log('启动失败：$e');
}
}
Future<void> _stopServer() async {
await _server?.close(force: true);
for (var s in _sockets) {
await s.sink.close();
}
_sockets.clear();
_server = null;
_log('服务已停止');
}
void _log(String msg) {
if (!mounted) return;
setState(() {
_logs.insert(0, '[${DateTime.now().hour}:${DateTime.now().minute}] $msg');
});
}
@override
void dispose() {
_stopServer();
super.dispose();
}
@override
Widget build(BuildContext context) {
return Scaffold(
appBar: AppBar(title: const Text('WebSocket Server')),
body: Column(
children: [
Container(
padding: const EdgeInsets.all(20),
color: Colors.blue[50],
child: Column(
children: [
Text('本机 IP: $_ipInfo', style: const TextStyle(fontSize: 18, fontWeight: FontWeight.bold)),
const SizedBox(height: 10),
Row(
mainAxisAlignment: MainAxisAlignment.center,
children: [
ElevatedButton(
onPressed: _server == null ? _startServer : null,
child: const Text('启动服务'),
),
const SizedBox(width: 20),
ElevatedButton(
style: ElevatedButton.styleFrom(backgroundColor: Colors.red[100]),
onPressed: _server != null ? _stopServer : null,
child: const Text('停止'),
),
],
),
],
),
),
const Divider(),
Expanded(
child: ListView.builder(
itemCount: _logs.length,
itemBuilder: (context, index) => ListTile(
leading: const Icon(Icons.info_outline, size: 16),
title: Text(_logs[index], style: const TextStyle(fontSize: 14)),
),
),
),
],
),
);
}
}
六、总结
通过 shelf_web_socket，我们让 OpenHarmony 手机不仅仅是信息的消费者，更成为了信息的生产者和中转站。
最佳实践：
- 异常处理：Websocket 连接很容易因网络波动断开，务必处理
onDone和onError。 - 资源释放：在组件销毁或应用退出时，务必关闭 Server 和所有 Channel 连接，防止端口占用。
- 心跳机制：为了保持连接活性，建议在应用层实现 Ping/Pong 心跳包。
