Netty實戰：設計一個IM框架

時間 2019-11-13

標籤 netty 實戰設計一個框架欄目 Netty 简体版

原文原文鏈接

bit0chat

bit0chat 是一個基於 Netty 的 IM 即時通信框架java

PS：bit0chat，bit後面沒有0，開源中國認爲我文章中包含辱罵的信息，想了半天可能只有這個緣由了！git

項目地址： https://github.com/all4you/bit0chat (將0刪掉)github

特性:算法

[x] IOC容器 : 經過 @Bean 註解能夠管理全部對象，經過 @Autowired 註解進行對象注入
[x] 自定義協議 : 一個自定義的 Packet 協議，業務的擴展很是簡單
[x] 編解碼器 : 內置 PacketCodec 編解碼器，解決拆包粘包的問題
[x] 業務處理器 : 業務處理器 PacketHandler 與 Packet 分離，支持各類自定義業務處理器
[x] 可選的業務處理方式 : 服務端支持同步或異步的業務處理，能夠由客戶端在 Packet 協議中自主選擇，默認是在業務線程池中異步處理
[x] 可選的序列化方式 : 支持多種序列化方式，能夠由客戶端在 Packet 協議中自主選擇，默認是 ProtoStuff方式
[x] 單機模式 : 支持單機模式
[x] 心跳檢測 : 服務端與客戶端自帶心跳檢查機制，客戶端支持斷線重連

TODO：數據庫

[ ] 集羣模式 : 支持服務端的集羣方式部署，造成一個 Router 層，客戶端經過 Router 獲取可用的服務端實例
[ ] Connection中心 : 一個 Connection 中心，目前是在內存中保存，將來須要支持 Connection 的持久化，
[ ] Message中心 : 消息的存儲的查詢
[ ] User中心 : 用戶與羣組的管理

快速開始

bit0chat-example 模塊提供了一個服務端與客戶端的實現示例，能夠參照該示例進行本身的業務實現。promise

啓動服務端

要啓動服務端，須要獲取一個 Server 的實例，能夠經過 ServerFactory 來獲取。服務器

目前只實現了單機模式下的 Server ，經過 SimpleServerFactory 只須要定義一個端口便可獲取一個單機的 Server 實例，以下所示：架構

public class StandaloneServerApplication {
    public static void main(String[] args) {
        Server server = SimpleServerFactory.getInstance()
            .newServer(8864);
        server.start();
    }
}

服務端啓動成功後，將顯示以下信息：框架

啓動客戶端

目前只實現了直連服務器的客戶端，經過 SimpleClientFactory 只須要指定一個 ServerAttr 便可獲取一個客戶端，而後進行客戶端與服務端的鏈接，以下所示：異步

public class DirectConnectServerClientApplication {

    public static void main(String[] args) {
        Client client = SimpleClientFactory.getInstance()
            .newClient(ServerAttr.getLocalServer(8864));
        client.connect();

        doClientBiz(client);
    }
}

客戶端鏈接上服務端後，將顯示以下信息：

體驗客戶端的功能

目前客戶端提供了三種 Func，分別是：登陸，查看在線用戶列表，發送單聊消息，每種 Func 有不一樣的命令格式。

登陸

經過在客戶端中執行如下命令 -lo houyi 123456 便可實現登陸，目前用戶中心還未實現，經過 Mock 的方式實現一個假的用戶服務，因此輸入任何的用戶名密碼都會登陸成功，而且會爲用戶建立一個用戶id。

登陸成功後，顯示以下：

查看在線用戶

再啓動一個客戶端，而且也執行登陸，登陸成功後，能夠執行 -lu 命令，獲取在線用戶列表，目前用戶是保存在內存中，獲取的結果以下所示：

發送單聊信息

用 gris 這個用戶向 houyi 這個用戶發送單聊信息，只要執行 -pc 1 hello,houyi 命令便可

其中第二個參數數要發送消息給那個用戶的用戶id，第三個參數是消息內容

消息發送方，發送完消息：

消息接收方，接收到消息：

客戶端斷線重連

客戶端和服務端之間維持着心跳，雙方都會檢查鏈接是否可用，客戶端每隔5s會向服務端發送一個 PingPacket，而服務端接收到這個 PingPacket 以後，會回覆一個 PongPacket，這樣表示雙方都是健康的。

當由於某種緣由，服務端沒有收到客戶端發送的消息，服務端將會把該客戶端的鏈接斷開，一樣的客戶端也會作這樣的檢查。

當客戶端與服務端之間的鏈接斷開以後，將會觸發客戶端 HealthyChecker 的 channelInactive 方法，從而進行客戶端的斷線重連。

總體架構

單機版

單機版的架構只涉及到服務端、客戶端，另外有二者之間的協議層，以下圖所示：

除了服務端和客戶端以外，還有三大中心：消息中心，用戶中心，連接中心。

消息中心：主要負責消息的存儲與歷史、離線消息的查詢
用戶中心：主要負責用戶和羣組相關的服務
連接中心：主要負責保存客戶端的連接，服務端從連接中心獲取客戶端的連接，向其推送消息

集羣版

單機版沒法作到高可用，性能與可服務的用戶數也有必定的限制，因此須要有可擴展的集羣版，集羣版在單機版的基礎上增長了一個路由層，客戶端經過路由層來得到可用的服務端地址，而後與服務端進行通信，以下圖所示：

客戶端發送消息給另外一個用戶，服務端接收到這個請求後，從 Connection中心中獲取目標用戶「掛」在哪一個服務端下，若是在本身名下，那最簡單直接將消息推送給目標用戶便可，若是在其餘服務端，則須要將該請求轉交給目標服務端，讓目標服務端將消息推送給目標用戶。

自定義協議

經過一個自定義協議來實現服務端與客戶端之間的通信，協議中有以下幾個字段：

*
* <p>
* The structure of a Packet is like blow:
* +----------+----------+----------------------------+
* |  size    |  value   |  intro                     |
* +----------+----------+----------------------------+
* | 1 bytes  | 0xBC     |  magic number              |
* | 1 bytes  |          |  serialize algorithm       |
* | 4 bytes  |          |  packet symbol             |
* | 4 bytes  |          |  content length            |
* | ? bytes  |          |  the content               |
* +----------+----------+----------------------------+
* </p>
*

每一個字段的含義

所佔字節	用途
1	魔數，默認爲 0xBC
1	序列化的算法
4	Packet 的類型
4	Packet 的內容長度
?	Packet 的內容

序列化算法將會決定該 Packet 在編解碼時，使用何種序列化方式。

Packet 的類型將會決定到達服務端的字節流將被反序列化爲什麼種 Packet，也決定了該 Packet 將會被哪一個 PacketHandler 進行處理。

內容長度將會解決 Packet 的拆包與粘包問題，服務端在解析字節流時，將會等到字節的長度達到內容的長度時，才進行字節的讀取。

除此以外，Packet 中還會存儲一個 sync 字段，該字段將指定服務端在處理該 Packet 的數據時是否須要使用異步的業務線程池來處理。

健康檢查

服務端與客戶端各自維護了一個健康檢查的服務，即 Netty 爲咱們提供的 IdleStateHandler，經過繼承該類，而且實現 channelIdle 方法便可實現鏈接「空閒」時的邏輯處理，當出現空閒時，目前咱們只關心讀空閒，咱們既能夠認爲這條連接出現問題了。

那麼只須要在連接出現問題時，將這條連接關閉便可，以下所示：

public class IdleStateChecker extends IdleStateHandler {

    private static final int DEFAULT_READER_IDLE_TIME = 15;

    private int readerTime;

    public IdleStateChecker(int readerIdleTime) {
        super(readerIdleTime == 0 ? DEFAULT_READER_IDLE_TIME : readerIdleTime, 0, 0, TimeUnit.SECONDS);
        readerTime = readerIdleTime == 0 ? DEFAULT_READER_IDLE_TIME : readerIdleTime;
    }

    @Override
    protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) {
        log.warn("[{}] Hasn't read data after {} seconds, will close the channel:{}", 
        IdleStateChecker.class.getSimpleName(), readerTime, ctx.channel());
        ctx.channel().close();
    }

}

另外，客戶端須要額外再維護一個健康檢查器，正常狀況下他負責定時向服務端發送心跳，當連接的狀態變成 inActive 時，該檢查器將負責進行重連，以下所示：

public class HealthyChecker extends ChannelInboundHandlerAdapter {

    private static final int DEFAULT_PING_INTERVAL = 5;

    private Client client;

    private int pingInterval;

    public HealthyChecker(Client client, int pingInterval) {
        Assert.notNull(client, "client can not be null");
        this.client = client;
        this.pingInterval = pingInterval <= 0 ? DEFAULT_PING_INTERVAL : pingInterval;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        schedulePing(ctx);
    }

    private void schedulePing(ChannelHandlerContext ctx) {
        ctx.executor().schedule(() -> {
            Channel channel = ctx.channel();
            if (channel.isActive()) {
                log.debug("[{}] Send a PingPacket", HealthyChecker.class.getSimpleName());
                channel.writeAndFlush(new PingPacket());
                schedulePing(ctx);
            }
        }, pingInterval, TimeUnit.SECONDS);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ctx.executor().schedule(() -> {
            log.info("[{}] Try to reconnecting...", HealthyChecker.class.getSimpleName());
            client.connect();
        }, 5, TimeUnit.SECONDS);
        ctx.fireChannelInactive();
    }

}

業務線程池

咱們知道，Netty 中維護着兩個 IO 線程池，一個 boss 主要負責連接的創建，另一個 worker 主要負責連接上的數據讀寫，咱們不該該使用 IO 線程來處理咱們的業務，由於這樣極可能會對 IO 線程形成阻塞，致使新連接沒法及時創建或者數據沒法及時讀寫。

爲了解決這個問題，咱們須要在業務線程池中來處理咱們的業務邏輯，可是這並非絕對的，若是咱們要執行的邏輯很簡單，不會形成太大的阻塞，則能夠直接在 IO 線程中處理，好比客戶端發送一個 Ping 服務端回覆一個 Pong，這種狀況是沒有必要在業務線程池中進行處理的，由於處理完了最終仍是要交給 IO 線程去寫數據。可是若是一個業務邏輯須要查詢數據庫或者讀取文件，這每每比較耗時間，因此就須要將這些操做封裝起來交給業務線程池去處理。

服務端容許客戶端在傳輸的 Packet 中指定採用何種方式進行業務的處理，服務端在將字節流解碼成 Packet 以後，會根據 Packet 中的 sync 字段的值，肯定怎樣對該 Packet 進行處理，以下所示：

public class ServerPacketDispatcher extends 
    SimpleChannelInboundHandler<Packet> {
    @Override
    public void channelRead0(ChannelHandlerContext ctx, Packet request) {
        // if the packet should be handled async
        if (request.getAsync() == AsyncHandle.ASYNC) {
            EventExecutor channelExecutor = ctx.executor();
            // create a promise
            Promise<Packet> promise = new DefaultPromise<>(channelExecutor);
            // async execute and get a future
            Future<Packet> future = executor.asyncExecute(promise, ctx, request);
            future.addListener(new GenericFutureListener<Future<Packet>>() {
                @Override
                public void operationComplete(Future<Packet> f) throws Exception {
                    if (f.isSuccess()) {
                        Packet response = f.get();
                        writeResponse(ctx, response);
                    }
                }
            });
        } else {
            // sync execute and get the response packet
            Packet response = executor.execute(ctx, request);
            writeResponse(ctx, response);
        }
    }
}