欢迎阅读本篇文章
使用Netty实现客户端和服务端之间的双向通信提示：本文只是提供部分核心代码，源码详见代码示例
- 前言
- 一、服务端
- 二、客户端
前言
在上个月的开发计划中，有一个系统控制喇叭播放的功能。当时就想到了使用netty进行通信操作。于是在调研途中，发现网上写的都是简单案例，不适用于当前的复杂通信模式。比如：超时断线，断线重连，通信监听，错误记录存储，以及断线和上线后的钉钉通知等等。所以自己从头到尾重新写了一个完整项目，供大家参考。
提示：以下是本篇文章正文内容，下面案例可供参考
一、服务端在netty的通讯服务中，需先启动服务端，提供自身的端口和IP供客户端连接。
代码如下（示例）：
@Slf4j
public class NettyServer {
public Result bind(int port) throws Exception {
try {
EventLoopGroup bossGroup = new NioEventLoopGroup(); //bossGroup就是parentGroup，是负责处理TCP/IP连接的
EventLoopGroup workerGroup = new NioEventLoopGroup(); //workerGroup就是childGroup,是负责处理Channel(通道)的I/O事件
ServerBootstrap sb = new ServerBootstrap();
sb.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 128) //初始化服务端可连接队列,指定了队列的大小128
.childOption(ChannelOption.SO_KEEPALIVE, true) //保持长连接
.childHandler(new ChannelInitializer() { // 绑定客户端连接时候触发操作
@Override
protected void initChannel(SocketChannel sh) throws Exception {
sh.pipeline()
.addLast(new RpcDecoder(RpcRequest.class)) //解码request
.addLast(new RpcEncoder(RpcResponse.class)) //编码response
//readerIdleTime：读超时时间； writerIdleTime：写超时时间； allIdleTime：所有类型超时时间；
//.addLast(new IdleStateHandler(10, 10, 10, TimeUnit.SECONDS))
.addLast(new ServerHandler()); //使用ServerHandler类来处理接收到的消息
}
});
//绑定监听端口，调用sync同步阻塞方法等待绑定操作完成，完成后返回ChannelFuture类似于JDK中Future
ChannelFuture future = sb.bind(port).sync();
log.info("服务端绑定端口:"+port+"成功！");
if (future.isSuccess()) {
log.info("服务端启动成功");
List channelFuture = ServerHandler.getChannelFuture();
channelFuture.add(future);
List eventLoopGroup = ServerHandler.getEventLoopGroup();
eventLoopGroup.add(workerGroup);
eventLoopGroup.add(bossGroup);
} else {
log.error("服务端启动失败");
future.cause().printStackTrace();
bossGroup.shutdownGracefully(); //关闭线程组
workerGroup.shutdownGracefully();
return Result.ERROR_INFO("Netty服务端启动失败失败！");
}
//成功绑定到端口之后,给channel增加一个 管道关闭的监听器并同步阻塞,直到channel关闭,线程才会往下执行,结束进程。
future.channel().closeFuture().sync();
}catch (Exception e){
e.printStackTrace();
}
return Result.ERROR_INFO("Netty服务端发送消息成功！");
}
}
在上面的代码示例中，最重要的就是 ServerHandler 方法。继承ChannelInboundHandlerAdapter。实现 channelActive（有客户端连接服务器会触发此函数） channelInactive（有客户端终止连接服务器会触发此函数）等一些方法。一些的读操作时捕获到异常就可以调用 exceptionCaught（）。更具体详细内容可以自行查阅。但是我的代码示例中已经基本包含了所需用到的方法，大家可以下载后可以看一看。
特别提醒：
在上面的服务端代码中，使用了 ChannelFuture future = sb.bind(port).sync()【绑定监听端口，调用sync同步阻塞方法等待绑定操作完成，完成后返回ChannelFuture类似于JDK中Future】。所以项目如果使用了启动就开启服务端，那么一定要加线程池进行操作，要不然项目就会卡在不动无法完成启动。
log.info("开始启动预加载服务");
if (contextRefreshedEvent.getApplicationContext().getParent() == null) {
try {
log.info("项目启动，开启Netty服务！");
new Thread(() -> {
try {
new NettyServer().bind(configProperties.getNettyServerPort());
} catch (Exception e) {
e.printStackTrace();
}
}).start();
} catch (Exception e) {
log.error("项目启动时，Netty服务开启失败！");
e.printStackTrace();
}
}
二、客户端
启动好服务端以后，开始启动客户端
代码如下（示例）：
@Slf4j
public class NettyClient {
private final String host;
private final int port;
//连接服务端的端口号地址和端口号
public NettyClient(String host, int port) {
this.host = host;
this.port = port;
}
private volatile Channel clientChannel;
Bootstrap bootstrap;
EventLoopGroup group;
public void start() {
group = new NioEventLoopGroup();
//创建客户端启动对象，注意客户端使用的不是ServerBootStrap而是BootStrap
bootstrap = new Bootstrap();
ChannelFuture future; //ChannelFuture的作用是用来保存Channel异步操作的结果
try {
bootstrap.group(group).channel(NioSocketChannel.class) // 使用NioSocketChannel来作为连接用的channel类
.handler(new ChannelInitializer() { // 绑定连接初始化器
@Override
public void initChannel(SocketChannel ch) throws Exception {
System.out.println("正在连接服务端中...");
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new RpcEncoder(RpcRequest.class)); //编码request
pipeline.addLast(new RpcDecoder(RpcResponse.class)); //解码response
// 3s 内如果没有向服务器写数据，会触发一个 IdleState#WRITER_IDLE 事件 用来判断是不是 读空闲时间过长，或 写空闲时间过长
//readerIdleTime：读超时时间； writerIdleTime：写超时时间； allIdleTime：所有类型超时时间；
//pipeline.addLast(new IdleStateHandler(10, 10, 10, TimeUnit.SECONDS));
pipeline.addLast(new ClientHandler(new NettyClient(host, port)));
}
});
//发起异步连接请求，绑定连接端口和host信息
future = bootstrap.connect(host, port).sync();
//添加重连测试
ClientHandler.getBootstrap().put("nettyServer",bootstrap);
future.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture arg0) throws Exception {
if (future.isSuccess()) {
System.out.println("连接服务器成功");
List channelFuture = ClientHandler.getChannelFuture();
channelFuture.add(future);
List eventLoopGroup = ClientHandler.getEventLoopGroup();
eventLoopGroup.add(group);
} else {
System.out.println("连接服务器失败");
future.cause().printStackTrace();
group.shutdownGracefully(); //关闭线程组
throw new BizException(500,"连接Netty服务器失败",null);
}
}
});
future.channel().closeFuture().sync();
}catch (Exception e){
e.printStackTrace();
log.error("客户端启动失败！服务端不在线！", e);
throw new BizException(500,"客户端启动失败！服务端不在线！"+e,null);
}
}
特别提醒：
若有时候真的忘记了先启动服务端怎么办呢，问题不大！在代码中我已经做了重连机制。使用线程池在规定的时间间隔中，循环进行尝试重连。
new Thread(new Runnable() {
@SneakyThrows
@Override
public void run() {
try {
new NettyClient(monitorIp, monitorPort).start(); //启动netty
}catch (BizException e){
if (e.getErrorMsg().contains("服务端不在线")){
log.error("--------------------------error----------------------->"+e.getErrorMsg());
//连接异常 服务端没有开启 进行重新连接
DingDingSendUtil.postSend("客户端自启动失败！Netty服务端不在线！请检查服务！");
new ClientHandler(new NettyClient(monitorIp, monitorPort)).initScheduledExecutor(null);
}
e.printStackTrace();
throw new BizException(500,"Netty客户端【KHD_ONE】，启动异常！"+e.getMessage(),null);
}
}
},"NettyServer").start();
Netty通信的服务就是这么多内容，核心都在 ClientHandler 和 ServerHandler 中，剩下的是一些通过TCP与喇叭进行通信的内容了。有兴趣的也可以看看如何进行操作。这个功能模块基本上用了一个月时间完成，使用到了异步通信，多线程与线程池，TCP通信，定时任务等等。有兴趣的朋友可以自行选择下载源码查阅。
