zheng-message 实时通知系统：基于 netty-socketio 与 socket.io 的推送架构实战指南

导读

本文以 zheng 分布式架构中的 zheng-message 实时通知系统模块为对象，围绕其技术选型（后端 netty-socketio、前端 socket.io）展开，结合仓库内真实的服务端 Launcher 示例、SDK 推送工具类与前端 Demo 页面，完整还原一个"实时通知"微服务从服务端事件监听、广播推送、ACK 确认、二进制传输、命名空间隔离、SSL 加密，到业务系统通过 HTTP 接口触发推送的完整链路。读完本文，你将掌握 netty-socketio 的典型配置方式、五种官方演示场景的实现要点，以及 zheng-message 为上层业务（如 CMS、UPMS 等）预留的消息推送调用方式。

一、zheng-message 模块定位与整体结构

在 zheng 这套以 Spring + SpringMVC + Mybatis 构建的分布式敏捷开发架构中，zheng-message 承担的是实时通知能力的公共微服务模块，用于解决"服务器如何主动、实时地把消息推送给浏览器/客户端"这一经典问题，典型场景包括站内信提醒、工单/审批流转通知、实时监控告警等。

从 zheng-message/pom.xml 可以看出，该模块采用 Maven 多模块聚合结构，聚合了三个子模块：

子模块	职责
zheng-message-sdk	推送 SDK，业务方依赖它即可通过 HTTP 触发服务端推送
zheng-message-server	推送服务端，基于 netty-socketio 维护长连接并分发事件
zheng-message-client	前端示例页面，演示浏览器端如何用 socket.io 收发消息

其中 zheng-message-server 的 pom.xml 还声明了对公共模块 zheng-common（版本 1.0.0）的依赖，说明该服务同样继承了 zheng 的公共基类与统一规范；其打包方式为 war，可通过 jetty-maven-plugin 直接以嵌入式容器运行（端口 8881，上下文路径 /），并内置了 dev、test、prod 三个 Maven profile 用于环境切换。

二、技术选型：为什么是 netty-socketio + socket.io

zheng-message/README.md 明确给出了本模块的技术选型：

• 后端：netty-socketio，一个基于 Netty 的 Java 开源实时推送框架，完整实现了 Socket.IO 协议；
• 前端：socket.io，JavaScript 实时应用框架的浏览器端实现。

这套组合的核心价值在于协议双向打通：服务端用 Java 直接实现 Socket.IO 协议（而非自己发明私有协议），因此浏览器端可以直接使用成熟的 socket.io.js 客户端库建立连接、订阅事件、发送事件，无需任何桥接层。同时 netty-socketio 底层基于 Netty 的 NIO 事件驱动模型，天然适合高并发的长连接场景。

从实现细节看，zheng-message-server 并没有直接引用 Maven 中央仓库的 netty-socketio，而是以 system scope + 本地 jar 的方式引入 lib/netty-socketio-1.7.12.zheng.jar（见 zheng-message-server/pom.xml），并在 copy-dependencies 阶段将 system scope 依赖拷贝进 WEB-INF/lib 目录，保证 war 包发布时本地 jar 也能被携带。这种用法说明该项目基于官方 1.7.12 版本做了定制打包，使用时要确保该 jar 随项目保留。

三、服务端核心：五种 Launcher 场景逐一剖析

zheng-message-server 的 Java 源码目录 zheng-message/zheng-message-server/src/main/java/com/zheng/message/server/ 下提供了五套可独立运行的入口类，覆盖了 netty-socketio 的五个核心能力。它们共同构成了本模块的"示例即文档"式说明。

3.1 基础广播：ChatLauncher

ChatLauncher.java 是最简的广播服务器，展示了 netty-socketio 的启动三部曲：

Configuration config = new Configuration();
config.setHostname("localhost");
config.setPort(9092);

final SocketIOServer server = new SocketIOServer(config);
server.addEventListener("chatevent", ChatObject.class, new DataListener<ChatObject>() {
    @Override
    public void onData(SocketIOClient client, ChatObject data, AckRequest ackRequest) {
        // broadcast messages to all clients
        server.getBroadcastOperations().sendEvent("chatevent", data);
    }
});

server.start();
Thread.sleep(Integer.MAX_VALUE);
server.stop();

关键点：

• Configuration 负责全部服务端配置，这里设置了监听主机名 localhost 与端口 9092（与前端 Demo 的连接地址 http://localhost:9092 一一对应）；
• addEventListener("chatevent", ChatObject.class, ...) 注册事件处理器：事件名为 chatevent，消息体反序列化为 ChatObject 对象；
• 处理器内通过 server.getBroadcastOperations().sendEvent(...) 将消息广播给所有已连接客户端，这是"实时通知"最典型的使用方式；
• Thread.sleep(Integer.MAX_VALUE) 让主线程常驻，保持服务运行；server.stop() 在理论上的退出路径执行清理。

事件消息体由 ChatObject.java 定义，仅含 userName 与 message 两个 String 字段，并提供了默认构造器——这是 netty-socketio 通过 Jackson 反序列化消息体的前提。业务系统若要做真正的通知推送，可仿照该类定义自己的消息体 POJO。

3.2 事件确认（ACK）：AckChatLauncher

EventChatLauncher.java 与 ChatLauncher 逻辑相同，仅命名上强调"事件"；而 AckChatLauncher.java 则完整演示了 Socket.IO 的 ACK 回执机制，是保证"通知已被客户端收到"的关键能力：

server.addEventListener("ackevent1", ChatObject.class, new DataListener<ChatObject>() {
    @Override
    public void onData(final SocketIOClient client, ChatObject data, final AckRequest ackRequest) {
        // 1. 若客户端请求了回执，则回复带数据的 ack
        if (ackRequest.isAckRequested()) {
            ackRequest.sendAckData("client message was delivered to server!", "yeah!");
        }
        // 2. 带数据回执的主动推送
        ChatObject ackChatObjectData = new ChatObject(data.getUserName(), "message with ack data");
        client.sendEvent("ackevent2", new AckCallback<String>(String.class) {
            @Override
            public void onSuccess(String result) {
                System.out.println("ack from client: " + client.getSessionId() + " data: " + result);
            }
        }, ackChatObjectData);
        // 3. 无数据（void）回执的主动推送
        ChatObject ackChatObjectData1 = new ChatObject(data.getUserName(), "message with void ack");
        client.sendEvent("ackevent3", new VoidAckCallback() {
            @Override
            protected void onSuccess() {
                System.out.println("void ack from: " + client.getSessionId());
            }
        }, ackChatObjectData1);
    }
});

这段代码演示了三种 ack 用法：

1. 服务端响应客户端的 ack 请求：通过 ackRequest.isAckRequested() 判断客户端在 emit 时是否携带了回调，若有则用 sendAckData(...) 回传数据；
2. 服务端发起带数据回执的推送：client.sendEvent(事件, new AckCallback<T>(期望类型), 数据)，客户端处理完事件后可回调数据给服务端，服务端在 onSuccess(result) 中拿到结果；
3. 服务端发起 void 回执的推送：使用 VoidAckCallback，服务端只关心客户端是否成功处理，回调中可通过 client.getSessionId() 精确识别是哪台客户端完成了确认。

这套机制对"通知已读/已处理"类业务尤为重要——例如工单系统需要确认用户确实收到了推送。

3.3 二进制事件传输：BinaryEventLauncher

BinaryEventLauncher.java 演示了二进制消息的传输（如图片、语音、文件内容）：

Configuration config = new Configuration();
config.setHostname("localhost");
config.setPort(9092);
config.setMaxFramePayloadLength(1024 * 1024);
config.setMaxHttpContentLength(1024 * 1024);

final SocketIOServer server = new SocketIOServer(config);
server.addEventListener("msg", byte[].class, new DataListener<byte[]>() {
    @Override
    public void onData(SocketIOClient client, byte[] data, AckRequest ackRequest) {
        client.sendEvent("msg", data);
    }
});

此处比基础示例多了两个关键配置：

• setMaxFramePayloadLength(1024 * 1024)：限制 WebSocket 单帧负载上限为 1MB；
• setMaxHttpContentLength(1024 * 1024)：限制 HTTP 长轮询方式下单次内容体上限为 1MB。

事件监听类型直接使用 byte[].class，服务端拿到的是原始字节数组，可直接透传回给客户端（如回声测试），也可先解码再做业务处理。这两个配置值的选取要结合推送内容大小调整，避免默认值过小导致大消息被拒。

3.4 命名空间隔离：NamespaceChatLauncher

NamespaceChatLauncher.java 演示了 Socket.IO 的 namespace（命名空间）隔离能力——在同一端口上划分多个互不干扰的"频道"：

final SocketIONamespace chat1namespace = server.addNamespace("/chat1");
chat1namespace.addEventListener("message", ChatObject.class, new DataListener<ChatObject>() {
    @Override
    public void onData(SocketIOClient client, ChatObject data, AckRequest ackRequest) {
        chat1namespace.getBroadcastOperations().sendEvent("message", data);
    }
});

final SocketIONamespace chat2namespace = server.addNamespace("/chat2");
// ...同样注册 "/chat2" 的事件并做命名空间内广播

关键差异在于：

• 事件注册不再挂在 server 上，而是挂在 server.addNamespace("/chat1") 返回的 SocketIONamespace 对象上；
• 广播操作也从 server.getBroadcastOperations() 变为 chat1namespace.getBroadcastOperations()，只把消息发给连接在 /chat1 命名空间下的客户端。

这非常适合"按业务域隔离通知"的场景：例如 /cms 只服务内容通知、/upms 只服务权限变更通知，客户端连接时指定对应命名空间即可避免串扰，服务端也可以为不同命名空间设置不同的事件处理器。

3.5 SSL 加密：SslChatLauncher

SslChatLauncher.java 演示了 TLS/SSL 加密连接的启用方式：

Configuration config = new Configuration();
config.setHostname("localhost");
config.setPort(10443);   // SSL 端口

config.setKeyStorePassword("test1234");
InputStream stream = SslChatLauncher.class.getResourceAsStream("/keystore.jks");
config.setKeyStore(stream);

要点：

• 端口改用 10443，与明文示例的 9092 区分；
• 通过 config.setKeyStore(stream) 加载 JKS 格式密钥库，setKeyStorePassword("test1234") 设置密钥库口令（示例口令，生产环境必须替换为自己的证书口令）；
• 密钥库文件 keystore.jks 位于类路径根目录，通过 getResourceAsStream("/keystore.jks") 读取。

启用 SSL 后，前端 socket.io 客户端需要以 https（或 wss）协议连接，且浏览器会校验证书链，生产环境应使用受信任 CA 签发的证书。

3.6 服务端 web 入口

服务端同时是一个标准 war 应用，提供了 SpringMVC 测试入口 IndexController.java：继承 zheng-common 的 BaseController，通过 @RequestMapping(value = "/index") 暴露 GET /index 接口返回字符串 "index"，用于验证 web 容器与 Spring 环境是否正常启动。

四、推送 SDK：业务系统如何触发实时通知

实时通知系统的另一端是业务系统。zheng-message 提供了独立的推送 SDK 模块 zheng-message-sdk，业务方只需引入该 jar 并调用 MessageHelper.publish(url, publishDto)，即可通过 HTTP 接口把推送请求交给推送服务端，无需关心底层长连接细节。

4.1 统一返回结果：BaseResult

BaseResult.java 定义了统一的返回结构，三个字段：

字段	含义
code	状态码，1 为成功，其他为失败
message	成功时为 "success"，失败时为失败原因
data	数据结果集

4.2 推送参数模型：PublishDto

PublishDto.java 是推送请求的参数载体，其字段与注释直接点明了系统的三种推送类型：

字段	说明
type	推送类型：broadcast（广播）/ topics（按主题）/ clients（点对点）
namespace	推送的目标命名空间
accessKey	鉴权 key，用于校验调用方身份
topics	按主题推送时，目标主题的订阅者列表
clients	点对点推送时，目标客户端标识列表
payload	消息体（Object 类型，可传任意 JSON 对象）

对应的取值常量定义在 MessageConstants.java：

public static final String PUBLISH_TYPE_BROADCAST = "broadcast";  // 广播
public static final String PUBLISH_TYPE_TOPICS   = "topics";      // 主题
public static final String PUBLISH_TYPE_CLIENTS  = "clients";     // 点对点

由此可以推断，zheng-message 的推送服务计划同时支持广播、按主题订阅分发、指定客户端直推三种模式，业务方按需选择。

4.3 推送工具类：MessageHelper

MessageHelper.java 是 SDK 的核心，基于 Apache HttpClient 实现了一个轻量的 HTTP JSON 推送客户端：

public BaseResult publish(String url, PublishDto publishDto) {
    HttpClient httpClient = new DefaultHttpClient();
    httpPost = new HttpPost(url);
    httpPost.setHeader("Content-type", "application/json; charset=utf-8");
    HttpEntity httpEntity = new StringEntity(JSONObject.toJSONString(publishDto), "utf-8");
    httpPost.setEntity(httpEntity);
    HttpResponse httpResponse = httpClient.execute(httpPost);
    // ... 解析响应为 BaseResult
    return new BaseResult(-1, "error", "publish error");  // 异常兜底
}

实现细节：

• 请求体用 fastjson 的 JSONObject.toJSONString(publishDto) 序列化为 JSON，UTF-8 编码，Content-type 显式声明为 application/json；
• 只有 HTTP 200 时才解析响应体为 BaseResult；任何异常都会兜底返回 code=-1、message="publish error" 的失败结果；
• 最后通过 httpPost.releaseConnection() 释放连接。

SDK 自带的 main 方法给出了完整调用示例：

String url = "http://localhost:8881/message/publish";
PublishDto publishDto = new PublishDto();
publishDto.setType(MessageConstants.PUBLISH_TYPE_BROADCAST);
publishDto.setNamespace("/demo");
publishDto.setAccessKey("1");
publishDto.setPayload("hi");
BaseResult result = new MessageHelper().publish(url, publishDto);
System.out.println(result.message);

注意这里的地址是 http://localhost:8881/message/publish，与 server 模块 jetty 插件配置的 8881 端口一致（见 zheng-message-server/pom.xml），说明 SDK 与推送服务通过该 HTTP 通道对接。需要说明的是：从当前仓库源码看，server 端可见的 web 入口仅有 IndexController 的 /index 接口，/message/publish 的具体接收实现未在已列出的源码中，因此上述为 SDK 的既定调用约定，部署对接时需以实际发布的服务端接口为准。

五、前端客户端：socket.io 实时页面示例

zheng-message-client 目录提供了可直接用浏览器打开的 Demo 页面，包括 index.html（基础聊天）、event-index.html（事件）、ack-index.html（ACK 回执）、binary-event-index.html（二进制）、namespace-index.html（命名空间）、ssl-event-index.html（SSL），与上文五类服务端 Launcher 一一对应，构成完整的"客户端-服务端"配对示例。

以 index.html 为例，其核心逻辑展示了 socket.io 客户端的标准用法：

var userName = 'user' + Math.floor((Math.random()*1000)+1);
var socket = io.connect('http://localhost:9092');

socket.on('connect', function() {
    output('<span class="connect-msg">Client has connected to the server!</span>');
});

socket.on('chatevent', function(data) {
    output('<span class="username-msg">' + data.userName + ':</span> ' + data.message);
});

socket.on('disconnect', function() {
    output('<span class="disconnect-msg">The client has disconnected!</span>');
});

function sendMessage() {
    var message = $('#msg').val();
    $('#msg').val('');
    var jsonObject = {userName: userName, message: message};
    socket.emit('chatevent', jsonObject);
}

几个值得注意的实践点：

• 连接地址必须与服务端一致：这里连的是 http://localhost:9092，与 ChatLauncher 的端口配置严格对应；若服务端启用 SSL 则应改用 https://localhost:10443（对应 ssl-event-index.html）；
• 事件名必须与服务端对齐：客户端 socket.emit('chatevent', jsonObject) 与服务端 addEventListener("chatevent", ChatObject.class, ...) 使用同一事件名，且发送的 JSON 结构（userName、message）与服务端 ChatObject 字段一一对应，netty-socketio 会按类字段自动反序列化；
• 收发分离：socket.on('chatevent', ...) 订阅服务端广播回来的事件完成渲染，页面还配套了 moment.js 格式化时间戳显示每条消息的发送时刻。

页面本身引用 js/socket.io/socket.io.js 与 moment.min.js，依赖 bootstrap.css 做基础样式，整体上是一个可直接替换为公司内部"实时通知控制台"的最小可运行原型。

六、运行与联调指南

综合以上源码，zheng-message 的本地联调路径如下：

1. 启动推送服务端：进入 zheng-message/zheng-message-server 模块，运行任一 Launcher（如 ChatLauncher）即启动纯 Socket.IO 服务（端口 9092）；若需要 web 容器与 Spring 环境，则用 jetty-maven-plugin 启动，mvn jetty:run 默认端口 8881、上下文 /，可访问 /index 验证；
2. 打开前端 Demo：直接用浏览器打开 zheng-message/zheng-message-client/index.html，页面连接 localhost:9092 后即可看到连接成功提示，输入消息发送即可在多个页面间实时互收；
3. 通过 SDK 触发推送：业务工程引入 zheng-message-sdk，按 MessageHelper.main 的写法构造 PublishDto 并指定 type/namespace/payload，调用 publish 方法将推送请求发往服务端 HTTP 接口。

环境前提说明：模块的 Maven 编译级别为 Java 1.7（见 zheng-message/pom.xml），netty-socketio 以 system scope 依赖本地 lib/netty-socketio-1.7.12.zheng.jar，因此构建该模块时需保证本地 jar 完整存在；ssl-event-index.html 等 SSL 场景还需准备好 JKS 密钥库并配置服务端证书。

七、总结

zheng-message 以"netty-socketio 服务端 + socket.io 客户端 + HTTP 推送 SDK"三段式架构，为 zheng 微服务体系补齐了实时通知这一公共能力。从仓库源码可以看到，它覆盖了广播、主题、点对点三种推送语义（SDK 常量层面），提供了事件确认、二进制传输、命名空间隔离、SSL 加密四类进阶能力（服务端 Launcher 层面），并配套了对应的前端演示页面。对于希望在 zheng 基础上自建站内信、实时告警、工单流转通知等功能的开发者而言，这套模块既是可运行的示例集合，也是可裁剪复用的推送基础设施。