項目背景

最近公司某物聯網項目需要使用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項目移除了消息中間件，于是需要自己創建一個本地隊列模擬真實使用場景

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{
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，日志如下：

可以看到第二條消息重新入列并被延遲消費了。