avatar

目录
基于Netty实现服务端与客户端通信

个人博客

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类

java
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)
//KeepAlive
.childOption(ChannelOption.SO_KEEPALIVE, true)
//Handler
.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);
}
// f.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
// workerGroup.shutdownGracefully();
// bossGroup.shutdownGracefully();
}
}

}

NettyServerHandler

在初始化时,需要指定Handle,用来处理Channel相关业务。

java
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

java
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)
// KeepAlive
.option(ChannelOption.SO_KEEPALIVE, true)
// Handler
.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

java
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类,用于测试服务端和客户端是否能正常连接。

java
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方法,输出日志如下:

Code
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方法中解析客户端发来的消息

java
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方法,当收到服务端的消息时,回复服务端

java
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);
}

}

运行后,输出日志如下:

Code
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方法

java
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方法

java
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);

//ByteBuf byteBuf = Unpooled.copiedBuffer("你好,服务端", Charset.forName("utf-8"));
//ctx.writeAndFlush(byteBuf);
}

运行后,输出的部分结果如下:

Code
1
2
3
4
5
6
7
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:基于长度域的拆包器,如果自定义协议中包含长度域的字段,就可以使用这个拆包器

在这里,我们选用分隔符拆包器

首先定义分隔符

java
1
2
3
public class Config {
public static final String DATA_PACK_SEPARATOR = "#$&*";
}

在服务端的channelHandler配置中,需要增加

java
1
2
3
4
5
6
@Override
protected void initChannel(SocketChannel channel) throws Exception {
//这个配置需要在添加Handler前设置
channel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,Unpooled.copiedBuffer(Config.DATA_PACK_SEPARATOR.getBytes())));
channel.pipeline().addLast(new NettyServerHandler());
}

在客户端的channelHandler的配置中,同样也需要增加

java
1
2
3
4
5
6
@Override
protected void initChannel(SocketChannel channel) throws Exception {
//这个配置需要在添加Handler前设置
channel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,Unpooled.copiedBuffer(Config.DATA_PACK_SEPARATOR.getBytes())));
channel.pipeline().addLast(new NettyServerHandler());
}

发送数据时,在数据的末尾增加分隔符:

java
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的配置中,需要增加

java
1
2
3
4
5
@Override
protected void initChannel(SocketChannel channel) throws Exception {
channel.pipeline().addLast(new IdleStateHandler(5, 5, 10));
//...
}

在NettyClientHandler中,重写userEventTriggered方法

java
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;
}
}

当写空闲达到配置的时间时,往服务端发送一个心跳消息

运行后,日志输出如下:

Code
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分别增加以下配置:

java
1
2
3
4
5
6
7
8
@Override
protected void initChannel(SocketChannel channel) throws Exception {
//...
//这个配置需要在添加Handler前设置
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

使用进阶

在上面的基础上,我们来实现一个下面的需求:

  • 客户端需要登录到服务端

  • 客户端登录成功后,服务端可以给客户端发送指令消息,客户端在收到消息及处理完消息后,都需要上报给服务端

封装连接

为便于程序扩展,我们将客户端连接服务端的部分抽取出来。通过一个接口来定义连接的方法,而连接的具体实现由子类来实现。

定义接口

java
1
2
3
4
5
6
7
8
9
10
11
12
public interface IConnection {

/**
* 连接服务器
*
* @param host 服务器地址
* @param port 端口
* @param callback 连接回调
*/
public void connect(String host, int port, IConnectionCallback callback);

}

在这里还需要定义连接的回调接口

java
1
2
3
4
5
6
7
8
public interface IConnectionCallback {

/**
* 连接成功
*/
public void onConnected();

}

具体的连接实现类

java
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();
}
}

}

为便于管理连接,定义一个连接的管理类

java
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);
}

}

调用连接:

java
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

java
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;

}

服务端发送欢迎消息

服务端发送消息

java
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类来统一管理消息分隔符

java
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;
}

/**
* 包装writeAndFlush方法
*
* @param object
*/
public void writeAndFlush(Object object) {
mContext.writeAndFlush(object + Config.DATA_PACK_SEPARATOR);
}

}

再进一步,通过定义MsgUtil类来封装发送欢迎消息

java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public class MsgUtil {

/**
* 发送欢迎消息
*
* @param wrapper
*/
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方法来实现

java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public interface IConnection {

/**
* 连接服务器
*
* @param host 服务器地址
* @param port 端口
* @param callback 连接回调
*/
public void connect(String host, int port, IConnectionCallback callback);

/**
* 注册消息回调
*
* @param callback
*/
public void registerMsgCallback(IMsgCallback callback);

}

在这里,还需要新增IMsgCallback接口

java
1
2
3
4
5
6
7
8
9
10
public interface IMsgCallback {

/**
* 接收到消息时的回调
*
* @param msg
*/
public void onMsgReceived(Msg msg);

}

对应到实现类

java
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

java
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类

java
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();
}

/**
* 将消息加入消息队列
*
* @param msg
*/
public void enqueueMsg(Msg msg) {
mQueue.add(msg);
}

/**
* 从消息队列获取消息
*
* @return
*/
public synchronized Msg next() {
if (using) {
return null;
}
Msg msg = mQueue.poll();
if (msg != null) {
makeUse(true);
}
return msg;
}

/**
* 标记使用状态
*
* @param use
*/
public synchronized void makeUse(boolean use) {
using = use;
}

/**
* 是否能够使用
*
* @return
*/
public synchronized boolean canUse() {
return !using;
}

}

增加消息的分发类MsgDispatcher

java
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,在这里定义了处理的方法,具体实现由子类实现

java
1
2
3
4
5
6
7
8
9
10
public interface IMsgHandler {

/**
* 处理消息
*
* @param msg
*/
public void handle(Msg msg);

}

为统一管理,定义Base类BaseCommandHandler

java
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

java
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);
}

}

对应的心跳类型消息、欢迎类型消息等,都可以新增对应的处理类来实现,这里不再展开。

接收到消息时的处理

java
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,增加登录的请求和响应

java
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;

/**
* 登录请求信息
*
* @author Administrator
*
*/
public static class LoginRuquestInfo {
/**
* 用户名
*/
public String user;

/**
* 密码
*/
public String pwd;

@Override
public String toString() {
return "LoginRuquestInfo [user=" + user + ", pwd=" + pwd + "]";
}
}

/**
* 登录响应信息
*
* @author Administrator
*
*/
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 + "]";
}

}
}

发送登录请求

java
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~

文章作者: milovetingting
文章链接: http://www.milovetingting.cn/2020/04/12/Android/%E5%9F%BA%E4%BA%8ENetty%E5%AE%9E%E7%8E%B0%E6%9C%8D%E5%8A%A1%E5%99%A8%E4%B8%8E%E5%AE%A2%E6%88%B7%E7%AB%AF%E9%80%9A%E4%BF%A1/
版权声明: 本博客所有文章除特别声明外,均采用 CC BY-NC-SA 4.0 许可协议。转载请注明来自 milovetingting

评论