Jack Jiang

          我的最新工程MobileIMSDK:http://git.oschina.net/jackjiang/MobileIMSDK
          posts - 499, comments - 13, trackbacks - 0, articles - 1

          本文作者芋艿,原題“使用 Netty 實現(xiàn) IM 聊天賊簡單”,本底價有修訂和改動。

          一、本文引言

          上篇《跟著源碼學IM(七):手把手教你用WebSocket打造Web端IM聊天》中,我們使用 WebSocket 實現(xiàn)了一個簡單的 IM 功能,支持身份認證、私聊消息、群聊消息。

          然后就有人發(fā)私信,希望使用純 Netty 實現(xiàn)一個類似的功能,因此就有了本文。

          注:源碼請從同步鏈接附件中下載,http://www.52im.net/thread-3489-1-1.html。

          學習交流:

          - 即時通訊/推送技術開發(fā)交流5群:215477170 [推薦]

          - 移動端IM開發(fā)入門文章:《新手入門一篇就夠:從零開發(fā)移動端IM

          - 開源IM框架源碼:https://github.com/JackJiang2011/MobileIMSDK

          本文同步發(fā)布于:http://www.52im.net/thread-3489-1-1.html

          二、知識準備

          可能有人不知道 Netty 是什么,這里簡單介紹下:

          Netty 是一個 Java 開源框架。Netty 提供異步的、事件驅動的網(wǎng)絡應用程序框架和工具,用以快速開發(fā)高性能、高可靠性的網(wǎng)絡服務器和客戶端程序。

          也就是說,Netty 是一個基于 NIO 的客戶、服務器端編程框架,使用Netty 可以確保你快速和簡單的開發(fā)出一個網(wǎng)絡應用,例如實現(xiàn)了某種協(xié)議的客戶,服務端應用。

          Netty 相當簡化和流線化了網(wǎng)絡應用的編程開發(fā)過程,例如,TCP 和 UDP 的 Socket 服務開發(fā)。

          以下是幾篇有關Netty的入門文章,值得一讀:

          如果你連Java的NIO都不知道是什么,下面的文章建議優(yōu)先讀一下:

          Netty源碼和API的在線閱讀地址:

          三、本文源碼

          本文完整代碼附件下載:請從同步鏈接附件中下載,http://www.52im.net/thread-3489-1-1.html。

          源碼的目錄結構,如下圖所示: 

          如上圖所示:

          • 1)lab-67-netty-demo-server 項目:搭建 Netty 服務端;
          • 2)lab-67-netty-demo-client 項目:搭建 Netty 客戶端;
          • 3)lab-67-netty-demo-common 項目:提供 Netty 的基礎封裝,提供消息的編解碼、分發(fā)的功能。

          另外,源碼中也會提供 Netty 常用功能的示例:

          • 1)心跳機制,實現(xiàn)服務端對客戶端的存活檢測;
          • 2)斷線重連,實現(xiàn)客戶端對服務端的重新連接。

          不嗶嗶,直接開干。

          五、通信協(xié)議

          在上一章中,我們實現(xiàn)了客戶端和服務端的連接功能。而本小節(jié),我們要讓它們兩能夠說上話,即進行數(shù)據(jù)的讀寫。

          在日常項目的開發(fā)中,前端和后端之間采用 HTTP 作為通信協(xié)議,使用文本內(nèi)容進行交互,數(shù)據(jù)格式一般是 JSON。但是在 TCP 的世界里,我們需要自己基于二進制構建,構建客戶端和服務端的通信協(xié)議。

          我們以客戶端向服務端發(fā)送消息來舉個例子,假設客戶端要發(fā)送一個登錄請求。

          對應的類如下:

          public class AuthRequest {

              /** 用戶名 **/

              private String username;

              /** 密碼 **/

              private String password;

          }

          顯然:我們無法將一個 Java 對象直接丟到 TCP Socket 當中,而是需要將其轉換成 byte 字節(jié)數(shù)組,才能寫入到 TCP Socket 中去。即,需要將消息對象通過序列化,轉換成 byte 字節(jié)數(shù)組。

          同時:在服務端收到 byte 字節(jié)數(shù)組時,需要將其又轉換成 Java 對象,即反序列化。不然,服務端對著一串 byte 字節(jié)處理個毛線?!

          友情提示:服務端向客戶端發(fā)消息,也是一樣的過程哈!

          序列化的工具非常多,例如說 Google 提供的 Protobuf,性能高效,且序列化出來的二進制數(shù)據(jù)較小。Netty 對 Protobuf 進行集成,提供了相應的編解碼器。

          如下圖所示: 

          但是考慮到很多可能對 Protobuf 并不了解,因為它實現(xiàn)序列化又增加額外學習成本。因此,仔細一個捉摸,還是采用 JSON 方式進行序列化??赡苡腥藭苫螅琂SON 不是將對象轉換成字符串嗎?嘿嘿,我們再把字符串轉換成 byte 字節(jié)數(shù)組就可以啦~

          下面,我們新建 lab-67-netty-demo-common 項目,并在 codec 包下,實現(xiàn)我們自定義的通信協(xié)議。

          如下圖所示:

          5.1、Invocation

          創(chuàng)建 Invocation 類,通信協(xié)議的消息體。

          代碼如下:

          /**

           * 通信協(xié)議的消息體

           */

          public class Invocation {

              /**

               * 類型

               */

              private String type;

              /**

               * 消息,JSON 格式

               */

              private String message;

           

              // 空構造方法

              public Invocation() {

              }

           

              public Invocation(String type, String message) {

                  this.type = type;

                  this.message = message;

              }

           

              public Invocation(String type, Message message) {

                  this.type = type;

                  this.message = JSON.toJSONString(message);

              }

           

              // ... 省略 setter、getter、toString 方法

          }

           type 屬性,類型,用于匹配對應的消息處理器。如果類比 HTTP 協(xié)議,type 屬性相當于請求地址。

           message 屬性,消息內(nèi)容,使用 JSON 格式。

          另外,Message 是我們定義的消息接口,代碼如下:

          public interface Message {

              // ... 空,作為標記接口

          }

          5.2、粘包與拆包

          在開始看 Invocation 的編解碼處理器之前,我們先了解下粘包與拆包的概念。

          5.2.1 產(chǎn)生原因

          產(chǎn)生粘包和拆包問題的主要原因是,操作系統(tǒng)在發(fā)送 TCP 數(shù)據(jù)的時候,底層會有一個緩沖區(qū),例如 1024 個字節(jié)大小。

          如果一次請求發(fā)送的數(shù)據(jù)量比較小,沒達到緩沖區(qū)大小,TCP 則會將多個請求合并為同一個請求進行發(fā)送,這就形成了粘包問題。

          例如說:在《詳解 Socket 編程 --- TCP_NODELAY 選項》文章中我們可以看到,在關閉 Nagle 算法時,請求不會等待滿足緩沖區(qū)大小,而是盡快發(fā)出,降低延遲。

          如果一次請求發(fā)送的數(shù)據(jù)量比較大,超過了緩沖區(qū)大小,TCP 就會將其拆分為多次發(fā)送,這就是拆包,也就是將一個大的包拆分為多個小包進行發(fā)送。

          如下圖展示了粘包和拆包的一個示意圖,演示了粘包和拆包的三種情況: 

          如上圖所示:

          • 1)A 和 B 兩個包都剛好滿足 TCP 緩沖區(qū)的大小,或者說其等待時間已經(jīng)達到 TCP 等待時長,從而還是使用兩個獨立的包進行發(fā)送;
          • 2)A 和 B 兩次請求間隔時間內(nèi)較短,并且數(shù)據(jù)包較小,因而合并為同一個包發(fā)送給服務端;
          • 3)B 包比較大,因而將其拆分為兩個包 B_1 和 B_2 進行發(fā)送,而這里由于拆分后的 B_2 比較小,其又與 A 包合并在一起發(fā)送。

          5.2.2 解決方案

          對于粘包和拆包問題,常見的解決方案有三種。

           客戶端在發(fā)送數(shù)據(jù)包的時候,每個包都固定長度。比如 1024 個字節(jié)大小,如果客戶端發(fā)送的數(shù)據(jù)長度不足 1024 個字節(jié),則通過補充空格的方式補全到指定長度。

          這種方式,暫時沒有找到采用這種方式的案例。

           客戶端在每個包的末尾使用固定的分隔符。例如 \r\n,如果一個包被拆分了,則等待下一個包發(fā)送過來之后找到其中的 \r\n,然后對其拆分后的頭部部分與前一個包的剩余部分進行合并,這樣就得到了一個完整的包。具體的案例,有 HTTP、WebSocket、Redis。

           將消息分為頭部和消息體,在頭部中保存有當前整個消息的長度,只有在讀取到足夠長度的消息之后才算是讀到了一個完整的消息。

          友情提示:方案 ③ 是 ① 的升級版,動態(tài)長度。

          本文將采用這種方式,在每次 Invocation 序列化成字節(jié)數(shù)組寫入 TCP Socket 之前,先將字節(jié)數(shù)組的長度寫到其中。

          如下圖所示: 

          5.3、InvocationEncoder

          創(chuàng)建 InvocationEncoder 類,實現(xiàn)將 Invocation 序列化,并寫入到 TCP Socket 中。

          代碼如下:

          public class InvocationEncoder extends MessageToByteEncoder<Invocation> {

           

              private Logger logger = LoggerFactory.getLogger(getClass());

           

              @Override

              protected void encode(ChannelHandlerContext ctx, Invocation invocation, ByteBuf out) {

                  // <2.1> 將 Invocation 轉換成 byte[] 數(shù)組

                  byte[] content = JSON.toJSONBytes(invocation);

                  // <2.2> 寫入 length

                  out.writeInt(content.length);

                  // <2.3> 寫入內(nèi)容

                  out.writeBytes(content);

                  logger.info("[encode][連接({}) 編碼了一條消息({})]", ctx.channel().id(), invocation.toString());

              }

          }

           MessageToByteEncoder 是 Netty 定義的編碼 ChannelHandler 抽象類,將泛型 消息轉換成字節(jié)數(shù)組。

           #encode(ChannelHandlerContext ctx, Invocation invocation, ByteBuf out) 方法,進行編碼的邏輯。

          <2.1> 處,調(diào)用 JSON 的 #toJSONBytes(Object object, SerializerFeature... features) 方法,將 Invocation 轉換成 字節(jié)數(shù)組。

          <2.2> 處,將字節(jié)數(shù)組的長度,寫入到 TCP Socket 當中。這樣,后續(xù)「5.4 InvocationDecoder」可以根據(jù)該長度,解析到消息,解決粘包和拆包的問題。

          友情提示:MessageToByteEncoder 會最終將 ByteBuf out 寫到 TCP Socket 中。

          <2.3> 處,將字節(jié)數(shù)組,寫入到 TCP Socket 當中。

          5.4、InvocationDecoder

          創(chuàng)建 InvocationDecoder 類,實現(xiàn)從 TCP Socket 讀取字節(jié)數(shù)組,反序列化成 Invocation。

          代碼如下: 

           ByteToMessageDecoder 是 Netty 定義的解碼 ChannelHandler 抽象類,在 TCP Socket 讀取到新數(shù)據(jù)時,觸發(fā)進行解碼。

           在 <2.1>、<2.2>、<2.3> 處,從 TCP Socket 中讀取長度。

           在 <3.1>、<3.2>、<3.3> 處,從 TCP Socket 中讀取字節(jié)數(shù)組,并反序列化成 Invocation 對象。

          最終,添加 List<Object> out 中,交給后續(xù)的 ChannelHandler 進行處理。稍后,我們將在「6. 消息分發(fā)」小結中,會看到 MessageDispatcher 將 Invocation 分發(fā)到其對應的 MessageHandler 中,進行業(yè)務邏輯的執(zhí)行。

          5.5、引入依賴

          創(chuàng)建 pom.xml 文件,引入 Netty、FastJSON 等等依賴。

          5.6、本章小結

          至此,我們已經(jīng)完成通信協(xié)議的定義、編解碼的邏輯,是不是蠻有趣的?!

          另外,我們在 NettyServerHandlerInitializer 和 NettyClientHandlerInitializer 的初始化代碼中,將編解碼器添加到其中。

          如下圖所示: 

          六、消息分發(fā)

          在 SpringMVC 中,DispatcherServlet 會根據(jù)請求地址、方法等,將請求分發(fā)到匹配的 Controller 的 Method 方法上。

          在 lab-67-netty-demo-client 項目的 dispatcher 包中,我們創(chuàng)建了 MessageDispatcher 類,實現(xiàn)和 DispatcherServlet 類似的功能,將 Invocation 分發(fā)到其對應的 MessageHandler 中,進行業(yè)務邏輯的執(zhí)行。 

          下面,我們來看看具體的代碼實現(xiàn)。

          6.1、Message

          創(chuàng)建 Message 接口,定義消息的標記接口。

          代碼如下:

          public interface Message {

          }

          下圖,是我們涉及到的 Message 實現(xiàn)類。

          如下圖所示:

          6.2、MessageHandler

          創(chuàng)建 MessageHandler 接口,消息處理器接口。

          代碼如下:

          public interface MessageHandler<T extendsMessage> {

              /**

               * 執(zhí)行處理消息

               *

               * @param channel 通道

               * @param message 消息

               */

              voide xecute(Channel channel, T message);

           

              /**

               * @return 消息類型,即每個 Message 實現(xiàn)類上的 TYPE 靜態(tài)字段

               */

              String getType();

          }

          如上述代碼所示:

          • 1)定義了泛型 <T> ,需要是 Message 的實現(xiàn)類;
          • 2)定義的兩個接口方法,自己看下注釋哈。

          下圖,是我們涉及到的 MessageHandler 實現(xiàn)類。

          如下圖所示: 

          6.3、MessageHandlerContainer

          創(chuàng)建 MessageHandlerContainer 類,作為 MessageHandler 的容器。

          代碼如下: 

           實現(xiàn) InitializingBean 接口,在 #afterPropertiesSet() 方法中,掃描所有 MessageHandler Bean ,添加到 MessageHandler 集合中。

           在 #getMessageHandler(String type) 方法中,獲得類型對應的 MessageHandler 對象。稍后,我們會在 MessageDispatcher 調(diào)用該方法。

           在 #getMessageClass(MessageHandler handler) 方法中,通過 MessageHandler 中,通過解析其類上的泛型,獲得消息類型對應的 Class 類。這是參考 rocketmq-spring 項目的 DefaultRocketMQListenerContainer#getMessageType() 方法,進行略微修改。

          6.4、MessageDispatcher

          創(chuàng)建 MessageDispatcher 類,將 Invocation 分發(fā)到其對應的 MessageHandler 中,進行業(yè)務邏輯的執(zhí)行。

          代碼如下:

          @ChannelHandler.Sharable

          public class MessageDispatcher extends SimpleChannelInboundHandler<Invocation> {

           

              @Autowired

              private MessageHandlerContainer messageHandlerContainer;

           

              private final ExecutorService executor =  Executors.newFixedThreadPool(200);

           

              @Override

              protected void channelRead0(ChannelHandlerContext ctx, Invocation invocation) {

                  // <3.1> 獲得 type 對應的 MessageHandler 處理器

                  MessageHandler messageHandler = messageHandlerContainer.getMessageHandler(invocation.getType());

                  // 獲得  MessageHandler 處理器的消息類

                  Class<? extendsMessage> messageClass = MessageHandlerContainer.getMessageClass(messageHandler);

                  // <3.2> 解析消息

                  Message message = JSON.parseObject(invocation.getMessage(), messageClass);

                  // <3.3> 執(zhí)行邏輯

                  executor.submit(newRunnable() {

           

                      @Override

                      public void run() {

                          // noinspection unchecked

                          messageHandler.execute(ctx.channel(), message);

                      }

                  });

              }

          }

          ① 在類上添加 @ChannelHandler.Sharable 注解,標記這個 ChannelHandler 可以被多個 Channel 使用。

          ② SimpleChannelInboundHandler 是 Netty 定義的消息處理 ChannelHandler 抽象類,處理消息的類型是 <I> 泛型時。

          ③ #channelRead0(ChannelHandlerContext ctx, Invocation invocation) 方法,處理消息,進行分發(fā)。

          <3.1> 處,調(diào)用 MessageHandlerContainer 的 #getMessageHandler(String type) 方法,獲得 Invocation 的 type 對應的 MessageHandler 處理器。

          然后,調(diào)用 MessageHandlerContainer 的 #getMessageClass(messageHandler) 方法,獲得  MessageHandler 處理器的消息類。

          <3.2> 處,調(diào)用 JSON 的 ## parseObject(String text, Class<T> clazz) 方法,將 Invocation 的 message 解析成 MessageHandler 對應的消息對象。

          <3.3> 處,丟到線程池中,然后調(diào)用 MessageHandler 的 #execute(Channel channel, T message) 方法,執(zhí)行業(yè)務邏輯。

          注意:為什么要丟到 executor 線程池中呢?我們先來了解下 EventGroup 的線程模型。

          友情提示:在我們啟動 Netty 服務端或者客戶端時,都會設置其 EventGroup。

          EventGroup 我們可以先簡單理解成一個線程池,并且線程池的大小僅僅是 CPU 數(shù)量 * 2。每個 Channel 僅僅會被分配到其中的一個線程上,進行數(shù)據(jù)的讀寫。并且,多個 Channel 會共享一個線程,即使用同一個線程進行數(shù)據(jù)的讀寫。

          那么試著思考下,MessageHandler 的具體邏輯視線中,往往會涉及到 IO 處理,例如說進行數(shù)據(jù)庫的讀取。這樣,就會導致一個 Channel 在執(zhí)行 MessageHandler 的過程中,阻塞了共享當前線程的其它 Channel 的數(shù)據(jù)讀取。

          因此,我們在這里創(chuàng)建了 executor 線程池,進行 MessageHandler 的邏輯執(zhí)行,避免阻塞 Channel 的數(shù)據(jù)讀取。

          可能會有人說,我們是不是能夠把 EventGroup 的線程池設置大一點,例如說 200 呢?對于長連接的 Netty 服務端,往往會有 1000 ~ 100000 的 Netty 客戶端連接上來,這樣無論設置多大的線程池,都會出現(xiàn)阻塞數(shù)據(jù)讀取的情況。

          友情提示:executor 線程池,我們一般稱之為業(yè)務線程池或者邏輯線程池,顧名思義,就是執(zhí)行業(yè)務邏輯的。這樣的設計方式,目前 Dubbo 等等 RPC 框架,都采用這種方式。后續(xù),可以認真閱讀下《【NIO 系列】——之 Reactor 模型》文章,進一步理解。

          6.5、NettyServerConfig

          創(chuàng)建 NettyServerConfig 配置類,創(chuàng)建 MessageDispatcher 和 MessageHandlerContainer Bean。

          代碼如下:

          @Configuration

          public class NettyServerConfig {

           

              @Bean

              public MessageDispatcher messageDispatcher() {

                  return new MessageDispatcher();

              }

           

              @Bean

              public MessageHandlerContainer messageHandlerContainer() {

                  return new MessageHandlerContainer();

              }

          }

          6.6、NettyClientConfig

          創(chuàng)建 NettyClientConfig 配置類,創(chuàng)建 MessageDispatcher 和 MessageHandlerContainer Bean。

          代碼如下:

          @Configuration

          public class NettyClientConfig {

              @Bean

              public MessageDispatcher messageDispatcher() {

                  return new MessageDispatcher();

              }

              @Bean

              public MessageHandlerContainer messageHandlerContainer() {

                  return new MessageHandlerContainer();

              }

          }

          6.7、本章小結

          后續(xù),我們將在如下小節(jié),具體演示消息分發(fā)的使用。

          七、斷開重連

          Netty 客戶端需要實現(xiàn)斷開重連機制,解決各種情況下的斷開情況。

          例如說:

          • 1)Netty 客戶端啟動時,Netty 服務端處于掛掉,導致無法連接上;
          • 2)在運行過程中,Netty 服務端掛掉,導致連接被斷開;
          • 3)任一一端網(wǎng)絡抖動,導致連接異常斷開。

          具體的代碼實現(xiàn)比較簡單,只需要在兩個地方增加重連機制:

          • 1)Netty 客戶端啟動時,無法連接 Netty 服務端時,發(fā)起重連;
          • 2)Netty 客戶端運行時,和 Netty 斷開連接時,發(fā)起重連。

          考慮到重連會存在失敗的情況,我們采用定時重連的方式,避免占用過多資源。

          7.1、具體代碼

          ① 在 NettyClient 中,提供 #reconnect() 方法,實現(xiàn)定時重連的邏輯。

          代碼如下:

          // NettyClient.java

          public void reconnect() {

              eventGroup.schedule(new Runnable() {

                  @Override

                  publicvoidrun() {

                      logger.info("[reconnect][開始重連]");

                      try{

                          start();

                      } catch(InterruptedException e) {

                          logger.error("[reconnect][重連失敗]", e);

                      }

                  }

              }, RECONNECT_SECONDS, TimeUnit.SECONDS);

              logger.info("[reconnect][{} 秒后將發(fā)起重連]", RECONNECT_SECONDS);

          }

          通過調(diào)用 EventLoop 提供的 #schedule(Runnable command, long delay, TimeUnit unit) 方法,實現(xiàn)定時邏輯。而在內(nèi)部的具體邏輯,調(diào)用 NettyClient 的 #start() 方法,發(fā)起連接 Netty 服務端。

          又因為 NettyClient 在 #start() 方法在連接 Netty 服務端失敗時,又會調(diào)用 #reconnect() 方法,從而再次發(fā)起定時重連。如此循環(huán)反復,知道 Netty 客戶端連接上 Netty 服務端。

          如下圖所示: 

          ② 在 NettyClientHandler 中,實現(xiàn) #channelInactive(ChannelHandlerContext ctx) 方法,在發(fā)現(xiàn)和 Netty 服務端斷開時,調(diào)用 Netty Client 的 #reconnect() 方法,發(fā)起重連。

          代碼如下:

          // NettyClientHandler.java

          @Override

          public void channelInactive(ChannelHandlerContext ctx) throws Exception {

              // 發(fā)起重連

              nettyClient.reconnect();

              // 繼續(xù)觸發(fā)事件

              super.channelInactive(ctx);

          }

          7.2、簡單測試

          ① 啟動 Netty Client,不要啟動 Netty Server,控制臺打印日志如下圖: 

          可以看到 Netty Client 在連接失敗時,不斷發(fā)起定時重連。

          ② 啟動 Netty Server,控制臺打印如下圖: 

          可以看到 Netty Client 成功重連上 Netty Server。

          八、心跳機制與空閑檢測

          我們可以了解到 TCP 自帶的空閑檢測機制,默認是 2 小時。這樣的檢測機制,從系統(tǒng)資源層面上來說是可以接受的。

          但是在業(yè)務層面,如果 2 小時才發(fā)現(xiàn)客戶端與服務端的連接實際已經(jīng)斷開,會導致中間非常多的消息丟失,影響客戶的使用體驗。

          因此,我們需要在業(yè)務層面,自己實現(xiàn)空閑檢測,保證盡快發(fā)現(xiàn)客戶端與服務端實際已經(jīng)斷開的情況。

          實現(xiàn)邏輯如下:

          • 1)服務端發(fā)現(xiàn) 180 秒未從客戶端讀取到消息,主動斷開連接;
          • 2)客戶端發(fā)現(xiàn) 180 秒未從服務端讀取到消息,主動斷開連接。

          考慮到客戶端和服務端之間并不是一直有消息的交互,所以我們需要增加心跳機制。

          邏輯如下:

          • 1)客戶端每 60 秒向服務端發(fā)起一次心跳消息,保證服務端可以讀取到消息;
          • 2)服務端在收到心跳消息時,回復客戶端一條確認消息,保證客戶端可以讀取到消息。

          友情提示:

          為什么是 180 秒?可以加大或者減小,看自己希望多快檢測到連接異常。過短的時間,會導致心跳過于頻繁,占用過多資源。

          為什么是 60 秒?三次機會,確認是否心跳超時。

          雖然聽起來有點復雜,但是實現(xiàn)起來并不復雜哈。

          8.1、服務端的空閑檢測

          在 NettyServerHandlerInitializer 中,我們添加了一個 ReadTimeoutHandler 處理器,它在超過指定時間未從對端讀取到數(shù)據(jù),會拋出 ReadTimeoutException 異常。

          如下圖所示:

          通過這樣的方式,實現(xiàn)服務端發(fā)現(xiàn) 180 秒未從客戶端讀取到消息,主動斷開連接。

          8.2、客戶端的空閑檢測

          在 NettyClientHandlerInitializer 中,我們添加了一個 ReadTimeoutHandler 處理器,它在超過指定時間未從對端讀取到數(shù)據(jù),會拋出 ReadTimeoutException 異常。

          如下圖所示: 

          通過這樣的方式,實現(xiàn)客戶端發(fā)現(xiàn) 180 秒未從服務端讀取到消息,主動斷開連接。

          8.3、心跳機制

          Netty 提供了 IdleStateHandler 處理器,提供空閑檢測的功能,在 Channel 的讀或者寫空閑時間太長時,將會觸發(fā)一個 IdleStateEvent 事件。

          這樣,我們只需要在 NettyClientHandler 處理器中,在接收到 IdleStateEvent 事件時,客戶端向客戶端發(fā)送一次心跳消息。

          如下圖所示:

          其中,HeartbeatRequest 是心跳請求。

          同時,我們在服務端項目中,創(chuàng)建了一個 HeartbeatRequestHandler 消息處理器,在收到客戶端的心跳請求時,回復客戶端一條確認消息。

          代碼如下:

          @Component

          public class HeartbeatRequestHandler implementsMessageHandler<HeartbeatRequest> {

              private Logger logger = LoggerFactory.getLogger(getClass());

           

              @Override

              public void execute(Channel channel, HeartbeatRequest message) {

                  logger.info("[execute][收到連接({}) 的心跳請求]", channel.id());

                  // 響應心跳

                  HeartbeatResponse response = newHeartbeatResponse();

                  channel.writeAndFlush(newInvocation(HeartbeatResponse.TYPE, response));

              }

           

              @Override

              public String getType() {

                  return HeartbeatRequest.TYPE;

              }

          }

          其中,HeartbeatResponse 是心跳確認響應。

          8.4、簡單測試

          啟動 Netty Server 服務端,再啟動 Netty Client 客戶端,耐心等待 60 秒后,可以看到心跳日志如下: 

          九、認證邏輯

          從本小節(jié)開始,我們就具體看看業(yè)務邏輯的處理示例。

          認證的過程,如下圖所示:

          9.1、AuthRequest

          創(chuàng)建 AuthRequest 類,定義用戶認證請求。

          代碼如下:

          public class AuthRequest implements Message {

              public static final String TYPE = "AUTH_REQUEST";

             /**

               * 認證 Token

               */

              private String accessToken;

              // ... 省略 setter、getter、toString 方法

          }

          這里我們使用 accessToken 認證令牌進行認證。

          因為一般情況下,我們使用 HTTP 進行登錄系統(tǒng),然后使用登錄后的身份標識(例如說 accessToken 認證令牌),將客戶端和當前用戶進行認證綁定。

          9.2、AuthResponse

          創(chuàng)建 AuthResponse 類,定義用戶認證響應。

          代碼如下:

          public class AuthResponse implements Message {

              public static final String TYPE = "AUTH_RESPONSE";

           

              /**

               * 響應狀態(tài)碼

               */

              private Integer code;

              /**

               * 響應提示

               */

              private String message;

           

              // ... 省略 setter、getter、toString 方法

          }

          9.3、AuthRequestHandler

          服務端...

          創(chuàng)建 AuthRequestHandler 類,為服務端處理客戶端的認證請求。

          代碼如下: 

          代碼比較簡單,看看 <1>、<2>、<3>、<4> 上的注釋。

          9.4、AuthResponseHandler

          客戶端...

          創(chuàng)建 AuthResponseHandler 類,為客戶端處理服務端的認證響應。

          代碼如下:

          @Component

          public class AuthResponseHandler implements MessageHandler<AuthResponse> {

           

              private Logger logger = LoggerFactory.getLogger(getClass());

           

              @Override

              public void execute(Channel channel, AuthResponse message) {

                  logger.info("[execute][認證結果:{}]", message);

              }

           

              @Override

              public String getType() {

                  return AuthResponse.TYPE;

              }

          }

          打印個認證結果,方便調(diào)試。

          9.5、TestController

          客戶端...

          創(chuàng)建 TestController 類,提供 /test/mock 接口,模擬客戶端向服務端發(fā)送請求。

          代碼如下:

          @RestController

          @RequestMapping("/test")

          public class TestController {

           

              @Autowired

              private NettyClient nettyClient;

           

              @PostMapping("/mock")

              public String mock(String type, String message) {

                  // 創(chuàng)建 Invocation 對象

                  Invocation invocation = new Invocation(type, message);

                  // 發(fā)送消息

                  nettyClient.send(invocation);

                  return "success";

              }

          }

          9.6、簡單測試

          啟動 Netty Server 服務端,再啟動 Netty Client 客戶端,然后使用 Postman 模擬一次認證請求。

          如下圖所示: 

          同時,可以看到認證成功的日志如下:

          十一、群聊邏輯

          群聊的過程,如下圖所示: 

          服務端負責將客戶端 A 發(fā)送的群聊消息,轉發(fā)給客戶端 A、B、C。

          友情提示:考慮到邏輯簡潔,提供的本小節(jié)的示例并不是一個一個群,而是所有人在一個大的群聊中哈~

          11.1、ChatSendToAllRequest

          創(chuàng)建 ChatSendToOneRequest 類,發(fā)送給所有人的群聊消息的請求。

          代碼如下:

          public class ChatSendToAllRequest implements Message {

              public static final String TYPE = "CHAT_SEND_TO_ALL_REQUEST";

              /**

               * 消息編號

               */

              private String msgId;

              /**

               * 內(nèi)容

               */

              private String content;

           

              // ... 省略 setter、getter、toString 方法

          }

          PS:如果是正經(jīng)的群聊,會有一個 groupId 字段,表示群編號。

          11.2、ChatSendToAllHandler

          服務端...

          創(chuàng)建 ChatSendToAllHandler 類,為服務端處理客戶端的群聊請求。

          代碼如下: 

          代碼比較簡單,看看 <1>、<2> 上的注釋。

          11.3、簡單測試

           啟動 Netty Server 服務端。

           啟動 Netty Client 客戶端 A。然后使用 Postman 模擬一次認證請求(用戶為 yunai)。

          如下圖所示:

           啟動 Netty Client 客戶端 B。注意,需要設置 --server.port 端口為 8081,避免沖突。

           啟動 Netty Client 客戶端 C。注意,需要設置 --server.port 端口為 8082,避免沖突。 

           最后使用 Postman 模擬一次發(fā)送群聊消息。

          如下圖所示: 

          同時,可以看到客戶端 A 群發(fā)給所有客戶端的日志如下:

          最后,要想系統(tǒng)地學習IM開發(fā)的方方面面,請繼續(xù)閱讀:《新手入門一篇就夠:從零開發(fā)移動端IM

          附錄、系列文章

          跟著源碼學IM(一):手把手教你用Netty實現(xiàn)心跳機制、斷線重連機制

          跟著源碼學IM(二):自已開發(fā)IM很難?手把手教你擼一個Andriod版IM

          跟著源碼學IM(三):基于Netty,從零開發(fā)一個IM服務端

          跟著源碼學IM(四):拿起鍵盤就是干,教你徒手開發(fā)一套分布式IM系統(tǒng)

          跟著源碼學IM(五):正確理解IM長連接、心跳及重連機制,并動手實現(xiàn)

          跟著源碼學IM(六):手把手教你用Go快速搭建高性能、可擴展的IM系統(tǒng)

          跟著源碼學IM(七):手把手教你用WebSocket打造Web端IM聊天

          跟著源碼學IM(八):萬字長文,手把手教你用Netty打造IM聊天》(* 本文)

          本文已同步發(fā)布于“即時通訊技術圈”公眾號。

          ▲ 本文在公眾號上的鏈接是:點此進入。同步發(fā)布鏈接是:http://www.52im.net/thread-3489-1-1.html



          作者:Jack Jiang (點擊作者姓名進入Github)
          出處:http://www.52im.net/space-uid-1.html
          交流:歡迎加入即時通訊開發(fā)交流群 215891622
          討論:http://www.52im.net/
          Jack Jiang同時是【原創(chuàng)Java Swing外觀工程BeautyEye】【輕量級移動端即時通訊框架MobileIMSDK】的作者,可前往下載交流。
          本博文 歡迎轉載,轉載請注明出處(也可前往 我的52im.net 找到我)。


          只有注冊用戶登錄后才能發(fā)表評論。


          網(wǎng)站導航:
           
          Jack Jiang的 Mail: jb2011@163.com, 聯(lián)系QQ: 413980957, 微信: hellojackjiang
          主站蜘蛛池模板: 澎湖县| 衡阳县| 大安市| 昭通市| 涟水县| 交城县| 贡嘎县| 永吉县| 宣化县| 阳谷县| 清丰县| 治多县| 万安县| 盐亭县| 内丘县| 元朗区| 白水县| 博客| 深州市| 历史| 尼木县| 云安县| 招远市| 铁力市| 同德县| 景东| 清涧县| 犍为县| 闸北区| 东乌珠穆沁旗| 玛纳斯县| 余姚市| 远安县| 永昌县| 广南县| 克拉玛依市| 厦门市| 乌鲁木齐市| 萝北县| 赣州市| 泰安市|