个人博客
http://www.milovetingting.cn
基于Netty实现服务端与客户端通信
前言
本文介绍基于Netty实现的服务端与客户端通信的简单使用方法，并在此基础上实现一个简单的服务端-客户端指令通信的Demo。
Netty是什么
Netty是一个NIO客户端-服务器框架，可以快速轻松地开发网络应用程序，例如协议服务器和客户端。它极大地简化了网络编程，例如TCP和UDP套接字服务器的开发。提供一个异步事件驱动的网络应用程序框架和工具，以快速开发可维护的高性能和高可扩展性协议服务器和客户端。
以上内容摘选自https://netty.io/wiki/user-guide-for-4.x.html
Netty具有以下特点：
- 适用于各种传输类型的统一API-阻塞和非阻塞套接字
- 更高的吞吐量，更低的延迟
- 减少资源消耗
- 减少不必要的内存复制
- 完整的SSL / TLS和StartTLS支持
以上内容摘选自https://netty.io/
使用入门
Netty的使用，可以参照Netty的官方文档，这里以4.x为例来演示Netty在服务端和客户端上使用。文档地址：https://netty.io/wiki/user-guide-for-4.x.html
这里用Eclipse来进行开发,服务端和客户端都放在一个工程里。
新建Java工程
服务端
首先需要导入netty的jar包。这里使用netty-all-4.1.48.Final.jar。
NettyServer
新建NettyServer类
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
31
32
33
34
35
36
37
38
39
40
| public class NettyServer {
private int mPort;
public NettyServer(int port) {
this.mPort = port;
}
public void run() {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel channel) throws Exception {
channel.pipeline().addLast(new NettyServerHandler());
}
});
ChannelFuture f = b.bind(mPort).sync();
if (f.isSuccess()) {
LogUtil.log("Server,启动Netty服务端成功，端口号:" + mPort);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
}
}
}
|
NettyServerHandler
在初始化时，需要指定Handle，用来处理Channel相关业务。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| public class NettyServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
LogUtil.log("Server,channelActive");
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
LogUtil.log("Server,接收到客户端发来的消息:" + msg);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
LogUtil.log("Server,exceptionCaught");
cause.printStackTrace();
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
LogUtil.log("Server,channelInactive");
}
}
|
经过上面这些步骤后，服务端最基本的设置就完成了。
客户端
客户端和服务端在初始化时大体是类似的，不过相比服务端要简单一些。
NettyClient
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
31
32
33
34
35
36
37
38
39
40
41
42
43
| public class NettyClient {
private String mHost;
private int mPort;
private NettyClientHandler mClientHandler;
private ChannelFuture mChannelFuture;
public NettyClient(String host, int port) {
this.mHost = host;
this.mPort = port;
}
public void connect() {
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
mClientHandler = new NettyClientHandler();
b.group(workerGroup).channel(NioSocketChannel.class)
.option(ChannelOption.SO_KEEPALIVE, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel channel) throws Exception {
channel.pipeline().addLast(mClientHandler);
}
});
mChannelFuture = b.connect(mHost, mPort).sync();
if (mChannelFuture.isSuccess()) {
LogUtil.log("Client,连接服务端成功");
}
mChannelFuture.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
workerGroup.shutdownGracefully();
}
}
}
|
NettyClientHandler
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| public class NettyClientHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
LogUtil.log("Client,channelActive");
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
LogUtil.log("Client,接收到服务端发来的消息:" + msg);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
LogUtil.log("Client,exceptionCaught");
cause.printStackTrace();
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
LogUtil.log("Client,channelInactive");
}
}
|
到这里，客户端最基本设置就完成了。
连接服务端
新建一个Main类，用于测试服务端和客户端是否能正常连接。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| public class Main {
public static void main(String[] args) {
try {
String host = "127.0.0.1";
int port = 12345;
NettyServer server = new NettyServer(port);
server.run();
Thread.sleep(1000);
NettyClient client = new NettyClient(host, port);
client.connect();
} catch (Exception e) {
e.printStackTrace();
}
}
}
|
运行main方法，输出日志如下：
1
2
3
4
| 2020-4-13 0:11:02--Server,启动Netty服务端成功，端口号:12345
2020-4-13 0:11:03--Client,channelActive
2020-4-13 0:11:03--Client,连接服务端成功
2020-4-13 0:11:03--Server,channelActive
|
可以看到，客户端成功连接上了服务端，服务端和客户端里设置的Handler的channelActive方法都会回调。
服务端与客户端通信
在服务端与客户端连接成功后，我们往往需要在双方间进行通信。这里假定，在连接成功后，服务端给客户端发送一个欢迎信息”你好，客户端”，而客户端在收到服务端的消息后，也给服务端回复一个消息”你好，服务端”。下面来实现具体的功能。
修改服务端NettyServerHandler中的channelActive方法和channelRead方法,在channelActive方法中给客户端发送消息，在channelRead方法中解析客户端发来的消息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| public class NettyServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
LogUtil.log("Server,channelActive");
ByteBuf byteBuf = Unpooled.copiedBuffer("你好，客户端", Charset.forName("utf-8"));
ctx.writeAndFlush(byteBuf);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
byte[] buffer = new byte[buf.readableBytes()];
buf.readBytes(buffer);
String message = new String(buffer, "utf-8");
LogUtil.log("Server,接收到客户端发来的消息:" + message);
}
}
|
修改客户端NettyClientHandler中的channelRead方法，当收到服务端的消息时，回复服务端
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| public class NettyClientHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
byte[] buffer = new byte[buf.readableBytes()];
buf.readBytes(buffer);
String message = new String(buffer,"utf-8");
LogUtil.log("Client,接收到服务端发来的消息:" + message);
ByteBuf byteBuf = Unpooled.copiedBuffer("你好，服务端", Charset.forName("utf-8"));
ctx.writeAndFlush(byteBuf);
}
}
|
运行后，输出日志如下：
1
2
3
4
5
6
| 2020-4-13 0:29:16--Server,启动Netty服务端成功，端口号:12345
2020-4-13 0:29:17--Client,channelActive
2020-4-13 0:29:17--Client,连接服务端成功
2020-4-13 0:29:17--Server,channelActive
2020-4-13 0:29:17--Client,接收到服务端发来的消息:你好，客户端
2020-4-13 0:29:17--Server,接收到客户端发来的消息:你好，服务端
|
可以看到，服务端与客户端已经可以正常通信。
粘包与拆包
在实际的使用场景中，可能会存在短时间内大量数据发送的问题。我们模拟这个场景。在客户端连接上服务端后，服务端给客户端发送100个消息，而为便于分析，客户端在收到服务端消息后，不作回复。
修改服务端中NettyServerHandler的channelActive方法
1
2
3
4
5
6
7
8
| @Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
LogUtil.log("Server,channelActive");
for (int i = 0; i < 100; i++) {
ByteBuf byteBuf = Unpooled.copiedBuffer("你好，客户端", Charset.forName("utf-8"));
ctx.writeAndFlush(byteBuf);
}
}
|
修改客户端中NettyClientHandler的channelRead方法
1
2
3
4
5
6
7
8
9
10
11
| @Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
byte[] buffer = new byte[buf.readableBytes()];
buf.readBytes(buffer);
String message = new String(buffer, "utf-8");
LogUtil.log("Client,接收到服务端发来的消息:" + message);
}
|
运行后，输出的部分结果如下:
1
2
3
4
5
6
7
8
| 2020-4-13 0:35:28--Server,启动Netty服务端成功，端口号:12345
2020-4-13 0:35:29--Client,channelActive
2020-4-13 0:35:29--Client,连接服务端成功
2020-4-13 0:35:29--Server,channelActive
2020-4-13 0:35:29--Client,接收到服务端发来的消息:你好，客户端
2020-4-13 0:35:29--Client,接收到服务端发来的消息:你好，客户端你好，客户端你好，客户端你好，客户端你好，客户端你好，客户端你好，客户端你好，客户端你好，客户端你好，客户端你好，客户端你好，客户端你好，客户端你好，客户端你好，客户端你好，客户端你好，客户端
2020-4-13 0:35:29--Client,接收到服务端发来的消息:你好，客户端
|
可以看到，出现了多条消息”粘”在一起的情况。
什么是粘包与拆包
TCP是个”流”协议，所谓流，就是没有界限的一串数据。TCP底层并不了解上层业务数据的具体含义，它会根据TCP缓冲区的实际情况进行包的划分，所以在业务上认为，一个完整的包可能会被TCP拆分成多个包进行发送，也有可能把多个小的包封装成一个大的数据包发送，这就是所谓的TCP粘包和拆包问题。
以上内容摘选自TCP粘包/拆包与Netty解决方案
解决方案
在没有 Netty 的情况下，用户如果自己需要拆包，基本原理就是不断从 TCP 缓冲区中读取数据，每次读取完都需要判断是否是一个完整的数据包 如果当前读取的数据不足以拼接成一个完整的业务数据包，那就保留该数据，继续从 TCP 缓冲区中读取，直到得到一个完整的数据包。 如果当前读到的数据加上已经读取的数据足够拼接成一个数据包，那就将已经读取的数据拼接上本次读取的数据，构成一个完整的业务数据包传递到业务逻辑，多余的数据仍然保留，以便和下次读到的数据尝试拼接。
以上内容摘选自彻底理解Netty，这一篇文章就够了
而使用Netty，则解决这个问题的方法就简单多了。Netty已经提供了四个拆包器:
- FixedLengthFrameDecoder:固定长度的拆包器,Netty会把固定长度的数据包发送给下一个channelHandler
- LineBasedFrameDecoder:行拆包器,每个数据包以换行符分隔发送
- DelimiterBasedFrameDecoder:分隔符拆包器，可以自定义分隔符，行拆包器是分隔符拆包器的一种特例
- LengthFieldBasedFrameDecoder：基于长度域的拆包器，如果自定义协议中包含长度域的字段，就可以使用这个拆包器
在这里，我们选用分隔符拆包器
首先定义分隔符
1
2
3
| public class Config {
public static final String DATA_PACK_SEPARATOR = "#$&*";
}
|
在服务端的channelHandler配置中，需要增加
1
2
3
4
5
6
| @Override
protected void initChannel(SocketChannel channel) throws Exception {
channel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,Unpooled.copiedBuffer(Config.DATA_PACK_SEPARATOR.getBytes())));
channel.pipeline().addLast(new NettyServerHandler());
}
|
在客户端的channelHandler的配置中，同样也需要增加
1
2
3
4
5
6
| @Override
protected void initChannel(SocketChannel channel) throws Exception {
channel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,Unpooled.copiedBuffer(Config.DATA_PACK_SEPARATOR.getBytes())));
channel.pipeline().addLast(new NettyServerHandler());
}
|
发送数据时，在数据的末尾增加分隔符:
1
2
3
4
5
6
7
8
| @Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
LogUtil.log("Server,channelActive");
for (int i = 0; i < 100; i++) {
ByteBuf byteBuf = Unpooled.copiedBuffer("你好，客户端"+Config.DATA_PACK_SEPARATOR, Charset.forName("utf-8"));
ctx.writeAndFlush(byteBuf);
}
}
|
运行后，可以发现，已经解决”粘包”与”拆包”的问题。
心跳
在网络应用中，为了判断连接是否还存在，一般会通过发送心跳包来检测。在Netty中，配置心跳包的步骤如下
在客户端的channelHandler的配置中，需要增加
1
2
3
4
5
| @Override
protected void initChannel(SocketChannel channel) throws Exception {
channel.pipeline().addLast(new IdleStateHandler(5, 5, 10));
}
|
在NettyClientHandler中，重写userEventTriggered方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| @Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
IdleStateEvent event = (IdleStateEvent) evt;
LogUtil.log("Client,Idle:" + event.state());
switch (event.state()) {
case READER_IDLE:
break;
case WRITER_IDLE:
ByteBuf byteBuf = Unpooled.copiedBuffer("心跳^v^v", Charset.forName("utf-8"));
break;
case ALL_IDLE:
break;
default:
super.userEventTriggered(ctx, evt);
break;
}
}
|
当写空闲达到配置的时间时，往服务端发送一个心跳消息
运行后，日志输出如下:
1
2
3
4
5
6
7
8
9
10
11
| 2020-4-13 1:22:50--Server,启动Netty服务端成功，端口号:12345
2020-4-13 1:22:51--Client,channelActive
2020-4-13 1:22:51--Client,连接服务端成功
2020-4-13 1:22:51--Server,channelActive
2020-4-13 1:22:51--Client,接收到服务端发来的消息:你好，客户端
2020-4-13 1:22:56--Client,Idle:WRITER_IDLE
2020-4-13 1:22:56--Server,接收到客户端发来的消息:心跳^v^
2020-4-13 1:22:56--Client,Idle:READER_IDLE
2020-4-13 1:23:01--Client,Idle:WRITER_IDLE
2020-4-13 1:23:01--Server,接收到客户端发来的消息:心跳^v^
2020-4-13 1:23:01--Client,Idle:READER_IDLE
|
可以看到，心跳包按我们配置的时间正常输出了。
配置编码器与解码器
我们上面在发送数据时，需要通过ByteBuf来转换String，而通过配置编码，解码器，我们就可以直接发送字符串。配置如下：
在服务端与客户端的channelHandler分别增加以下配置：
1
2
3
4
5
6
7
8
| @Override
protected void initChannel(SocketChannel channel) throws Exception {
channel.pipeline().addLast("encoder", new StringEncoder());
channel.pipeline().addLast("decoder", new StringDecoder());
}
|
在发送消息时，则可以直接通过ctx.writeAndFlush("心跳^v^" + Config.DATA_PACK_SEPARATOR)的形式来发送。
源码
到此，最简单的服务端与客户端通信的Demo已经完成。源码地址：https://github.com/milovetingting/Samples/tree/master/NettyDemo
使用进阶
在上面的基础上，我们来实现一个下面的需求：
封装连接
为便于程序扩展，我们将客户端连接服务端的部分抽取出来。通过一个接口来定义连接的方法，而连接的具体实现由子类来实现。
定义接口
1
2
3
4
5
6
7
8
9
10
11
12
| public interface IConnection {
public void connect(String host, int port, IConnectionCallback callback);
}
|
在这里还需要定义连接的回调接口
1
2
3
4
5
6
7
8
| public interface IConnectionCallback {
public void onConnected();
}
|
具体的连接实现类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| public class NettyConnection implements IConnection {
private NettyClient mClient;
@Override
public void connect(String host, int port, IConnectionCallback callback) {
if (mClient == null) {
mClient = new NettyClient(host, port);
mClient.setConnectionCallBack(callback);
mClient.connect();
}
}
}
|
为便于管理连接，定义一个连接的管理类
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
31
32
33
| public class ConnectionManager implements IConnection {
private static IConnection mConnection;
private ConnectionManager() {
}
static class ConnectionManagerInner {
private static ConnectionManager INSTANCE = new ConnectionManager();
}
public static ConnectionManager getInstance() {
return ConnectionManagerInner.INSTANCE;
}
public static void initConnection(IConnection connection) {
mConnection = connection;
}
private void checkInit() {
if (mConnection == null) {
throw new IllegalAccessError("please invoke initConnection first!");
}
}
@Override
public void connect(String host, int port, IConnectionCallback callback) {
checkInit();
mConnection.connect(host, port, callback);
}
}
|
调用连接：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| public class Main {
public static void main(String[] args) {
try {
String host = "127.0.0.1";
int port = 12345;
NettyServer server = new NettyServer(port);
server.run();
Thread.sleep(1000);
ConnectionManager.initConnection(new NettyConnection());
ConnectionManager.getInstance().connect(host, port, new IConnectionCallback() {
@Override
public void onConnected() {
LogUtil.log("Main,onConnected"););
}
});
} catch (Exception e) {
e.printStackTrace();
}
}
}
|
在调用connect方法前，需要先调用initConnection来指定具体的连接类
消息Bean的定义
在连接成功后，服务端会给客户端发送一个欢迎的消息。为便于管理，我们定义一个消息Bean
1
2
3
4
5
6
7
8
9
10
11
12
| public class Msg {
public static final int TYPE_WELCOME = 0;
public int type;
public String msg;
}
|
服务端发送欢迎消息
服务端发送消息
1
2
3
4
5
6
7
8
9
10
11
| public class NettyServerHandler extends ChannelInboundHandlerAdapter {
private ChannelHandlerContextWrapper mChannelHandlerContextWrapper;
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
LogUtil.log("Server,channelActive");
mChannelHandlerContextWrapper = new ChannelHandlerContextWrapper(ctx);
MsgUtil.sendWelcomeMsg(mChannelHandlerContextWrapper);
}
}
|
在这里，通过定义一个ChannelHandlerContextWrapper类来统一管理消息分隔符
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| public class ChannelHandlerContextWrapper {
private ChannelHandlerContext mContext;
public ChannelHandlerContextWrapper(ChannelHandlerContext context) {
this.mContext = context;
}
public void writeAndFlush(Object object) {
mContext.writeAndFlush(object + Config.DATA_PACK_SEPARATOR);
}
}
|
再进一步，通过定义MsgUtil类来封装发送欢迎消息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| public class MsgUtil {
public static void sendWelcomeMsg(ChannelHandlerContextWrapper wrapper) {
Msg msg = new Msg();
msg.type = Msg.TYPE_WELCOME;
msg.msg = "你好，客户端";
wrapper.writeAndFlush(Global.sGson.toJson(msg));
}
}
|
客户端消息接收
对于客户端而言，为方便处理消息，我们需要定义一个方法来接收消息。通过在IConnection接口中新增一个registerMsgCallback方法来实现
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| public interface IConnection {
public void connect(String host, int port, IConnectionCallback callback);
public void registerMsgCallback(IMsgCallback callback);
}
|
在这里，还需要新增IMsgCallback接口
1
2
3
4
5
6
7
8
9
10
| public interface IMsgCallback {
public void onMsgReceived(Msg msg);
}
|
对应到实现类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| public class NettyConnection implements IConnection {
private NettyClient mClient;
@Override
public void connect(String host, int port, IConnectionCallback callback) {
if (mClient == null) {
mClient = new NettyClient(host, port);
mClient.setConnectionCallBack(callback);
mClient.connect();
}
}
@Override
public void registerMsgCallback(IMsgCallback callback) {
if (mClient == null) {
throw new IllegalAccessError("please invoke connect first!");
}
mClient.registerMsgCallback(callback);
}
}
|
消息的分发
在客户端，为便于处理消息，我们对消息类型进行划分
修改消息Bean
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
| public class Msg {
public static final int TYPE_WELCOME = 0;
public static final int TYPE_HEART_BEAT = 1;
public static final int TYPE_LOGIN = 2;
public static final int TYPE_COMMAND_A = 3;
public static final int TYPE_COMMAND_B = 4;
public static final int TYPE_COMMAND_C = 5;
public int type;
public String msg;
}
|
假定消息是串行的，需要一个一个地处理。为便于管理消息，增加MsgQueue类
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
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
| public class MsgQueue {
private PriorityBlockingQueue<Msg> mQueue;
private boolean using;
private MsgQueue() {
mQueue = new PriorityBlockingQueue<>(128, new Comparator<Msg>() {
@Override
public int compare(Msg msg1, Msg msg2) {
int res = msg2.priority - msg1.priority;
if (res == 0 && msg1.time != msg2.time) {
return (int) (msg2.time - msg1.time);
}
return res;
}
});
}
public static MsgQueue getInstance() {
return MsgQueueInner.INSTANCE;
}
private static class MsgQueueInner {
private static final MsgQueue INSTANCE = new MsgQueue();
}
public void enqueueMsg(Msg msg) {
mQueue.add(msg);
}
public synchronized Msg next() {
if (using) {
return null;
}
Msg msg = mQueue.poll();
if (msg != null) {
makeUse(true);
}
return msg;
}
public synchronized void makeUse(boolean use) {
using = use;
}
public synchronized boolean canUse() {
return !using;
}
}
|
增加消息的分发类MsgDispatcher
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
31
32
33
34
35
36
37
38
| public class MsgDispatcher {
private static Map<Integer, Class<? extends IMsgHandler>> mHandlerMap;
static {
mHandlerMap = new HashMap<>();
mHandlerMap.put(Msg.TYPE_WELCOME, WelcomeMsgHandler.class);
mHandlerMap.put(Msg.TYPE_HEART_BEAT, HeartBeatMsgHandler.class);
mHandlerMap.put(Msg.TYPE_LOGIN, HeartBeatMsgHandler.class);
mHandlerMap.put(Msg.TYPE_COMMAND_A, CommandAMsgHandler.class);
mHandlerMap.put(Msg.TYPE_COMMAND_B, CommandBMsgHandler.class);
mHandlerMap.put(Msg.TYPE_COMMAND_C, CommandCMsgHandler.class);
}
public static void dispatch() {
if (MsgQueue.getInstance().canUse()) {
Msg msg = MsgQueue.getInstance().next();
if (msg == null) {
return;
}
dispatch(msg);
}
}
public static void dispatch(Msg msg) {
try {
IMsgHandler handler = (IMsgHandler) Class.forName(mHandlerMap.get(msg.type).getName()).newInstance();
handler.handle(msg);
} catch (InstantiationException e) {
e.printStackTrace();
} catch (IllegalAccessException e) {
e.printStackTrace();
} catch (ClassNotFoundException e) {
e.printStackTrace();
}
}
}
|
消息的处理
定义IMsgHandler,在这里定义了处理的方法，具体实现由子类实现
1
2
3
4
5
6
7
8
9
10
| public interface IMsgHandler {
public void handle(Msg msg);
}
|
为统一管理，定义Base类BaseCommandHandler
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| public abstract class BaseCommandHandler implements IMsgHandler {
@Override
public void handle(Msg msg) {
execute(msg);
}
public final void execute(Msg msg) {
LogUtil.log("Client,received command:" + msg);
doHandle(msg);
MsgQueue.getInstance().makeUse(false);
LogUtil.log("Client,report command:" + msg);
MsgDispatcher.dispatch();
}
public abstract void doHandle(Msg msg);
}
|
在BaseCommandHandler中，定义execute方法，顺序调用：上报消息已接收成功、处理消息、上报消息已处理完成。这里的消息上报部分，都只是输出一个日志来代替，在实际的业务中，可以抽取出一个抽象方法，让子类来实现。
定义子类，继承自BaseCommandHandler
1
2
3
4
5
6
7
8
| public class LoginMsgHandler extends BaseCommandHandler {
@Override
public void doHandle(Msg msg) {
LogUtil.log("Client,handle msg:" + msg);
}
}
|
对应的心跳类型消息、欢迎类型消息等，都可以新增对应的处理类来实现，这里不再展开。
接收到消息时的处理
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
31
32
33
| public class Main {
public static void main(String[] args) {
try {
String host = "127.0.0.1";
int port = 12345;
NettyServer server = new NettyServer(port);
server.run();
Thread.sleep(1000);
ConnectionManager.initConnection(new NettyConnection());
ConnectionManager.getInstance().connect(host, port, new IConnectionCallback() {
@Override
public void onConnected() {
LogUtil.log("Main,onConnected");
ConnectionManager.getInstance().registerMsgCallback(new IMsgCallback() {
@Override
public void onMsgReceived(Msg msg) {
MsgQueue.getInstance().enqueueMsg(msg);
MsgDispatcher.dispatch();
}
});
}
});
} catch (Exception e) {
e.printStackTrace();
}
}
}
|
客户端登录
修改消息Bean，增加登录的请求和响应
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
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
| public class Msg {
public static final int TYPE_WELCOME = 0;
public static final int TYPE_HEART_BEAT = 1;
public static final int TYPE_LOGIN = 2;
public static final int TYPE_COMMAND_A = 3;
public static final int TYPE_COMMAND_B = 4;
public static final int TYPE_COMMAND_C = 5;
public int type;
public String msg;
public int priority;
public long time;
public static class LoginRuquestInfo {
public String user;
public String pwd;
@Override
public String toString() {
return "LoginRuquestInfo [user=" + user + ", pwd=" + pwd + "]";
}
}
public static class LoginResponseInfo {
public static final int CODE_SUCCESS = 0;
public static final int CODE_FAILED = 100;
public int code;
public String data;
public static class ResponseData {
public String token;
}
@Override
public String toString() {
return "LoginResponseInfo [code=" + code + ", data=" + data + "]";
}
}
}
|
发送登录请求
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
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
| public class Main {
public static void main(String[] args) {
try {
String host = "127.0.0.1";
int port = 12345;
NettyServer server = new NettyServer(port);
server.run();
Thread.sleep(1000);
ConnectionManager.initConnection(new NettyConnection());
ConnectionManager.getInstance().connect(host, port, new IConnectionCallback() {
@Override
public void onConnected() {
LogUtil.log("Main,onConnected");
ConnectionManager.getInstance().registerMsgCallback(new IMsgCallback() {
@Override
public void onMsgReceived(Msg msg) {
MsgQueue.getInstance().enqueueMsg(msg);
MsgDispatcher.dispatch();
}
});
Msg msg = new Msg();
msg.type = Msg.TYPE_LOGIN;
Msg.LoginRuquestInfo request = new LoginRuquestInfo();
request.user = "wangyz";
request.pwd = "wangyz";
Gson gson = new Gson();
msg.msg = gson.toJson(request);
ConnectionManager.getInstance().sendMsg(msg);
}
});
} catch (Exception e) {
e.printStackTrace();
}
}
}
|
这里，引入Gson，将消息Bean转成json字符串后发送。
对应到服务端，为便于解析出消息，也需要对应的修改消息的Bean。服务端对消息的具体分发与处理，和客户端类似，这里不再展开。
源码
由于篇幅限制，Demo中指令的优先级处理，模拟服务端指令下发等，这里没有再进一步详细介绍，具体可以参考源码:https://github.com/milovetingting/Samples/tree/master/Netty
后记
本文介绍了基于Netty实现服务端与客户端通信的基本用法，以及在此基础上，实现处理服务端指令并上报。Demo中通信的数据格式，用到了json，而优化的做法，可以用protobuf来实现，这里只展示通信的流程及简单的封装，因而未使用protobuf。Demo中只实现大体的流程，可能存在未测试到的Bug，权当一个参考的思路吧。
End~
