基于netty的websocket服务
2017-08-21 21:59
357 查看
我们下面则将一些实际场景都添加进去，比如用户身份的验证，游客只能浏览不能发言，多房间（频道）的聊天。
该博客非常适合 Java 新手，非常适合作为学习 Java 的切入点，不需要考虑tomcat、spring、mybatis等。
唯一的知识点就是 maven 的基础使用。
├── WebSocketServer.java 启动服务器端口监听
├── WebSocketServerInitializer.java 初始化服务
├── WebSocketServerHandler.java 接管WebSocket数据连接
├── dto
│ └── Response.java 返回给客户端数据对象
├── entity
│ └── Client.java 每个连接到WebSocket服务的客户端对象
└── service
├── MessageService.java 完成发送消息
└── RequestService.java WebSocket初始化连接握手时的数据处理
客户端将用户 id 、进入的房间的 rid、用户 token
id 和 token 的验证（我的做法是 token 存放在redis string 5秒的过期时间）
使用一个Map
ID，然后为该房间 ID 新建一个
这段代码需要理解吗？这是 netty 的套路，可以先记住 netty 的线程模型是一个 react 的一种变型，这里有两个nio线程组，一个是接受客户端的请求，一个是worker组专门处理客户端的请求。
可以简单的理解下面的代码就构建了一个nginx服务器。所以不用管。
package net.mengkang;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
public final class WebSocketServer {
private static final int PORT = 8083;
public static void main(String[] args) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new WebSocketServerInitializer());
Channel ch = b.bind(PORT).sync().channel();
ch.closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
package net.mengkang;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
public class WebSocketServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new HttpServerCodec());
pipeline.addLast(new HttpObjectAggregator(65536));
pipeline.addLast(new WebSocketServerCompressionHandler());
pipeline.addLast(new WebSocketServerHandler());
}
}
下面程序中最的处理在握手阶段
握手完成之后的消息传递则在
整理的执行流程，大家可以对各个方法打断点予以调试，就会很清楚整个执行的脉络啦。
package net.mengkang;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.GlobalEventExecutor;
import net.mengkang.dto.Response;
import net.mengkang.entity.Client;
import net.mengkang.service.MessageService;
import net.mengkang.service.RequestService;
import org.json.JSONObject;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import static io.netty.handler.codec.http.HttpHeaderNames.HOST;
import static io.netty.handler.codec.http.HttpMethod.GET;
import static io.netty.handler.codec.http.HttpResponseStatus.*;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
// websocket 服务的 uri
private static final String WEBSOCKET_PATH = "/websocket";
// 一个 ChannelGroup 代表一个直播频道
private static Map<Integer, ChannelGroup> channelGroupMap = new ConcurrentHashMap <>();
// 本次请求的 code
private static final String HTTP_REQUEST_STRING = "request";
private Client client = null;
private WebSocketServerHandshaker handshaker;
@Override
public void messageReceived(ChannelHandlerContext ctx, Object msg) {
if (msg instanceof FullHttpRequest) {
handleHttpRequest(ctx, (FullHttpRequest) msg);
} else if (msg instanceof WebSocketFrame) {
handleWebSocketFrame(ctx, (WebSocketFrame) msg);
}
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
ctx.flush();
}
private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
// Handle a bad request.
if (!req.decoderResult().isSuccess()) {
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
return;
}
// Allow only GET methods.
if (req.method() != GET) {
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));
return;
}
if ("/favicon.ico".equals(req.uri()) || ("/".equals(req.uri()))) {
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, NOT_FOUND));
return;
}
QueryStringDecoder queryStringDecoder = new QueryStringDecoder(req.uri());
Map<String, List<String>> parameters = queryStringDecoder.parameters();
if (parameters.size() == 0 || !parameters.containsKey(HTTP_REQUEST_STRING)) {
System.err.printf(HTTP_REQUEST_STRING + "参数不可缺省");
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, NOT_FOUND));
return;
}
client = RequestService.clientRegister(parameters.get(HTTP_REQUEST_STRING).get(0));
if (client.getRoomId() == 0) {
System.err.printf("房间号不可缺省");
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, NOT_FOUND));
return;
}
// 房间列表中如果不存在则为该频道,则新增一个频道 ChannelGroup
if (!channelGroupMap.containsKey(client.getRoomId())) {
channelGroupMap.put(client.getRoomId(), new DefaultChannelGroup(GlobalEventExecutor.INSTANCE));
}
// 确定有房间号,才将客户端加入到频道中
channelGroupMap.get(client.getRoomId()).add(ctx.channel());
// Handshake
WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(getWebSocketLocation(req), null, true);
handshaker = wsFactory.newHandshaker(req);
if (handshaker == null) {
WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
} else {
ChannelFuture channelFuture = handshaker.handshake(ctx.channel(), req);
// 握手成功之后,业务逻辑
if (channelFuture.isSuccess()) {
if (client.getId() == 0) {
System.out.println(ctx.channel() + " 游客");
return;
}
}
}
}
private void broadcast(ChannelHandlerContext ctx, WebSocketFrame frame) {
if (client.getId() == 0) {
Response response = new Response(1001, "没登录不能聊天哦");
String msg = new JSONObject(response).toString();
ctx.channel().write(new TextWebSocketFrame(msg));
return;
}
String request = ((TextWebSocketFrame) frame).text();
System.out.println(" 收到 " + ctx.channel() + request);
Response response = MessageService.sendMessage(client, request);
String msg = new JSONObject(response).toString();
if (channelGroupMap.containsKey(client.getRoomId())) {
channelGroupMap.get(client.getRoomId()).writeAndFlush(new TextWebSocketFrame(msg));
}
}
private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
if (frame instanceof CloseWebSocketFrame) {
handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
return;
}
if (frame instanceof PingWebSocketFrame) {
ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
return;
}
if (!(frame instanceof TextWebSocketFrame)) {
throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass().getName()));
}
broadcast(ctx, frame);
}
private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
if (res.status().code() != 200) {
ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
res.content().writeBytes(buf);
buf.release();
HttpHeaderUtil.setContentLength(res, res.content().readableBytes());
}
ChannelFuture f = ctx.channel().writeAndFlush(res);
if (!HttpHeaderUtil.isKeepAlive(req) || res.status().code() != 200) {
f.addListener(ChannelFutureListener.CLOSE);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
Channel incoming = ctx.channel();
System.out.println("收到" + incoming.remoteAddress() + " 握手请求");
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
if (client != null && channelGroupMap.containsKey(client.getRoomId())) {
channelGroupMap.get(client.getRoomId()).remove(ctx.channel());
}
}
private static String getWebSocketLocation(FullHttpRequest req) {
String location = req.headers().get(HOST) + WEBSOCKET_PATH;
return "ws://" + location;
}
}
服务器端就写完啦，还有一些客户端对象的构想验证什么的就不一一细说了，都很简单，都在代码里。下面是客户端。
<html>
<head><title></title></head>
<body>
<script type="text/javascript">
var socket;
if (!window.WebSocket) {
window.WebSocket = window.MozWebSocket;
}
if (window.WebSocket) {
socket = new WebSocket("ws://localhost:8083/websocket/?request=e2lkOjE7cmlkOjI2O3Rva2VuOiI0MzYwNjgxMWM3MzA1Y2NjNmFiYjJiZTExNjU3OWJmZCJ9");
socket.onmessage = function(event) {
console.log(event.data);
};
socket.onopen = function(event) {
console.log("websocket 打开了");
};
socket.onclose = function(event) {
console.log("websocket 关闭了");
};
}
function send(message) {
if (!window.WebSocket) { return; }
if (socket.readyState == WebSocket.OPEN) {
socket.send(message);
} else {
alert("The socket is not open.");
}
}
</script>
<form onsubmit="return false;">
<input type="text" name="message" value="Hello, World!"/>
<input type="button" value="Send Web Socket Data" onclick="send(this.form.message.value)" />
</form>
</body>
</html>
该博客非常适合 Java 新手，非常适合作为学习 Java 的切入点，不需要考虑tomcat、spring、mybatis等。
唯一的知识点就是 maven 的基础使用。
├── WebSocketServer.java 启动服务器端口监听
├── WebSocketServerInitializer.java 初始化服务
├── WebSocketServerHandler.java 接管WebSocket数据连接
├── dto
│ └── Response.java 返回给客户端数据对象
├── entity
│ └── Client.java 每个连接到WebSocket服务的客户端对象
└── service
├── MessageService.java 完成发送消息
└── RequestService.java WebSocket初始化连接握手时的数据处理
功能设计概述
身份认证
客户端将用户 id 、进入的房间的 rid、用户 token json_encode，例如
{id:1;rid:21;token:'43606811c7305ccc6abb2be116579bfd'}。然后在 base64处理，通过参数
request传到服务器，然后在服务器做
id 和 token 的验证（我的做法是 token 存放在redis string 5秒的过期时间）
房间表
使用一个Map channelGroupMap来存放各个房间（频道），以客户端传握手时传过来的 base64 字符串中获取到定义的房间
ID，然后为该房间 ID 新建一个
ChannelGroup（
ChannelGroup方便对该组内的所有客户端广播消息）
在 pom.xml 中引入netty 5
<dependencies> <dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <version>5.0.0.Alpha2</version> </dependency> <dependency> <groupId>com.jcraft</groupId> <artifactId>jzlib</artifactId> <version>1.1.2</version> </dependency> <dependency> <groupId>org.json</groupId> <artifactId>json</artifactId> <version>20141113</version> </dependency> <dependency> <groupId>commons-codec</groupId> <artifactId>commons-codec</artifactId> <version>1.10</version> </dependency> </dependencies>
创建服务器
这段代码需要理解吗？这是 netty 的套路，可以先记住 netty 的线程模型是一个 react 的一种变型，这里有两个nio线程组，一个是接受客户端的请求，一个是worker组专门处理客户端的请求。可以简单的理解下面的代码就构建了一个nginx服务器。所以不用管。
package net.mengkang;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
public final class WebSocketServer {
private static final int PORT = 8083;
public static void main(String[] args) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new WebSocketServerInitializer());
Channel ch = b.bind(PORT).sync().channel();
ch.closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
package net.mengkang;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
public class WebSocketServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new HttpServerCodec());
pipeline.addLast(new HttpObjectAggregator(65536));
pipeline.addLast(new WebSocketServerCompressionHandler());
pipeline.addLast(new WebSocketServerHandler());
}
}
处理长连接
下面程序中最的处理在握手阶段handleHttpRequest，里面处理参数的判断，用户的认证，登录用户表的维护，直播房间表维护。详细的请大家对照代码来浏览。
握手完成之后的消息传递则在
handleWebSocketFrame中处理。
整理的执行流程，大家可以对各个方法打断点予以调试，就会很清楚整个执行的脉络啦。
package net.mengkang;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.GlobalEventExecutor;
import net.mengkang.dto.Response;
import net.mengkang.entity.Client;
import net.mengkang.service.MessageService;
import net.mengkang.service.RequestService;
import org.json.JSONObject;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import static io.netty.handler.codec.http.HttpHeaderNames.HOST;
import static io.netty.handler.codec.http.HttpMethod.GET;
import static io.netty.handler.codec.http.HttpResponseStatus.*;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
// websocket 服务的 uri
private static final String WEBSOCKET_PATH = "/websocket";
// 一个 ChannelGroup 代表一个直播频道
private static Map<Integer, ChannelGroup> channelGroupMap = new ConcurrentHashMap <>();
// 本次请求的 code
private static final String HTTP_REQUEST_STRING = "request";
private Client client = null;
private WebSocketServerHandshaker handshaker;
@Override
public void messageReceived(ChannelHandlerContext ctx, Object msg) {
if (msg instanceof FullHttpRequest) {
handleHttpRequest(ctx, (FullHttpRequest) msg);
} else if (msg instanceof WebSocketFrame) {
handleWebSocketFrame(ctx, (WebSocketFrame) msg);
}
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
ctx.flush();
}
private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
// Handle a bad request.
if (!req.decoderResult().isSuccess()) {
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
return;
}
// Allow only GET methods.
if (req.method() != GET) {
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));
return;
}
if ("/favicon.ico".equals(req.uri()) || ("/".equals(req.uri()))) {
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, NOT_FOUND));
return;
}
QueryStringDecoder queryStringDecoder = new QueryStringDecoder(req.uri());
Map<String, List<String>> parameters = queryStringDecoder.parameters();
if (parameters.size() == 0 || !parameters.containsKey(HTTP_REQUEST_STRING)) {
System.err.printf(HTTP_REQUEST_STRING + "参数不可缺省");
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, NOT_FOUND));
return;
}
client = RequestService.clientRegister(parameters.get(HTTP_REQUEST_STRING).get(0));
if (client.getRoomId() == 0) {
System.err.printf("房间号不可缺省");
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, NOT_FOUND));
return;
}
// 房间列表中如果不存在则为该频道,则新增一个频道 ChannelGroup
if (!channelGroupMap.containsKey(client.getRoomId())) {
channelGroupMap.put(client.getRoomId(), new DefaultChannelGroup(GlobalEventExecutor.INSTANCE));
}
// 确定有房间号,才将客户端加入到频道中
channelGroupMap.get(client.getRoomId()).add(ctx.channel());
// Handshake
WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(getWebSocketLocation(req), null, true);
handshaker = wsFactory.newHandshaker(req);
if (handshaker == null) {
WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
} else {
ChannelFuture channelFuture = handshaker.handshake(ctx.channel(), req);
// 握手成功之后,业务逻辑
if (channelFuture.isSuccess()) {
if (client.getId() == 0) {
System.out.println(ctx.channel() + " 游客");
return;
}
}
}
}
private void broadcast(ChannelHandlerContext ctx, WebSocketFrame frame) {
if (client.getId() == 0) {
Response response = new Response(1001, "没登录不能聊天哦");
String msg = new JSONObject(response).toString();
ctx.channel().write(new TextWebSocketFrame(msg));
return;
}
String request = ((TextWebSocketFrame) frame).text();
System.out.println(" 收到 " + ctx.channel() + request);
Response response = MessageService.sendMessage(client, request);
String msg = new JSONObject(response).toString();
if (channelGroupMap.containsKey(client.getRoomId())) {
channelGroupMap.get(client.getRoomId()).writeAndFlush(new TextWebSocketFrame(msg));
}
}
private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
if (frame instanceof CloseWebSocketFrame) {
handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
return;
}
if (frame instanceof PingWebSocketFrame) {
ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
return;
}
if (!(frame instanceof TextWebSocketFrame)) {
throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass().getName()));
}
broadcast(ctx, frame);
}
private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
if (res.status().code() != 200) {
ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
res.content().writeBytes(buf);
buf.release();
HttpHeaderUtil.setContentLength(res, res.content().readableBytes());
}
ChannelFuture f = ctx.channel().writeAndFlush(res);
if (!HttpHeaderUtil.isKeepAlive(req) || res.status().code() != 200) {
f.addListener(ChannelFutureListener.CLOSE);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
Channel incoming = ctx.channel();
System.out.println("收到" + incoming.remoteAddress() + " 握手请求");
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
if (client != null && channelGroupMap.containsKey(client.getRoomId())) {
channelGroupMap.get(client.getRoomId()).remove(ctx.channel());
}
}
private static String getWebSocketLocation(FullHttpRequest req) {
String location = req.headers().get(HOST) + WEBSOCKET_PATH;
return "ws://" + location;
}
}
服务器端就写完啦，还有一些客户端对象的构想验证什么的就不一一细说了，都很简单，都在代码里。下面是客户端。
<html>
<head><title></title></head>
<body>
<script type="text/javascript">
var socket;
if (!window.WebSocket) {
window.WebSocket = window.MozWebSocket;
}
if (window.WebSocket) {
socket = new WebSocket("ws://localhost:8083/websocket/?request=e2lkOjE7cmlkOjI2O3Rva2VuOiI0MzYwNjgxMWM3MzA1Y2NjNmFiYjJiZTExNjU3OWJmZCJ9");
socket.onmessage = function(event) {
console.log(event.data);
};
socket.onopen = function(event) {
console.log("websocket 打开了");
};
socket.onclose = function(event) {
console.log("websocket 关闭了");
};
}
function send(message) {
if (!window.WebSocket) { return; }
if (socket.readyState == WebSocket.OPEN) {
socket.send(message);
} else {
alert("The socket is not open.");
}
}
</script>
<form onsubmit="return false;">
<input type="text" name="message" value="Hello, World!"/>
<input type="button" value="Send Web Socket Data" onclick="send(this.form.message.value)" />
</form>
</body>
</html>
内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 
相关文章推荐
- RPC , netty ,webSocket基于长链接的描述
- Netty之——基于Netty5.0高级案例NettyWebsocket
- 基于Netty5.0高级案例一之NettyWebsocket
- 基于Netty与RabbitMQ的消息服务
- 基于Netty与RabbitMQ的消息服务
- [置顶] 基于netty实现的远程服务框架
- 基于Netty 的webSocket开发
- 使用JMeter测试基于WebSocket协议的服务
- 基于netty-socketio的web推送服务
- 基于TCP的协议封装及Netty搭建高可用网络通信服务
- 基于Netty4构建HTTP服务----浏览器访问和Netty客户端访问
- Netty-WebSocket长连接推送服务
- Netty学习——基于Netty的WebSocket协议开发
- 基于netty-socketio的web推送服务
- RPC、基于netty的长连接和websocket
- 基于Netty与RabbitMQ的消息服务
- 基于Netty5.0高级案例一之NettyWebsocket
- 基于netty轻量的高性能分布式RPC服务框架forest<下篇>
- 基于netty-socketio的web推送服务