項目背景

最近公司某物聯網項目需要使用socket長連接進行消息通訊，搗鼓了一版代碼上線，結果BUG不斷，本猿寢食難安，于是求助度娘，數日未眠項目終于平穩(wěn)運行了，本著開源共享的精神，本猿把項目代碼提煉成了一個demo項目，盡量摒棄了其中丑陋的業(yè)務部分，希望與同學們共同學習進步。

基于 Spring Boot + MyBatis Plus + Vue & Element 實現的后臺管理系統(tǒng) + 用戶小程序，支持 RBAC 動態(tài)權限、多租戶、數據權限、工作流、三方登錄、支付、短信、商城等功能

正文

一、項目架構

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

二、項目模塊

本項目目錄結構如下圖：

netty-tcp-core是公共模塊，主要是工具類。netty-tcp-server是netty服務端，服務端僅作測試使用，實際項目中我們只使用了客戶端。netty-tcp-client是客戶端，也是本文的重點。

三、業(yè)務流程

我們實際項目中使用RocketMQ作為消息隊列，本項目由于是demo項目于是改為了BlockingQueue。數據流為：

生產者->消息隊列->消費者(客戶端)->tcp通道->服務端->tcp通道->客戶端。

當消費者接收到某設備發(fā)送的消息后，將判斷緩存中是否存在該設備與服務端的連接，如果存在并且通道活躍則使用該通道發(fā)送消息，如果不存在則創(chuàng)建通道并在通道激活后立即發(fā)送消息，當客戶端收到來自服務端的消息時進行響應的業(yè)務處理。

四、代碼詳解

1.消息隊列

由于本demo項目移除了消息中間件，于是需要自己創(chuàng)建一個本地隊列模擬真實使用場景

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

使用一個類保存隊列的靜態(tài)實例以便在任何類中都可以快速引用。接下來我們需要啟動一個線程去監(jiān)聽隊列中的消息，一但消息投遞到隊列中，我們就取出消息然后異步多線程處理該消息。

publicclassLoopThreadimplementsRunnable{
@Override
publicvoidrun(){
for(inti=0;i{
while(true){
//取走BlockingQueue里排在首位的對象,若BlockingQueue為空,阻斷進入等待狀態(tài)直到
try{
NettyMsgModelnettyMsgModel=QueueHolder.get().take();
messageProcessor.process(nettyMsgModel);
}catch(InterruptedExceptione){
log.error(e.getMessage(),e);
}
}
});
}
}
}

使用take方法會使該線程一直阻塞直到隊列收到消息后進入下一次循環(huán)。

2.執(zhí)行類

process方法來自于MessageProcessor類，該類為單例，但是會有多線程同時執(zhí)行。

publicvoidprocess(NettyMsgModelnettyMsgModel){
Stringimei=nettyMsgModel.getImei();
try{
synchronized(this){//為避免收到同一臺設備多條消息后重復創(chuàng)建客戶端，必須加鎖
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);
}
}
//緩存中存在則發(fā)送消息
if(NettyClientHolder.get().containsKey(imei)){
NettyClientnettyClient=NettyClientHolder.get().get(imei);
if(null!=nettyClient.getChannelFuture()&&nettyClient.getChannelFuture().channel().isActive()){//通道活躍直接發(fā)送消息
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();
//重新創(chuàng)建客戶端發(fā)送
this.createClientAndSend(nettyMsgModel);
}
}else{//緩存中不存在則創(chuàng)建新的客戶端
this.createClientAndSend(nettyMsgModel);
}
}catch(Exceptione){
log.error(e.getMessage(),e);
}finally{
//執(zhí)行完后解鎖
redisCache.deleteObject(NETTY_QUEUE_LOCK+imei);
}

}

其中imei是我們設備的唯一標識，我們可以用imei作為緩存的key來確認是否已創(chuàng)建過連接。由于我們消息的并發(fā)量可能會很大，所以存在當某設備的連接正在創(chuàng)建的過程中，另一個線程收到該設備消息也開始創(chuàng)建連接的情況，所以我們使用synchronized 代碼塊以及redis分布式鎖來避免此情況的發(fā)生。當一條消息獲得鎖后，在鎖釋放前，后續(xù)消息將會被重新放回消息隊列并延遲消費。

獲取鎖的線程會根據imei判斷緩存是否存在連接，如果存在直接發(fā)送消息，如果不存在則進入創(chuàng)建客戶端的方法。

privatevoidcreateClientAndSend(NettyMsgModelnettyMsgModel){
log.info("創(chuàng)建客戶端執(zhí)行中imei={}",nettyMsgModel.getImei());
//此處的DemoClientHandler可以根據自己的業(yè)務定義
NettyClientnettyClient=SpringUtils.getBean(NettyClient.class,nettyMsgModel.getImei(),nettyMsgModel.getBizData(),
this.createDefaultWorkGroup(this.workerThread),DemoClientHandler.class);
executor.execute(nettyClient);//執(zhí)行客戶端初始化
try{
//利用鎖等待客戶端激活
synchronized(nettyClient){
longc1=System.currentTimeMillis();
nettyClient.wait(5000);//最多阻塞5秒5秒后客戶端仍然未激活則自動解鎖
longc2=System.currentTimeMillis();
log.info("創(chuàng)建客戶端wait耗時={}ms",c2-c1);
}
if(null!=nettyClient.getChannelFuture()&&nettyClient.getChannelFuture().channel().isActive()){//連接成功
//存入緩存
NettyClientHolder.get().put(nettyMsgModel.getImei(),nettyClient);
//客戶端激活后發(fā)送消息
nettyClient.send(nettyMsgModel.getMsg());
}else{//連接失敗
log.warn("客戶端創(chuàng)建失敗，imei={}",nettyMsgModel.getImei());
nettyClient.close();
//可以把消息重新入列處理
}
}catch(Exceptione){
log.error("客戶端初始化發(fā)送消息異常===>{}",e.getMessage(),e);
}
}

當netty客戶端實例創(chuàng)建后使用線程池執(zhí)行初始化，由于是異步執(zhí)行，我們此時立刻發(fā)送消息很可能客戶端還沒有完成連接，因此必須加鎖等待。進入synchronized 代碼塊，使用wait方法等待客戶端激活后解鎖，參數5000為自動解鎖的毫秒數，意思是如果客戶端出現異常情況遲遲未能連接成功并激活通道、解鎖，則最多5000毫秒后該鎖自動解開。

這參數在實際使用時可以視情況調整，在并發(fā)量很大的情況下，5秒的阻塞可能會導致線程池耗盡，或內存溢出。待客戶端創(chuàng)建成功并激活后則立即發(fā)送消息。

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;
//自定義業(yè)務數據
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("隊列消息發(fā)送===>{}",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();//連接成功后將持續(xù)阻塞該線程
}
}
}

netty客戶端為多實例，每個實例綁定一個線程，持續(xù)阻塞到客戶端關閉為止，每個客戶端中可以保存自己的業(yè)務數據，以便在后續(xù)與服務端交互時處理業(yè)務使用。客戶端執(zhí)行連接時，給了2次重試的機會，如果3次都沒連接成功則放棄。后續(xù)可以選擇將該消息重新入列消費。我們實際項目中，此處還應該預先給服務端發(fā)送一條登錄消息，待服務端確認后才能執(zhí)行后續(xù)通訊，這需要視實際情況進行調整。

另一個需要注意的點是EventLoopGroup是從構造函數傳入的，而不是在客戶端中創(chuàng)建的，因為當客戶端數量非常多時，每個客戶端都創(chuàng)建自己的線程組會極大的消耗服務器資源，因此我們在實際使用中是按業(yè)務去創(chuàng)建統(tǒng)一的線程組給該業(yè)務下的所有客戶端共同使用的，線程組的大小需要根據業(yè)務需求靈活配置。

在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){//當通道激活后解鎖隊列線程，然后再發(fā)送消息
this.nettyClient.notify();
}
}

@Override
publicvoidchannelInactive(ChannelHandlerContextctx)throwsException{
log.warn("客戶端imei={}，通道斷開連接",this.imei);
}

@Override
publicvoidchannelRead(ChannelHandlerContextctx,Objectmsg)throwsException{
log.info("客戶端imei={}，收到消息:{}",this.imei,msg);
//處理業(yè)務...
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={}觸發(fā)閑讀或寫第{}次",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引用，以便在后續(xù)處理具體業(yè)務。在channelActive方法中，我們可以看到執(zhí)行了客戶端實例的notify方法，此處就是在客戶端創(chuàng)建成功并且通道激活后解除wait鎖的地方。channelRead方法就是我們處理服務端發(fā)送過來的消息的方法，我們的具體業(yè)務應該在該方法執(zhí)行，當然不建議長時間阻塞客戶端的工作線程，可以考慮異步處理。

最后我們看一下客戶端緩存類。

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{

/**
*間隔發(fā)送兩條消息
*/
@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!"));
}

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

/**
*連續(xù)發(fā)送兩條消息第二條由于redis鎖將會重新放回隊列延遲消費
*/
@GetMapping("testThree")
publicvoidtestThree(){
QueueHolder.get().offer(NettyMsgModel.create("12345678","HelloWorld!"));
QueueHolder.get().offer(NettyMsgModel.create("12345678","HelloWorldToo!"));
}
}

測試接口代碼如上，調用testOne，日志如下：

可以看到第一條消息觸發(fā)了客戶端創(chuàng)建流程，創(chuàng)建后發(fā)送了消息，而5秒后的第二條消息直接通過已有通道發(fā)送了。

測試接口代碼如上，調用testTwo，日志如下：

發(fā)送shutdown可以主動斷開已有連接。

測試接口代碼如上，調用testThree，日志如下：

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