基于Netty和SpringBoot的TCP长连接通讯方案-德赢Vwin官网网

项目背景

最近公司某物联网项目需要使用socket长连接进行消息通讯，捣鼓了一版代码上线，结果BUG不断，本猿寝食难安，于是求助度娘，数日未眠项目终于平稳运行了，本着开源共享的精神，本猿把项目代码提炼成了一个demo项目，尽量摒弃了其中丑陋的业务部分，希望与同学们共同学习进步。

基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能

正文

一、项目架构

本项目使用了netty、redis以及springboot2.2.0

二、项目模块

本项目目录结构如下图：

netty-tcp-core是公共模块，主要是工具类。netty-tcp-server是netty服务端，服务端仅作测试使用，实际项目中我们只使用了客户端。netty-tcp-client是客户端，也是本文的重点。

三、业务流程

我们实际项目中使用RocketMQ作为消息队列，本项目由于是demo项目于是改为了BlockingQueue。数据流为：

生产者->消息队列->消费者(客户端)->tcp通道->服务端->tcp通道->客户端。

当消费者接收到某设备发送的消息后，将判断缓存中是否存在该设备与服务端的连接，如果存在并且通道活跃则使用该通道发送消息，如果不存在则创建通道并在通道激活后立即发送消息，当客户端收到来自服务端的消息时进行响应的业务处理。

四、代码详解

1.消息队列

由于本demo项目移除了消息中间件，于是需要自己创建一个本地队列vwin 真实使用场景

packageorg.example.client;

importorg.example.client.model.NettyMsgModel;

importjava.util.concurrent.ArrayBlockingQueue;

/**
*本项目为演示使用本地队列实际生产中应该使用消息中间件代替（rocketmq或rabbitmq）
*
*@authorReWind00
*@date2023/2/1511:20
*/
publicclassQueueHolder{

privatestaticfinalArrayBlockingQueuequeue=newArrayBlockingQueue<>(100);

publicstaticArrayBlockingQueueget(){
returnqueue;
}
}

使用一个类保存队列的静态实例以便在任何类中都可以快速引用。接下来我们需要启动一个线程去监听队列中的消息，一但消息投递到队列中，我们就取出消息然后异步多线程处理该消息。

publicclassLoopThreadimplementsRunnable{
@Override
publicvoidrun(){
for(inti=0;i< MAIN_THREAD_POOL_SIZE; i++) {
            executor.execute(() ->{
while(true){
//取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到
try{
NettyMsgModelnettyMsgModel=QueueHolder.get().take();
messageProcessor.process(nettyMsgModel);
}catch(InterruptedExceptione){
log.error(e.getMessage(),e);
}
}
});
}
}
}

使用take方法会使该线程一直阻塞直到队列收到消息后进入下一次循环。

2.执行类

process方法来自于MessageProcessor类，该类为单例，但是会有多线程同时执行。

publicvoidprocess(NettyMsgModelnettyMsgModel){
Stringimei=nettyMsgModel.getImei();
try{
synchronized(this){//为避免收到同一台设备多条消息后重复创建客户端，必须加锁
if(redisCache.hasKey(NETTY_QUEUE_LOCK+imei)){//上一条消息处理中
log.info("imei={}消息处理中，重新入列",imei);
//放回队列重新等待消费延迟x秒（实际项目中应该使用rocketmq或者rabbitmq实现延迟消费）
newTimer().schedule(newTimerTask(){
@Override
publicvoidrun(){
QueueHolder.get().offer(nettyMsgModel);
}
},2000);
log.info("imei={}消息处理中，重新入列完成",imei);
return;
}else{
//如果没有在连接中的直接加锁
redisCache.setCacheObject(NETTY_QUEUE_LOCK+imei,"1",120,TimeUnit.SECONDS);
}
}
//缓存中存在则发送消息
if(NettyClientHolder.get().containsKey(imei)){
NettyClientnettyClient=NettyClientHolder.get().get(imei);
if(null!=nettyClient.getChannelFuture()&&nettyClient.getChannelFuture().channel().isActive()){//通道活跃直接发送消息
if(!nettyClient.getChannelFuture().channel().isWritable()){
log.warn("警告，通道不可写，imei={}，channelId={}",nettyClient.getImei(),
nettyClient.getChannelFuture().channel().id());
}
nettyClient.send(nettyMsgModel.getMsg());
}else{
log.info("clientimei={}，通道不活跃，主动关闭",nettyClient.getImei());
nettyClient.close();
//重新创建客户端发送
this.createClientAndSend(nettyMsgModel);
}
}else{//缓存中不存在则创建新的客户端
this.createClientAndSend(nettyMsgModel);
}
}catch(Exceptione){
log.error(e.getMessage(),e);
}finally{
//执行完后解锁
redisCache.deleteObject(NETTY_QUEUE_LOCK+imei);
}

}

其中imei是我们设备的唯一标识，我们可以用imei作为缓存的key来确认是否已创建过连接。由于我们消息的并发量可能会很大，所以存在当某设备的连接正在创建的过程中，另一个线程收到该设备消息也开始创建连接的情况，所以我们使用synchronized 代码块以及redis分布式锁来避免此情况的发生。当一条消息获得锁后，在锁释放前，后续消息将会被重新放回消息队列并延迟消费。

获取锁的线程会根据imei判断缓存是否存在连接，如果存在直接发送消息，如果不存在则进入创建客户端的方法。

privatevoidcreateClientAndSend(NettyMsgModelnettyMsgModel){
log.info("创建客户端执行中imei={}",nettyMsgModel.getImei());
//此处的DemoClientHandler可以根据自己的业务定义
NettyClientnettyClient=SpringUtils.getBean(NettyClient.class,nettyMsgModel.getImei(),nettyMsgModel.getBizData(),
this.createDefaultWorkGroup(this.workerThread),DemoClientHandler.class);
executor.execute(nettyClient);//执行客户端初始化
try{
//利用锁等待客户端激活
synchronized(nettyClient){
longc1=System.currentTimeMillis();
nettyClient.wait(5000);//最多阻塞5秒5秒后客户端仍然未激活则自动解锁
longc2=System.currentTimeMillis();
log.info("创建客户端wait耗时={}ms",c2-c1);
}
if(null!=nettyClient.getChannelFuture()&&nettyClient.getChannelFuture().channel().isActive()){//连接成功
//存入缓存
NettyClientHolder.get().put(nettyMsgModel.getImei(),nettyClient);
//客户端激活后发送消息
nettyClient.send(nettyMsgModel.getMsg());
}else{//连接失败
log.warn("客户端创建失败，imei={}",nettyMsgModel.getImei());
nettyClient.close();
//可以把消息重新入列处理
}
}catch(Exceptione){
log.error("客户端初始化发送消息异常===>{}",e.getMessage(),e);
}
}

当netty客户端实例创建后使用线程池执行初始化，由于是异步执行，我们此时立刻发送消息很可能客户端还没有完成连接，因此必须加锁等待。进入synchronized 代码块，使用wait方法等待客户端激活后解锁，参数5000为自动解锁的毫秒数，意思是如果客户端出现异常情况迟迟未能连接成功并激活通道、解锁，则最多5000毫秒后该锁自动解开。

这参数在实际使用时可以视情况调整，在并发量很大的情况下，5秒的阻塞可能会导致线程池耗尽，或内存溢出。待客户端创建成功并激活后则立即发送消息。

3.客户端

packageorg.example.client;

importio.netty.bootstrap.Bootstrap;
importio.netty.buffer.Unpooled;
importio.netty.channel.*;
importio.netty.channel.socket.SocketChannel;
importio.netty.channel.socket.nio.NioSocketChannel;
importio.netty.handler.codec.DelimiterBasedFrameDecoder;
importio.netty.handler.codec.string.StringDecoder;
importio.netty.handler.codec.string.StringEncoder;
importio.netty.handler.timeout.IdleStateHandler;
importio.netty.util.CharsetUtil;
importlombok.Getter;
importlombok.NoArgsConstructor;
importlombok.extern.slf4j.Slf4j;
importorg.example.client.handler.BaseClientHandler;
importorg.example.core.util.SpringUtils;
importorg.springframework.beans.factory.annotation.Value;
importorg.springframework.context.annotation.Scope;
importorg.springframework.stereotype.Component;
importorg.springframework.util.StringUtils;

importjava.util.Map;
importjava.util.concurrent.TimeUnit;
importjava.util.concurrent.atomic.AtomicBoolean;
importjava.util.concurrent.atomic.AtomicInteger;

/**
*@authorReWind00
*@date2023/2/159:59
*/
@Slf4j
@Component
@Scope("prototype")
@Getter
@NoArgsConstructor
publicclassNettyClientimplementsRunnable{

@Value("${netty.server.port}")
privateintport;

@Value("${netty.server.host}")
privateStringhost;
//客户端唯一标识
privateStringimei;
//自定义业务数据
privateMapbizData;

privateEventLoopGroupworkGroup;

privateClassclientHandlerClass;

privateChannelFuturechannelFuture;

publicNettyClient(Stringimei,MapbizData,EventLoopGroupworkGroup,ClassclientHandlerClass){
this.imei=imei;
this.bizData=bizData;
this.workGroup=workGroup;
this.clientHandlerClass=clientHandlerClass;
}

@Override
publicvoidrun(){
try{
this.init();
log.info("客户端启动imei={}",imei);
}catch(Exceptione){
log.error("客户端启动失败:{}",e.getMessage(),e);
}
}

publicvoidclose(){
if(null!=this.channelFuture){
this.channelFuture.channel().close();
}
NettyClientHolder.get().remove(this.imei);
}

publicvoidsend(Stringmessage){
try{
if(!this.channelFuture.channel().isActive()){
log.info("通道不活跃imei={}",this.imei);
return;
}
if(!StringUtils.isEmpty(message)){
log.info("队列消息发送===>{}",message);
this.channelFuture.channel().writeAndFlush(message);
}
}catch(Exceptione){
log.error(e.getMessage(),e);
}
}

privatevoidinit()throwsException{
//将本实例传递到handler
BaseClientHandlerclientHandler=SpringUtils.getBean(clientHandlerClass,this);
Bootstrapb=newBootstrap();
//2通过辅助类去构造server/client
b.group(workGroup)
.channel(NioSocketChannel.class)
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,3000)
.option(ChannelOption.SO_RCVBUF,1024*32)
.option(ChannelOption.SO_SNDBUF,1024*32)
.handler(newChannelInitializer(){
@Override
protectedvoidinitChannel(SocketChannelch)throwsException{
ch.pipeline().addLast(newDelimiterBasedFrameDecoder(1024*1024,Unpooled.copiedBuffer("
".getBytes())));
ch.pipeline().addLast(newStringEncoder(CharsetUtil.UTF_8));//String解码。
ch.pipeline().addLast(newStringDecoder(CharsetUtil.UTF_8));//String解码。
////心跳设置
ch.pipeline().addLast(newIdleStateHandler(0,0,600,TimeUnit.SECONDS));
ch.pipeline().addLast(clientHandler);
}
});
this.connect(b);
}

privatevoidconnect(Bootstrapb)throwsInterruptedException{
longc1=System.currentTimeMillis();
finalintmaxRetries=2;//重连2次
finalAtomicIntegercount=newAtomicInteger();
finalAtomicBooleanflag=newAtomicBoolean(false);
try{
this.channelFuture=b.connect(host,port).addListener(
newChannelFutureListener(){
publicvoidoperationComplete(ChannelFuturefuture)throwsException{
if(!future.isSuccess()){
if(count.incrementAndGet()>maxRetries){
log.warn("imei={}重连超过{}次",imei,maxRetries);
}else{
log.info("imei={}重连第{}次",imei,count);
b.connect(host,port).addListener(this);
}

}else{
log.info("imei={}连接成功,连接IP:{}连接端口:{}",imei,host,port);
flag.set(true);
}
}
}).sync();//同步连接
}catch(Exceptione){
log.error(e.getMessage(),e);
}
log.info("设备imei={}，channelId={}连接耗时={}ms",imei,channelFuture.channel().id(),System.currentTimeMillis()-c1);
if(flag.get()){
channelFuture.channel().closeFuture().sync();//连接成功后将持续阻塞该线程
}
}
}

netty客户端为多实例，每个实例绑定一个线程，持续阻塞到客户端关闭为止，每个客户端中可以保存自己的业务数据，以便在后续与服务端交互时处理业务使用。客户端执行连接时，给了2次重试的机会，如果3次都没连接成功则放弃。后续可以选择将该消息重新入列消费。我们实际项目中，此处还应该预先给服务端发送一条登录消息，待服务端确认后才能执行后续通讯，这需要视实际情况进行调整。

另一个需要注意的点是EventLoopGroup是从构造函数传入的，而不是在客户端中创建的，因为当客户端数量非常多时，每个客户端都创建自己的线程组会极大的消耗服务器资源，因此我们在实际使用中是按业务去创建统一的线程组给该业务下的所有客户端共同使用的，线程组的大小需要根据业务需求灵活配置。

在init方法中，我们给客户端加上了一个handler来处理与服务端的交互，下面来看一下具体实现。

packageorg.example.client.handler;

importio.netty.channel.ChannelHandlerContext;
importio.netty.handler.timeout.IdleState;
importio.netty.handler.timeout.IdleStateEvent;
importlombok.extern.slf4j.Slf4j;
importorg.example.client.NettyClient;
importorg.springframework.context.annotation.Scope;
importorg.springframework.stereotype.Component;

importjava.util.Map;

/**
*@authorReWind00
*@date2023/2/1510:09
*/
@Slf4j
@Component
@Scope("prototype")
publicclassDemoClientHandlerextendsBaseClientHandler{

privatefinalStringimei;

privatefinalMapbizData;

privatefinalNettyClientnettyClient;

privateintallIdleCounter=0;

privatestaticfinalintMAX_IDLE_TIMES=3;

publicDemoClientHandler(NettyClientnettyClient){
this.nettyClient=nettyClient;
this.imei=nettyClient.getImei();
this.bizData=nettyClient.getBizData();
}

@Override
publicvoidchannelActive(ChannelHandlerContextctx)throwsException{
log.info("客户端imei={}，通道激活成功",this.imei);
synchronized(this.nettyClient){//当通道激活后解锁队列线程，然后再发送消息
this.nettyClient.notify();
}
}

@Override
publicvoidchannelInactive(ChannelHandlerContextctx)throwsException{
log.warn("客户端imei={}，通道断开连接",this.imei);
}

@Override
publicvoidchannelRead(ChannelHandlerContextctx,Objectmsg)throwsException{
log.info("客户端imei={}，收到消息:{}",this.imei,msg);
//处理业务...
if("shutdown".equals(msg)){
this.nettyClient.close();
}
}

@Override
publicvoiduserEventTriggered(ChannelHandlerContextctx,Objectevt)throwsException{
if(evtinstanceofIdleStateEvent){
IdleStateEvente=(IdleStateEvent)evt;
booleanflag=false;
if(e.state()==IdleState.ALL_IDLE){
this.allIdleCounter++;
log.info("客户端imei={}触发闲读或写第{}次",this.imei,this.allIdleCounter);
if(this.allIdleCounter>=MAX_IDLE_TIMES){
flag=true;
}
}
if(flag){
log.warn("读写超时达到{}次，主动断开连接",MAX_IDLE_TIMES);
ctx.channel().close();
}
}
}

@Override
publicvoidexceptionCaught(ChannelHandlerContextctx,Throwablecause)throwsException{
log.error("客户端imei={}，连接异常{}",imei,cause.getMessage(),cause);
}
}

DemoClientHandler也是多实例bean，每个实例持有自己的NettyClient引用，以便在后续处理具体业务。在channelActive方法中，我们可以看到执行了客户端实例的notify方法，此处就是在客户端创建成功并且通道激活后解除wait锁的地方。channelRead方法就是我们处理服务端发送过来的消息的方法，我们的具体业务应该在该方法执行，当然不建议长时间阻塞客户端的工作线程，可以考虑异步处理。

最后我们看一下客户端缓存类。

packageorg.example.client;

importjava.util.concurrent.ConcurrentHashMap;

/**
*@authorReWind00
*@date2023/2/1511:01
*/
publicclassNettyClientHolder{

privatestaticfinalConcurrentHashMapclientMap=newConcurrentHashMap<>();

publicstaticConcurrentHashMapget(){
returnclientMap;
}

}

由于netty的通道无法序列化，因此不能存入redis，只能缓存在本地内存中，其本质就是一个ConcurrentHashMap。

五、测试

packageorg.example.client.controller;

importorg.example.client.QueueHolder;
importorg.example.client.model.NettyMsgModel;
importorg.springframework.web.bind.annotation.GetMapping;
importorg.springframework.web.bind.annotation.RequestMapping;
importorg.springframework.web.bind.annotation.RequestParam;
importorg.springframework.web.bind.annotation.RestController;

/**
*@authorReWind00
*@date2023/2/1513:48
*/
@RestController
@RequestMapping("/demo")
publicclassDemoController{

/**
*间隔发送两条消息
*/
@GetMapping("testOne")
publicvoidtestOne(){
QueueHolder.get().offer(NettyMsgModel.create("87654321","HelloWorld!"));
try{
Thread.sleep(5000);
}catch(InterruptedExceptione){
e.printStackTrace();
}
QueueHolder.get().offer(NettyMsgModel.create("87654321","HelloWorldToo!"));
}

/**
*任意发送消息
*
*@paramimei
*@parammsg
*/
@GetMapping("testTwo")
publicvoidtestTwo(@RequestParamStringimei,@RequestParamStringmsg){
QueueHolder.get().offer(NettyMsgModel.create(imei,msg));
}

/**
*连续发送两条消息第二条由于redis锁将会重新放回队列延迟消费
*/
@GetMapping("testThree")
publicvoidtestThree(){
QueueHolder.get().offer(NettyMsgModel.create("12345678","HelloWorld!"));
QueueHolder.get().offer(NettyMsgModel.create("12345678","HelloWorldToo!"));
}
}

测试接口代码如上，调用testOne，日志如下：

可以看到第一条消息触发了客户端创建流程，创建后发送了消息，而5秒后的第二条消息直接通过已有通道发送了。

测试接口代码如上，调用testTwo，日志如下：

发送shutdown可以主动断开已有连接。

测试接口代码如上，调用testThree，日志如下：

可以看到第二条消息重新入列并被延迟消费了。

审核编辑：汤梓红

声明：本文内容及配图由入驻作者撰写或者入驻合作网站授权转载。文章观点仅代表作者本人，不代表德赢Vwin官网网立场。文章及其配图仅供工程师学习之用，如有内容侵权或者其他违规问题，请联系本站处理。举报投诉

物联网

物联网

+关注

关注
2909

文章
44557

浏览量
372774
通讯

通讯

+关注

关注
9

文章
902

浏览量
34889
TCP

TCP

+关注

关注
8

文章
1353

浏览量
79055
spring

spring

+关注

关注
0

文章
340

浏览量
14338
SpringBoot

SpringBoot

+关注

关注
0

文章
173

浏览量
177

原文标题：Netty+SpringBoot 打造一个 TCP 长连接通讯方案

文章出处：【微信号：芋道源码，微信公众号：芋道源码】欢迎添加关注！文章转载请注明出处。

Modbus Tcp通讯中断问题

Labview 2013 + Modbus Tcp + AC500(PLC) 通讯，数据量很小，也就发几个指令。连线为笔记本通过一根2米长的网线直接连在PLC上，中间没有路由器。在公司测试，一切正常

发表于 03-13 14:54

springboot集成mqtt

springboot集成mqtt，大纲一.数据入库1.数据入库解决方案二.开发实时订阅发布展示页面1.及时通讯技术2.技术整合

发表于 07-16 07:53

NETTY自定义协议的TCP服务器

一、想法及需求1.1最初设想1.2需求分析二、硬件2.1原理图解释2.2PCB绘制2.3焊接及成品三、软件3.1NETTY自定义协议的TCP服务器3.1.1使用原因为什么要使用自定义的协议呢，原因有

发表于 12-21 08:30

怎样使用springboot整合netty来开发一套高性能的通信系统呢

怎样使用springboot整合netty来开发一套高性能的通信系统呢？为什么要用这两个框架来实现通信服务呢？如何去实现呢？

发表于 02-22 06:09

基于Netty的长连接客户端

用netty开发都是作为服务端来的大致思路如下：（1）创建一个eventLoopGroup，用于维护nio的io事件（2）创建一个niosocketchanel，然后将其注册到

发表于 11-27 15:52 •7412次阅读

mpu6050与单片机串口连接通讯实验资料下载

mpu6050与单片机串口连接通讯实验资料下载

发表于 04-23 09:11 •15次下载

mpu6050与单片机串口<b class='flag-5'>连接通讯</b>实验资料下载

如何使用SpringBoot集成Netty开发一个基于WebSocket的聊天室说明

本文档的主要内容详细介绍的是基于SpringBoot，借助Netty控制长链接，使用WebSocket协议做一个实时的聊天室。

发表于 05-29 17:56 •1次下载

如何使用<b class='flag-5'>SpringBoot</b>集成<b class='flag-5'>Netty</b>开发一个基于WebSocket的聊天室说明

TCP, ISO- on- TCP, UDP连接

TSEND“ & „TRCV “ 发送和接收数据(TCP 和ISO - on- TCP)„TUSEND“ & „TURCV“ 发送和接收数据(UDP) 自动连接管理的通讯块

发表于 06-12 15:11 •5109次阅读

<b class='flag-5'>TCP</b>, ISO- on- <b class='flag-5'>TCP</b>, UDP<b class='flag-5'>连接</b>

Springboot整合netty框架实现终端、通讯板子(单片机)TCP/UDP通信案例

如何springboot和netty案例的源代码一个springboot整合netty框架的开发小案例，实现服务端与单片机终端实时通信的通讯

发表于 12-29 18:55 •20次下载

<b class='flag-5'>Springboot</b>整合<b class='flag-5'>netty</b>框架实现终端、<b class='flag-5'>通讯</b>板子(单片机)<b class='flag-5'>TCP</b>/UDP通信案例

netty推送消息接口及实现

学过 Netty 的都知道，Netty 对 NIO 进行了很好的封装，简单的 API，庞大的开源社区。深受广大程序员喜爱。基于此本文分享一下基础的 netty 使用。实战制作一个 Netty

发表于 11-02 16:14 •1488次阅读

一步步解决长连接Netty服务内存泄漏

线上应用长连接 Netty 服务出现内存泄漏了！真让人头大

发表于 04-27 14:06 •1124次阅读

使用Netty+SpringBoot打造的TCP长连接通讯方案

最近公司某物联网项目需要使用socket长连接进行消息通讯，捣鼓了一版代码上线，结果BUG不断，本猿寝食难安，于是求助度娘，数日未眠项目终于平稳运行了，本着开源共享的精神，本猿把项目代码提炼成了一个demo项目，尽量摒弃了其中丑

发表于 04-27 14:25 •1490次阅读

TCP通信过程中的长连接与短连接是什么？

当面试官问你：TCP 通信过程中的长连接与短连接是什么？

发表于 08-08 11:30 •1203次阅读

SpringBoot 连接ElasticSearch的使用方式

SpringBoot，今天我们就以 SpringBoot 整合 ElasticSearch 为例，给大家详细的介绍 ElasticSearch 的使用！ SpringBoot 连接

发表于 10-09 10:35 •1098次阅读

TCP的长连接和短连接

TCP在真正开始进行数据传输之前，Server 和 Client 之间必须建立一个连接。当数据传输完成后，双方不再需要这个连接时，就可以释放这个连接。

发表于 11-13 10:46 •1009次阅读

搜索历史

基于Netty和SpringBoot的TCP长连接通讯方案

评论

Modbus Tcp通讯中断问题

springboot集成mqtt

NETTY自定义协议的TCP服务器

怎样使用springboot整合netty来开发一套高性能的通信系统呢

基于Netty的长连接客户端

mpu6050与单片机串口连接通讯实验资料下载

如何使用SpringBoot集成Netty开发一个基于WebSocket的聊天室说明

TCP, ISO- on- TCP, UDP连接

Springboot整合netty框架实现终端、通讯板子(单片机)TCP/UDP通信案例

netty推送消息接口及实现

一步步解决长连接Netty服务内存泄漏

使用Netty+SpringBoot打造的TCP长连接通讯方案

TCP通信过程中的长连接与短连接是什么？

SpringBoot 连接ElasticSearch的使用方式

TCP的长连接和短连接