實(shí)戰(zhàn)即時(shí)聊天，干起?。?！
liu點(diǎn)擊上方藍(lán)色字體，選擇“標(biāo)星公眾號(hào)”
優(yōu)質(zhì)文章，第一時(shí)間送達(dá)
作者 | 失足成萬(wàn)古風(fēng)流人物
來源 | urlify.cn/BviMBr
一、前言
說實(shí)話，寫這個(gè)玩意兒是我上周剛剛產(chǎn)生的想法，本想寫完后把代碼掛上來賺點(diǎn)積分也不錯(cuò)。寫完后發(fā)現(xiàn)這東西值得寫一篇文章，授人予魚不如授人以漁嘛（這句話是這么說的吧），順便賺點(diǎn)應(yīng)屆學(xué)生MM的膜拜那就更妙了。然后再掛一個(gè)收款二維碼，一個(gè)人1塊錢，一天10000個(gè)人付款，一個(gè)月30萬(wàn)，一年360萬(wàn)。。?？闪瞬坏昧?，離一個(gè)億的小目標(biāo)就差幾十年了。
不知道博客園對(duì)夢(mèng)話有沒有限制，有的話請(qǐng)告知，我會(huì)盡快刪除上述文字。
那么現(xiàn)在回到現(xiàn)實(shí)中，這篇博文如果能有>2個(gè)評(píng)論，我后續(xù)會(huì)再出一個(gè)Netty相關(guān)的專欄。否則，就不出了。有人會(huì)好奇，為什么把閾值定義成>2呢？不為什么，因?yàn)槲铱隙〞?huì)先用我媳婦兒的號(hào)留個(gè)言，然后用自己的號(hào)留個(gè)言。
好了，廢話不多說了，后面還有好多事兒呢，洗菜、做飯、刷碗、跪搓衣。。。好了，言歸正傳吧。
二、最終效果
為什么先看最終效果？因?yàn)榇丝檀a已經(jīng)擼完了。更重要的是我們帶著感官的目標(biāo)去進(jìn)行后續(xù)的分析，可以更好地理解。標(biāo)題中提到了，整個(gè)工程包含三個(gè)部分：
1、聊天服務(wù)器
聊天服務(wù)器的職責(zé)一句話解釋：負(fù)責(zé)接收所有用戶發(fā)送的消息，并將消息轉(zhuǎn)發(fā)給目標(biāo)用戶。
聊天服務(wù)器沒有任何界面，但是卻是IM中最重要的角色，為表達(dá)敬意，必須要給它放個(gè)效果圖：
2021-05-11 10:41:40.037 INFO 9392 --- [ntLoopGroup-3-1] c.e.o.s.netty.handler.HeartBeatHandler : server收到心跳包：{"time":1620700900029,"messageType":"99"}
2021-05-11 10:41:50.049 INFO 9392 --- [ntLoopGroup-3-1] c.e.o.s.n.handler.BussMessageHandler : 收到消息：{"time":1620700910045,"messageType":"14","sendUserName":"guodegang","recvUserName":"yuqian","sendMessage":"于老師你好"}
2021-05-11 10:41:50.055 INFO 9392 --- [ntLoopGroup-3-2] c.e.o.s.netty.executor.SendMsgExecutor : 消息轉(zhuǎn)發(fā)成功:{"time":1620700910052,"messageType":"14","sendUserName":"guodegang","recvUserName":"yuqian","sendMessage":"于老師你好"}
2021-05-11 10:41:54.068 INFO 9392 --- [ntLoopGroup-3-2] c.e.o.s.netty.handler.HeartBeatHandler : server收到心跳包：{"time":1620700914064,"messageType":"99"}
2021-05-11 10:41:57.302 INFO 9392 --- [ntLoopGroup-3-2] c.e.o.s.n.handler.BussMessageHandler : 收到消息：{"time":1620700917301,"messageType":"14","sendUserName":"yuqian","recvUserName":"guodegang","sendMessage":"郭老師你好"}
2021-05-11 10:41:57.304 INFO 9392 --- [ntLoopGroup-3-1] c.e.o.s.netty.executor.SendMsgExecutor : 消息轉(zhuǎn)發(fā)成功:{"time":1620700917303,"messageType":"14","sendUserName":"yuqian","recvUserName":"guodegang","sendMessage":"郭老師你好"}
2021-05-11 10:42:05.050 INFO 9392 --- [ntLoopGroup-3-1] c.e.o.s.netty.handler.HeartBeatHandler : server收到心跳包：{"time":1620700925049,"messageType":"99"}
2021-05-11 10:42:12.309 INFO 9392 --- [ntLoopGroup-3-2] c.e.o.s.netty.handler.HeartBeatHandler : server收到心跳包：{"time":1620700932304,"messageType":"99"}
2021-05-11 10:42:20.066 INFO 9392 --- [ntLoopGroup-3-1] c.e.o.s.netty.handler.HeartBeatHandler : server收到心跳包：{"time":1620700940050,"messageType":"99"}
2021-05-11 10:42:27.311 INFO 9392 --- [ntLoopGroup-3-2] c.e.o.s.netty.handler.HeartBeatHandler : server收到心跳包：{"time":1620700947309,"messageType":"99"}
2021-05-11 10:42:35.070 INFO 9392 --- [ntLoopGroup-3-1] c.e.o.s.netty.handler.HeartBeatHandler : server收到心跳包：{"time":1620700955068,"messageType":"99"}
2021-05-11 10:42:42.316 INFO 9392 --- [ntLoopGroup-3-2] c.e.o.s.netty.handler.HeartBeatHandler : server收到心跳包：{"time":1620700962312,"messageType":"99"}
2021-05-11 10:42:50.072 INFO 9392 --- [ntLoopGroup-3-1] c.e.o.s.netty.handler.HeartBeatHandler : server收到心跳包：{"time":1620700970071,"messageType":"99"}
2021-05-11 10:42:57.316 INFO 9392 --- [ntLoopGroup-3-2] c.e.o.s.netty.handler.HeartBeatHandler : server收到心跳包：{"time":1620700977315,"messageType":"99"}
從效果圖我們看到了一些內(nèi)容：收到心跳包、收到消息，轉(zhuǎn)發(fā)消息，這些內(nèi)容后面會(huì)詳細(xì)講解。
2、聊天客戶端
聊天客戶端的職責(zé)一句話解釋：登陸，給別人發(fā)聊天內(nèi)容，收其它人發(fā)給自己的聊天內(nèi)容。
下面為方便演示，我會(huì)打開兩個(gè)客戶端，用兩個(gè)不同用戶登陸，然后發(fā)消息。
3、Web管理控制臺(tái)
目前只做了一個(gè)賬戶管理，具體看圖吧：
三、需求分析
無(wú)（見第二章節(jié)）。
四、概要設(shè)計(jì)
1、技術(shù)選型
1）聊天服務(wù)端
聊天服務(wù)器與客戶端通過TCP協(xié)議進(jìn)行通信，使用長(zhǎng)連接、全雙工通信模式，基于經(jīng)典通信框架Netty實(shí)現(xiàn)。
那么什么是長(zhǎng)連接？顧名思義，客戶端和服務(wù)器連上后，會(huì)在這條連接上面反復(fù)收發(fā)消息，連接不會(huì)斷開。與長(zhǎng)連接對(duì)應(yīng)的當(dāng)然就是短連接了，短連接每次發(fā)消息之前都需要先建立連接，然后發(fā)消息，最后斷開連接。顯然，即時(shí)聊天適合使用長(zhǎng)連接。
那么什么又是全雙工？當(dāng)長(zhǎng)連接建立起來后，在這條連接上既有上行的數(shù)據(jù)，又有下行的數(shù)據(jù)，這就叫全雙工。那么對(duì)應(yīng)的半雙工、單工，大家自行百度吧。
2）Web管理控制臺(tái)
Web管理端使用SpringBoot腳手架，前端使用Layuimini（一個(gè)基于Layui前端框架封裝的前端框架），后端使用SpringMVC+Jpa+Shiro。
3）聊天客戶端
使用SpringBoot+JavaFX，做了一個(gè)極其簡(jiǎn)陋的客戶端，JavaFX是一個(gè)開發(fā)Java桌面程序的框架，本人也是第一次使用，代碼中的寫法都是網(wǎng)上查的，這并不是本文的重點(diǎn)，有興趣的仔細(xì)百度吧。
4）SpringBoot
以上三個(gè)組件，全部以SpringBoot做為腳手架開發(fā)。
5）代碼構(gòu)建
Maven。
2、數(shù)據(jù)庫(kù)設(shè)計(jì)
我們只簡(jiǎn)單用到一張用戶表，比較簡(jiǎn)單直接貼腳本：
CREATE TABLE `sys_user` (
`id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主鍵',
`user_name` varchar(64) DEFAULT NULL COMMENT '用戶名：登陸賬號(hào)',
`pass_word` varchar(128) DEFAULT NULL COMMENT '密碼',
`name` varchar(16) DEFAULT NULL COMMENT '昵稱',
`sex` char(1) DEFAULT NULL COMMENT '性別:1-男，2女',
`status` bit(1) DEFAULT NULL COMMENT '用戶狀態(tài)：1-有效，0-無(wú)效',
`online` bit(1) DEFAULT NULL COMMENT '在線狀態(tài)：1-在線，0-離線',
`salt` varchar(128) DEFAULT NULL COMMENT '密碼鹽值',
`admin` bit(1) DEFAULT NULL COMMENT '是否管理員（只有管理員才能登錄Web端）：1-是，0-否',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
這張表都在什么時(shí)候用到？
1）Web管理端登陸的時(shí)候；2）聊天客戶端將登陸請(qǐng)求發(fā)送到聊天服務(wù)端時(shí)，聊天服務(wù)端進(jìn)行用戶認(rèn)證；3）聊天客戶端的好友列表加載。
3、通信設(shè)計(jì)
本節(jié)將會(huì)是本文的核心內(nèi)容之一，主要描述通信報(bào)文協(xié)議格式、以及通信報(bào)文的交互場(chǎng)景。
1）報(bào)文協(xié)議格式
下面這張圖應(yīng)該能說明99%了：
剩下的1%在這里說：
a）粘包問題，TCP長(zhǎng)連接中，粘包是第一個(gè)需要解決的問題。通俗的講，粘包的意思是消息接收方往往收到的不是“整個(gè)”報(bào)文，有時(shí)候比“整個(gè)”多一點(diǎn)，有時(shí)候比“整個(gè)”少一點(diǎn)，這樣就導(dǎo)致接收方無(wú)法解析這個(gè)報(bào)文。那么上圖中的頭8個(gè)字節(jié)就為了解決這個(gè)問題，接收方根據(jù)頭8個(gè)字節(jié)標(biāo)識(shí)的長(zhǎng)度來獲取到“整個(gè)”報(bào)文，從而進(jìn)行正常的業(yè)務(wù)處理；
b）2字節(jié)報(bào)文類型，為了方便解析報(bào)文而設(shè)計(jì)。根據(jù)這兩個(gè)字節(jié)將后面的json轉(zhuǎn)成相應(yīng)的實(shí)體以便進(jìn)行后續(xù)處理；
c）變長(zhǎng)報(bào)文體實(shí)際上就是json格式的串，當(dāng)然，你可以自己設(shè)計(jì)報(bào)文格式，我這里為了方便處理就直接放json了；
d）當(dāng)然，你可以把報(bào)文設(shè)計(jì)的更復(fù)雜、更專業(yè)，比如加密、加簽名等。
2）報(bào)文交互場(chǎng)景
a）登陸
b）發(fā)送消息-成功
c）發(fā)送消息-目標(biāo)客戶端不在線
d）發(fā)送消息-目標(biāo)客戶端在線，但消息轉(zhuǎn)發(fā)失敗
五、編碼實(shí)現(xiàn)
前面說了那么多，現(xiàn)在總得說點(diǎn)有用的。
1、先說說Netty
Netty是一個(gè)相當(dāng)優(yōu)秀的通信框架，大多數(shù)的頂級(jí)開源框架中都有Netty的身影。具體它有多么優(yōu)秀，建議大家自行百度，我不如百度說的好。我只從應(yīng)用方面說說Netty。應(yīng)用過程中，它最核心的東西叫handler，我們可以簡(jiǎn)單理解它為消息處理器。收到的消息和出去的消息都會(huì)經(jīng)過一系列的handler加工處理。收到的消息我們叫它入站消息，發(fā)出去的消息我們叫它出站消息，因此handler又分為出站handler和入站handler。收到的消息只會(huì)被入站handler處理，發(fā)出去的消息只會(huì)被出站handler處理。
舉個(gè)例子，我們從網(wǎng)絡(luò)上收到的消息是二進(jìn)制的字節(jié)碼，我們的目標(biāo)是將消息轉(zhuǎn)換成java bean，這樣方便我們程序處理，針對(duì)這個(gè)場(chǎng)景我設(shè)計(jì)這么幾個(gè)入站handler：
1）將字節(jié)轉(zhuǎn)換成String的handler；
2）將String轉(zhuǎn)成java bean的handler；
3）對(duì)java bean進(jìn)行業(yè)務(wù)處理的handler。
發(fā)出去的消息呢，我設(shè)計(jì)這么幾個(gè)出站handler：
1）java bean 轉(zhuǎn)成String的handler；
2）String轉(zhuǎn)成byte的handler。
以上是關(guān)于handler的說明。
接下來再說一下Netty的異步。異步的意思是當(dāng)你做完一個(gè)操作后，不會(huì)立馬得到操作結(jié)果，而是有結(jié)果后Netty會(huì)通知你。通過下面的一段代碼來說明：
channel.writeAndFlush(sendMsgRequest).addListener(new GenericFutureListener<Future<? super Void>>() {
@Override
public void operationComplete(Future<? super Void> future) throws Exception {
if (future.isSuccess()){
logger.info("消息發(fā)送成功:{}",sendMsgRequest);
}else {
logger.info("消息發(fā)送失敗:{}",sendMsgRequest);
}
}
});
上面的writeAndFlush操作無(wú)法立即返回結(jié)果，如果你關(guān)注結(jié)果，那么為他添加一個(gè)listener，有結(jié)果后會(huì)在listener中響應(yīng)。
到這里，百度上搜到的Netty相關(guān)的代碼你基本就能看懂了。
2、聊天服務(wù)端
首先看主入口的代碼
public void start(){
EventLoopGroup boss = new NioEventLoopGroup();
EventLoopGroup worker = new NioEventLoopGroup();
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(boss, worker)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
//心跳
ch.pipeline().addLast(new IdleStateHandler(25, 20, 0, TimeUnit.SECONDS));
//收整包
ch.pipeline().addLast(new StringLengthFieldDecoder());
//轉(zhuǎn)字符串
ch.pipeline().addLast(new StringDecoder(Charset.forName("UTF-8")));
//json轉(zhuǎn)對(duì)象
ch.pipeline().addLast(new JsonDecoder());
//心跳
ch.pipeline().addLast(new HeartBeatHandler());
//實(shí)體轉(zhuǎn)json
ch.pipeline().addLast(new JsonEncoder());
//消息處理
ch.pipeline().addLast(bussMessageHandler);
}
});
try {
ChannelFuture f = serverBootstrap.bind(port).sync();
f.channel().closeFuture().sync();
}catch (InterruptedException e) {
logger.error("服務(wù)啟動(dòng)失?。簕}", ExceptionUtils.getStackTrace(e));
}finally {
worker.shutdownGracefully();
boss.shutdownGracefully();
}
}
代碼中除了initChannel方法中的代碼，其他代碼都是固定寫法。那么什么叫固定寫法呢？通俗來講就是可以Ctrl+c、Ctrl+v。
下面我們著重看initChannel方法里面的代碼。這里面就是上面講到的各種handler，我們下面挨個(gè)講這些handler都是干啥的。
1）IdleStateHandler。這個(gè)是Netty內(nèi)置的一個(gè)handler，既是出站handler又是入站handler。它的作用一般是用來實(shí)現(xiàn)心跳監(jiān)測(cè)。所謂心跳，就是客戶端和服務(wù)端建立連接后，服務(wù)端要實(shí)時(shí)監(jiān)控客戶端的健康狀態(tài)，如果客戶端掛了或者h(yuǎn)ung住了，服務(wù)端及時(shí)釋放相應(yīng)的資源，以及做出其他處理比如通知運(yùn)維。所以在我們的場(chǎng)景中，客戶端需要定時(shí)上報(bào)自己的心跳，如果服務(wù)端檢測(cè)到一段時(shí)間內(nèi)沒收到客戶端上報(bào)的心跳，那么及時(shí)做出處理，我們這里就是簡(jiǎn)單的將其連接斷開，并修改數(shù)據(jù)庫(kù)中相應(yīng)賬戶的在線狀態(tài)。
現(xiàn)在開始說IdleStateHandler，第一個(gè)參數(shù)叫讀超時(shí)時(shí)間，第二個(gè)參數(shù)叫寫超時(shí)時(shí)間，第三個(gè)參數(shù)叫讀寫超時(shí)時(shí)間，第四個(gè)參數(shù)時(shí)時(shí)間單位秒。這個(gè)handler表達(dá)的意思是當(dāng)25秒內(nèi)沒讀到客戶端的消息，或者20秒內(nèi)沒往客戶端發(fā)消息，就會(huì)產(chǎn)生一個(gè)超時(shí)事件。那么這個(gè)超時(shí)事件我們?cè)搶?duì)他做什么處理呢，請(qǐng)看下一條。
2）HeartBeatHandler。結(jié)合a）一起看，當(dāng)發(fā)生超時(shí)事件時(shí)，HeartBeatHandler會(huì)收到這個(gè)事件，并對(duì)它做出處理：第一將鏈接斷開；第二講數(shù)據(jù)庫(kù)中相應(yīng)的賬戶更新為不在線狀態(tài)。
public class HeartBeatHandler extends ChannelInboundHandlerAdapter {
private static Logger logger = LoggerFactory.getLogger(HeartBeatHandler.class);
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent){
IdleStateEvent event = (IdleStateEvent)evt;
if (event.state() == IdleState.READER_IDLE) {
//讀超時(shí)，應(yīng)將連接斷掉
InetSocketAddress socketAddress = (InetSocketAddress)ctx.channel().remoteAddress();
String ip = socketAddress.getAddress().getHostAddress();
ctx.channel().disconnect();
logger.info("【{}】連接超時(shí)，斷開",ip);
String userName = SessionManager.removeSession(ctx.channel());
SpringContextUtil.getBean(UserService.class).updateOnlineStatus(userName,Boolean.FALSE);
}else {
super.userEventTriggered(ctx, evt);
}
}else {
super.userEventTriggered(ctx, evt);
}
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof HeartBeat){
//收到心跳包，不處理
logger.info("server收到心跳包：{}",msg);
return;
}
super.channelRead(ctx, msg);
}
}
3）StringLengthFieldDecoder。這是個(gè)入站handler，他的作用就是解決上面提到的粘包問題：
public class StringLengthFieldDecoder extends LengthFieldBasedFrameDecoder {
public StringLengthFieldDecoder() {
super(10*1024*1024,0,8,0,8);
}
@Override
protected long getUnadjustedFrameLength(ByteBuf buf, int offset, int length, ByteOrder order) {
buf = buf.order(order);
byte[] lenByte = new byte[length];
buf.getBytes(offset, lenByte);
String lenStr = new String(lenByte);
Long len = Long.valueOf(lenStr);
return len;
}
}
只需要集成Netty提供的LengthFieldBasedFrameDecoder 類，并重寫getUnadjustedFrameLength方法即可。
首先看構(gòu)造方法中的5個(gè)參數(shù)。第一個(gè)表示能處理的包的最大長(zhǎng)度；第二三個(gè)參數(shù)應(yīng)該結(jié)合起來理解，表示長(zhǎng)度字段從第幾位開始，長(zhǎng)度的長(zhǎng)度是多少，也就是上面報(bào)文格式協(xié)議中的頭8個(gè)字節(jié)；第四個(gè)參數(shù)表示長(zhǎng)度是否需要校正，舉例理解，比如頭8個(gè)字節(jié)解析出來的長(zhǎng)度=包體長(zhǎng)度+頭8個(gè)字節(jié)的長(zhǎng)度，那么這里就需要校正8個(gè)字節(jié)，我們的協(xié)議中長(zhǎng)度只包含報(bào)文體，因此這個(gè)參數(shù)填0；最后一個(gè)參數(shù)，表示接收到的報(bào)文是否要跳過一些字節(jié)，本例中設(shè)置為8，表示跳過頭8個(gè)字節(jié)，因此經(jīng)過這個(gè)handler后，我們收到的數(shù)據(jù)就只有報(bào)文本身了，不再包含8個(gè)長(zhǎng)度字節(jié)了。
再看getUnadjustedFrameLength方法，其實(shí)就是將頭8個(gè)字符串型的長(zhǎng)度為轉(zhuǎn)換成long型。重寫完這個(gè)方法后，Netty就知道如何收一個(gè)“完整”的數(shù)據(jù)包了。
4）StringDecoder。這個(gè)是Netty自帶的入站handler，會(huì)將字節(jié)流以指定的編碼解析成String。
5）JsonDecoder。是我們自定義的一個(gè)入站handler，目的是將json String轉(zhuǎn)換成java bean，以方便后續(xù)處理：
public class JsonDecoder extends MessageToMessageDecoder<String> {
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, String o, List<Object> list) throws Exception {
Message msg = MessageEnDeCoder.decode(o);
list.add(msg);
}
}
這里會(huì)調(diào)用我們自定義的一個(gè)編解碼幫助類進(jìn)行轉(zhuǎn)換：
public static Message decode(String message){
if (StringUtils.isEmpty(message) || message.length() < 2){
return null;
}
String type = message.substring(0,2);
message = message.substring(2);
if (type.equals(LoginRequest)){
return JsonUtil.jsonToObject(message,LoginRequest.class);
}else if (type.equals(LoginResponse)){
return JsonUtil.jsonToObject(message,LoginResponse.class);
}else if (type.equals(LogoutRequest)){
return JsonUtil.jsonToObject(message,LogoutRequest.class);
}else if (type.equals(LogoutResponse)){
return JsonUtil.jsonToObject(message,LogoutResponse.class);
}else if (type.equals(SendMsgRequest)){
return JsonUtil.jsonToObject(message,SendMsgRequest.class);
}else if (type.equals(SendMsgResponse)){
return JsonUtil.jsonToObject(message,SendMsgResponse.class);
}else if (type.equals(HeartBeat)){
return JsonUtil.jsonToObject(message,HeartBeat.class);
}
return null;
}
6）BussMessageHandler。先看這個(gè)入站handler，是我們的一個(gè)業(yè)務(wù)處理主入口，他的主要工作就是將消息分發(fā)給線程池去處理，另外還負(fù)載一個(gè)小場(chǎng)景，當(dāng)客戶端主動(dòng)斷開時(shí)，需要將相應(yīng)的賬戶數(shù)據(jù)庫(kù)中狀態(tài)更新為不在線。
public class BussMessageHandler extends ChannelInboundHandlerAdapter {
private static Logger logger = LoggerFactory.getLogger(BussMessageHandler.class);
@Autowired
private TaskDispatcher taskDispatcher;
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
logger.info("收到消息：{}",msg);
if (msg instanceof Message){
taskDispatcher.submit(ctx.channel(),(Message)msg);
}
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
//客戶端連接斷開
InetSocketAddress socketAddress = (InetSocketAddress)ctx.channel().remoteAddress();
String ip = socketAddress.getAddress().getHostAddress();
logger.info("客戶端斷開：{}",ip);
String userName = SessionManager.removeSession(ctx.channel());
SpringContextUtil.getBean(UserService.class).updateOnlineStatus(userName,Boolean.FALSE);
super.channelInactive(ctx);
}
}
接下來還差線程池的處理邏輯，也非常簡(jiǎn)單，就是將任務(wù)封裝成executor然后交給線程池處理：
public class TaskDispatcher {
private ThreadPoolExecutor threadPool;
public TaskDispatcher(){
int corePoolSize = 15;
int maxPoolSize = 50;
int keepAliveSeconds = 30;
int queueCapacity = 1024;
BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(queueCapacity);
this.threadPool = new ThreadPoolExecutor(
corePoolSize, maxPoolSize, keepAliveSeconds, TimeUnit.SECONDS,
queue);
}
public void submit(Channel channel, Message msg){
ExecutorBase executor = null;
String messageType = msg.getMessageType();
if (messageType.equals(MessageEnDeCoder.LoginRequest)){
executor = new LoginExecutor(channel,msg);
}
if (messageType.equalsIgnoreCase(MessageEnDeCoder.SendMsgRequest)){
executor = new SendMsgExecutor(channel,msg);
}
if (executor != null){
this.threadPool.submit(executor);
}
}
}
接下來看一下消息轉(zhuǎn)發(fā)executor是怎么做的：
public class SendMsgExecutor extends ExecutorBase {
private static Logger logger = LoggerFactory.getLogger(SendMsgExecutor.class);
public SendMsgExecutor(Channel channel, Message message) {
super(channel, message);
}
@Override
public void run() {
SendMsgResponse response = new SendMsgResponse();
response.setMessageType(MessageEnDeCoder.SendMsgResponse);
response.setTime(new Date());
SendMsgRequest request = (SendMsgRequest)message;
String recvUserName = request.getRecvUserName();
String sendContent = request.getSendMessage();
Channel recvChannel = SessionManager.getSession(recvUserName);
if (recvChannel != null){
SendMsgRequest sendMsgRequest = new SendMsgRequest();
sendMsgRequest.setTime(new Date());
sendMsgRequest.setMessageType(MessageEnDeCoder.SendMsgRequest);
sendMsgRequest.setRecvUserName(recvUserName);
sendMsgRequest.setSendMessage(sendContent);
sendMsgRequest.setSendUserName(request.getSendUserName());
recvChannel.writeAndFlush(sendMsgRequest).addListener(new GenericFutureListener<Future<? super Void>>() {
@Override
public void operationComplete(Future<? super Void> future) throws Exception {
if (future.isSuccess()){
logger.info("消息轉(zhuǎn)發(fā)成功:{}",sendMsgRequest);
response.setResultCode("0000");
response.setResultMessage(String.format("發(fā)給用戶[%s]消息成功",recvUserName));
channel.writeAndFlush(response);
}else {
logger.error(ExceptionUtils.getStackTrace(future.cause()));
logger.info("消息轉(zhuǎn)發(fā)失敗:{}",sendMsgRequest);
response.setResultCode("9999");
response.setResultMessage(String.format("發(fā)給用戶[%s]消息失敗",recvUserName));
channel.writeAndFlush(response);
}
}
});
}else {
logger.info("用戶{}不在線，消息轉(zhuǎn)發(fā)失敗",recvUserName);
response.setResultCode("9999");
response.setResultMessage(String.format("用戶[%s]不在線",recvUserName));
channel.writeAndFlush(response);
}
}
}
整體邏輯：一獲取要把消息發(fā)給那個(gè)賬號(hào)；二獲取該賬號(hào)對(duì)應(yīng)的連接；三在此連接上發(fā)送消息；四獲取消息發(fā)送結(jié)果，將結(jié)果發(fā)給消息“發(fā)起者”。
下面是登陸處理的executor：
public class LoginExecutor extends ExecutorBase {
private static Logger logger = LoggerFactory.getLogger(LoginExecutor.class);
public LoginExecutor(Channel channel, Message message) {
super(channel, message);
}
@Override
public void run() {
LoginRequest request = (LoginRequest)message;
String userName = request.getUserName();
String password = request.getPassword();
UserService userService = SpringContextUtil.getBean(UserService.class);
boolean check = userService.checkLogin(userName,password);
LoginResponse response = new LoginResponse();
response.setUserName(userName);
response.setMessageType(MessageEnDeCoder.LoginResponse);
response.setTime(new Date());
response.setResultCode(check?"0000":"9999");
response.setResultMessage(check?"登陸成功":"登陸失敗，用戶名或密碼錯(cuò)");
if (check){
userService.updateOnlineStatus(userName,Boolean.TRUE);
SessionManager.addSession(userName,channel);
}
channel.writeAndFlush(response).addListener(new GenericFutureListener<Future<? super Void>>() {
@Override
public void operationComplete(Future<? super Void> future) throws Exception {
//登陸失敗，斷開連接
if (!check){
logger.info("用戶{}登陸失敗，斷開連接",((LoginRequest) message).getUserName());
channel.disconnect();
}
}
});
}
}
登陸邏輯也不復(fù)雜，登陸成功則更新用戶在線狀態(tài)，并且無(wú)論登陸成功還是失敗，都會(huì)返一個(gè)登陸應(yīng)答。同時(shí)，如果登陸校驗(yàn)失敗，在返回應(yīng)答成功后，需要將鏈接斷開。
7）JsonEncoder。最后看這個(gè)唯一的出站handler，服務(wù)端發(fā)出去的消息都會(huì)被出站handler處理，他的職責(zé)就是將java bean轉(zhuǎn)成我們之前定義的報(bào)文協(xié)議格式：
public class JsonEncoder extends MessageToByteEncoder<Message> {
@Override
protected void encode(ChannelHandlerContext channelHandlerContext, Message message, ByteBuf byteBuf) throws Exception {
String msgStr = MessageEnDeCoder.encode(message);
int length = msgStr.getBytes(Charset.forName("UTF-8")).length;
String str = String.valueOf(length);
String lenStr = StringUtils.leftPad(str,8,'0');
msgStr = lenStr + msgStr;
byteBuf.writeBytes(msgStr.getBytes("UTF-8"));
}
}
8）SessionManager。剩下最后一個(gè)東西沒說，這個(gè)是用來保存每個(gè)登陸成功賬戶的鏈接的，底層是個(gè)map，key為用戶賬戶，value為鏈接：
public class SessionManager {
private static ConcurrentHashMap<String,Channel> sessionMap = new ConcurrentHashMap<>();
public static void addSession(String userName,Channel channel){
sessionMap.put(userName,channel);
}
public static String removeSession(String userName){
sessionMap.remove(userName);
return userName;
}
public static String removeSession(Channel channel){
for (String key:sessionMap.keySet()){
if (channel.id().asLongText().equalsIgnoreCase(sessionMap.get(key).id().asLongText())){
sessionMap.remove(key);
return key;
}
}
return null;
}
public static Channel getSession(String userName){
return sessionMap.get(userName);
}
}
到這里，整個(gè)服務(wù)端的邏輯就走完了，是不是，很簡(jiǎn)單呢！
3、聊天客戶端
客戶端中界面相關(guān)的東西是基于JavaFX框架做的，這個(gè)我是第一次用，所以不打算講這塊，怕誤導(dǎo)大家。主要還是講Netty作為客戶端是如何跟服務(wù)端通信的。
按照慣例，還是先貼出主入口：
public void login(String userName,String password) throws Exception {
Bootstrap clientBootstrap = new Bootstrap();
EventLoopGroup clientGroup = new NioEventLoopGroup();
try {
clientBootstrap.group(clientGroup)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,10000);
clientBootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new IdleStateHandler(20, 15, 0, TimeUnit.SECONDS));
ch.pipeline().addLast(new StringLengthFieldDecoder());
ch.pipeline().addLast(new StringDecoder(Charset.forName("UTF-8")));
ch.pipeline().addLast(new JsonDecoder());
ch.pipeline().addLast(new JsonEncoder());
ch.pipeline().addLast(bussMessageHandler);
ch.pipeline().addLast(new HeartBeatHandler());
}
});
ChannelFuture future = clientBootstrap.connect(server,port).sync();
if (future.isSuccess()){
channel = (SocketChannel)future.channel();
LoginRequest request = new LoginRequest();
request.setTime(new Date());
request.setUserName(userName);
request.setPassword(password);
request.setMessageType(MessageEnDeCoder.LoginRequest);
channel.writeAndFlush(request).addListener(new GenericFutureListener<Future<? super Void>>() {
@Override
public void operationComplete(Future<? super Void> future) throws Exception {
if (future.isSuccess()){
logger.info("登陸消息發(fā)送成功");
}else {
logger.info("登陸消息發(fā)送失?。簕}", ExceptionUtils.getStackTrace(future.cause()));
Platform.runLater(new Runnable() {
@Override
public void run() {
LoginController.setLoginResult("網(wǎng)絡(luò)錯(cuò)誤，登陸消息發(fā)送失敗");
}
});
}
}
});
}else {
clientGroup.shutdownGracefully();
throw new RuntimeException("網(wǎng)絡(luò)錯(cuò)誤");
}
}catch (Exception e){
clientGroup.shutdownGracefully();
throw new RuntimeException("網(wǎng)絡(luò)錯(cuò)誤");
}
}
對(duì)這段代碼，我們主要關(guān)注這幾點(diǎn)：一所有handler的初始化；二connect服務(wù)端。
所有handler中，除了bussMessageHandler是客戶端特有的外，其他的handler在服務(wù)端章節(jié)已經(jīng)講過了，不再贅述。
1）先看連接服務(wù)端的操作。首先發(fā)起連接，連接成功后發(fā)送登陸報(bào)文。發(fā)起連接需要對(duì)成功和失敗進(jìn)行處理。發(fā)送登陸報(bào)文也需要對(duì)成功和失敗進(jìn)行處理。注意，這里的成功失敗只是代表當(dāng)前操作的網(wǎng)絡(luò)層面的成功失敗，這時(shí)候并不能獲取服務(wù)端返回的應(yīng)答中的業(yè)務(wù)層面的成功失敗，如果不理解這句話，可以翻看前面講過的“異步”相關(guān)內(nèi)容。
2）BussMessageHandler。整體流程還是跟服務(wù)端一樣，將受到的消息扔給線程池處理，我們直接看處理消息的各個(gè)executor。
先看客戶端發(fā)出登陸請(qǐng)求后，收到登陸應(yīng)答消息后是怎么處理的（這段代碼可以結(jié)合1）的內(nèi)容一起理解）：
public class LoginRespExecutor extends ExecutorBase {
private static Logger logger = LoggerFactory.getLogger(LoginRespExecutor.class);
public LoginRespExecutor(Channel channel, Message message) {
super(channel, message);
}
@Override
public void run() {
LoginResponse response = (LoginResponse)message;
logger.info("登陸結(jié)果：{}->{}",response.getResultCode(),response.getResultMessage());
if (!response.getResultCode().equals("0000")){
Platform.runLater(new Runnable() {
@Override
public void run() {
LoginController.setLoginResult("登陸失敗，用戶名或密碼錯(cuò)誤");
}
});
}else {
LoginController.setCurUserName(response.getUserName());
ClientApplication.getScene().setRoot(SpringContextUtil.getBean(MainView.class).getView());
}
}
}
接下來看客戶端是怎么發(fā)聊天信息的：
public void sendMessage(Message message) {
channel.writeAndFlush(message).addListener(new GenericFutureListener<Future<? super Void>>() {
@Override
public void operationComplete(Future<? super Void> future) throws Exception {
SendMsgRequest send = (SendMsgRequest)message;
if (future.isSuccess()){
Platform.runLater(new Runnable() {
@Override
public void run() {
MainController.setMessageHistory(String.format("[我]在[%s]發(fā)給[%s]的消息[%s]，發(fā)送成功",
DateFormatUtils.format(send.getTime(),"yyyy-MM-dd HH:mm:ss"),send.getRecvUserName(),send.getSendMessage()));
}
});
}else {
Platform.runLater(new Runnable() {
@Override
public void run() {
MainController.setMessageHistory(String.format("[我]在[%s]發(fā)給[%s]的消息[%s]，發(fā)送失敗",
DateFormatUtils.format(send.getTime(),"yyyy-MM-dd HH:mm:ss"),send.getRecvUserName(),send.getSendMessage()));
}
});
}
}
});
}
實(shí)際上，到這里通信相關(guān)的代碼已經(jīng)貼完了。剩下的都是界面處理相關(guān)的代碼，不再貼了。
客戶端，是不是，非常簡(jiǎn)單！
4、Web管理端
Web管理端可以說是更沒任何技術(shù)含量，就是Shiro登陸認(rèn)證、列表增刪改查。增刪改沒什么好說的，下面重點(diǎn)說一下Shiro登陸和列表查詢。
1）Shiro登陸
首先定義一個(gè)Realm，至于這是什么概念，自行百度吧，這里并不是本文重點(diǎn)：
public class UserDbRealm extends AuthorizingRealm {
@Override
protected AuthorizationInfo doGetAuthorizationInfo(PrincipalCollection principalCollection) {
return null;
}
@Override
protected AuthenticationInfo doGetAuthenticationInfo(AuthenticationToken authenticationToken) throws AuthenticationException {
RequestAttributes attributes = RequestContextHolder.getRequestAttributes();
UsernamePasswordToken upToken = (UsernamePasswordToken) authenticationToken;
String username = upToken.getUsername();
String password = "";
if (upToken.getPassword() != null)
{
password = new String(upToken.getPassword());
}
// TODO: 2021/5/13 校驗(yàn)用戶名密碼，不通過則拋認(rèn)證異常即可
ShiroUser user = new ShiroUser();
SimpleAuthenticationInfo info = new SimpleAuthenticationInfo(user, password, getName());
return info;
}
}
接下來把這個(gè)Realm注冊(cè)成Spring Bean，同時(shí)定義過濾鏈：
@Bean
public Realm realm() {
UserDbRealm realm = new UserDbRealm();
realm.setAuthorizationCachingEnabled(true);
realm.setCacheManager(cacheManager());
return realm;
}
@Bean
public ShiroFilterChainDefinition shiroFilterChainDefinition() {
DefaultShiroFilterChainDefinition chainDefinition = new DefaultShiroFilterChainDefinition();
chainDefinition.addPathDefinition("/css/**", "anon");
chainDefinition.addPathDefinition("/img/**", "anon");
chainDefinition.addPathDefinition("/js/**", "anon");
chainDefinition.addPathDefinition("/logout", "logout");
chainDefinition.addPathDefinition("/login", "anon");
chainDefinition.addPathDefinition("/captchaImage", "anon");
chainDefinition.addPathDefinition("/**", "authc");
return chainDefinition;
}
到現(xiàn)在為止，Shiro配置好了，下面看如何調(diào)起登陸：
@PostMapping("/login")
@ResponseBody
public Result<String> login(String username, String password, Boolean rememberMe)
{
Result<String> ret = new Result<>();
UsernamePasswordToken token = new UsernamePasswordToken(username, password);
Subject subject = SecurityUtils.getSubject();
try
{
subject.login(token);
return ret;
}
catch (AuthenticationException e)
{
String msg = "用戶或密碼錯(cuò)誤";
if (StringUtils.isNotEmpty(e.getMessage()))
{
msg = e.getMessage();
}
ret.setCode(Result.FAIL);
ret.setMessage(msg);
return ret;
}
}
登陸代碼就這么愉快的完成了。
2）列表查詢
查是個(gè)很簡(jiǎn)單的操作，但是卻是所有web系統(tǒng)中使用最頻繁的操作。因此，做一個(gè)通用性的封裝，非常有必要。以下代碼不做過多講解，初級(jí)工程師到高級(jí)工程師，就差這段代碼了（手動(dòng)捂臉）：
a）Controller
@RequestMapping("/query")
@ResponseBody
public Result<Page<User>> query(@RequestParam Map<String,Object> params, String sort, String order, Integer pageIndex, Integer pageSize){
Page<User> page = userService.query(params,sort,order,pageIndex,pageSize);
Result<Page<User>> ret = new Result<>();
ret.setData(page);
return ret;
}
b）Service
@Autowired
private UserDao userDao;
@Autowired
private QueryService queryService;
public Page<User> query(Map<String,Object> params, String sort, String order, Integer pageIndex, Integer pageSize){
return queryService.query(userDao,params,sort,order,pageIndex,pageSize);
}
public class QueryService {
public <T> com.easy.okim.common.model.Page<T> query(JpaSpecificationExecutor<T> dao, Map<String,Object> filters, String sort, String order, Integer pageIndex, Integer pageSize){
com.easy.okim.common.model.Page<T> ret = new com.easy.okim.common.model.Page<T>();
Map<String,Object> params = new HashMap<>();
if (filters != null){
filters.remove("sort");
filters.remove("order");
filters.remove("pageIndex");
filters.remove("pageSize");
for (String key:filters.keySet()){
Object value = filters.get(key);
if (value != null && StringUtils.isNotEmpty(value.toString())){
params.put(key,value);
}
}
}
Pageable pageable = null;
pageIndex = pageIndex - 1;
if (StringUtils.isEmpty(sort)){
pageable = PageRequest.of(pageIndex,pageSize);
}else {
Sort s = Sort.by(Sort.Direction.ASC,sort);
if (StringUtils.isNotEmpty(order) && order.equalsIgnoreCase("desc")){
s = Sort.by(Sort.Direction.DESC,sort);
}
pageable = PageRequest.of(pageIndex,pageSize,s);
}
Page<T> page = null;
if (params.size() ==0){
page = dao.findAll(null,pageable);
}else {
Specification<T> specification = new Specification<T>() {
@Override
public Predicate toPredicate(Root<T> root, CriteriaQuery<?> criteriaQuery, CriteriaBuilder builder) {
List<Predicate> predicates = new ArrayList<>();
for (String filter : params.keySet()) {
Object value = params.get(filter);
if (value == null || StringUtils.isEmpty(value.toString())) {
continue;
}
String field = filter;
String operator = "=";
String[] arr = filter.split("\\|");
if (arr.length == 2) {
field = arr[0];
operator = arr[1];
}
if (arr.length == 3) {
field = arr[0];
operator = arr[1];
String type = arr[2];
if (type.equalsIgnoreCase("boolean")){
value = Boolean.parseBoolean(value.toString());
}else if (type.equalsIgnoreCase("integer")){
value = Integer.parseInt(value.toString());
}else if (type.equalsIgnoreCase("long")){
value = Long.parseLong(value.toString());
}
}
String[] names = StringUtils.split(field, ".");
Path expression = root.get(names[0]);
for (int i = 1; i < names.length; i++) {
expression = expression.get(names[i]);
}
// logic operator
switch (operator) {
case "=":
predicates.add(builder.equal(expression, value));
break;
case "!=":
predicates.add(builder.notEqual(expression, value));
break;
case "like":
predicates.add(builder.like(expression, "%" + value + "%"));
break;
case ">":
predicates.add(builder.greaterThan(expression, (Comparable) value));
break;
case "<":
predicates.add(builder.lessThan(expression, (Comparable) value));
break;
case ">=":
predicates.add(builder.greaterThanOrEqualTo(expression, (Comparable) value));
break;
case "<=":
predicates.add(builder.lessThanOrEqualTo(expression, (Comparable) value));
break;
case "isnull":
predicates.add(builder.isNull(expression));
break;
case "isnotnull":
predicates.add(builder.isNotNull(expression));
break;
case "in":
CriteriaBuilder.In in = builder.in(expression);
String[] arr1 = StringUtils.split(filter.toString(), ",");
for (String e : arr1) {
in.value(e);
}
predicates.add(in);
break;
}
}
// 將所有條件用 and 聯(lián)合起來
if (!predicates.isEmpty()) {
return builder.and(predicates.toArray(new Predicate[predicates.size()]));
}
return builder.conjunction();
}
};
page = dao.findAll(specification,pageable);
}
ret.setTotal(page.getTotalElements());
ret.setRows(page.getContent());
return ret;
}
}
c）Dao
public interface UserDao extends JpaRepository<User,Long>,JpaSpecificationExecutor<User> {
//啥都不用寫，繼承Spring Data Jpa提供的類就行了
}
六、結(jié)語(yǔ)
雖然標(biāo)題起的有些嘩眾取寵了，但內(nèi)容也確實(shí)都是實(shí)實(shí)在在的干貨，希望本文能對(duì)大家有一些幫助，源代碼工程不打算貼了，希望你能跟著文章自己手敲一遍。
鋒哥最新SpringCloud分布式電商秒殺課程發(fā)布
??????
??長(zhǎng)按上方微信二維碼 2 秒
感謝點(diǎn)贊支持下哈 