Netty實戰(zhàn)：設(shè)計一個IM框架就這么簡單！

bitchat 是一個基于 Netty 的 IM 即時通訊框架

項目地址：https://github.com/all4you/bitchat

快速開始

bitchat-example 模塊提供了一個服務(wù)端與客戶端的實現(xiàn)示例，可以參照該示例進行自己的業(yè)務(wù)實現(xiàn)。

啟動服務(wù)端

要啟動服務(wù)端，需要獲取一個 Server 的實例，可以通過 ServerFactory 來獲取。

目前只實現(xiàn)了單機模式下的 Server ，通過 SimpleServerFactory 只需要定義一個端口即可獲取一個單機的 Server 實例，如下所示：

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

服務(wù)端啟動成功后，將顯示如下信息：

啟動客戶端

目前只實現(xiàn)了直連服務(wù)器的客戶端，通過 SimpleClientFactory 只需要指定一個 ServerAttr 即可獲取一個客戶端，然后進行客戶端與服務(wù)端的連接，如下所示：

public class DirectConnectServerClientApplication {

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

        doClientBiz(client);
    }
}

客戶端連接上服務(wù)端后，將顯示如下信息：

體驗客戶端的功能

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

登錄

通過在客戶端中執(zhí)行以下命令?-lo houyi 123456?即可實現(xiàn)登錄，目前用戶中心還未實現(xiàn)，通過 Mock 的方式實現(xiàn)一個假的用戶服務(wù)，所以輸入任何的用戶名密碼都會登錄成功，并且會為用戶創(chuàng)建一個用戶id。

登錄成功后，顯示如下：

查看在線用戶

再啟動一個客戶端，并且也執(zhí)行登錄，登錄成功后，可以執(zhí)行?-lu?命令，獲取在線用戶列表，目前用戶是保存在內(nèi)存中，獲取的結(jié)果如下所示：

發(fā)送單聊信息

用 gris 這個用戶向 houyi 這個用戶發(fā)送單聊信息，只要執(zhí)行?-pc 1 hello,houyi?命令即可

其中第二個參數(shù)數(shù)要發(fā)送消息給那個用戶的用戶id，第三個參數(shù)是消息內(nèi)容

消息發(fā)送方，發(fā)送完消息：

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

客戶端斷線重連

客戶端和服務(wù)端之間維持著心跳，雙方都會檢查連接是否可用，客戶端每隔5s會向服務(wù)端發(fā)送一個 PingPacket，而服務(wù)端接收到這個 PingPacket 之后，會回復(fù)一個 PongPacket，這樣表示雙方都是健康的。

當因為某種原因，服務(wù)端沒有收到客戶端發(fā)送的消息，服務(wù)端將會把該客戶端的連接斷開，同樣的客戶端也會做這樣的檢查。

當客戶端與服務(wù)端之間的連接斷開之后，將會觸發(fā)客戶端 HealthyChecker 的 channelInactive 方法，從而進行客戶端的斷線重連。

整體架構(gòu)

單機版

單機版的架構(gòu)只涉及到服務(wù)端、客戶端，另外有兩者之間的協(xié)議層，如下圖所示：

除了服務(wù)端和客戶端之外，還有三大中心：消息中心，用戶中心，鏈接中心。

• 消息中心：主要負責消息的存儲與歷史、離線消息的查詢

• 用戶中心：主要負責用戶和群組相關(guān)的服務(wù)

• 鏈接中心：主要負責保存客戶端的鏈接，服務(wù)端從鏈接中心獲取客戶端的鏈接，向其推送消息

集群版

單機版無法做到高可用，性能與可服務(wù)的用戶數(shù)也有一定的限制，所以需要有可擴展的集群版，集群版在單機版的基礎(chǔ)上增加了一個路由層，客戶端通過路由層來獲得可用的服務(wù)端地址，然后與服務(wù)端進行通訊，如下圖所示：

客戶端發(fā)送消息給另一個用戶，服務(wù)端接收到這個請求后，從 Connection中心中獲取目標用戶“掛”在哪個服務(wù)端下，如果在自己名下，那最簡單直接將消息推送給目標用戶即可，如果在其他服務(wù)端，則需要將該請求轉(zhuǎn)交給目標服務(wù)端，讓目標服務(wù)端將消息推送給目標用戶。

自定義協(xié)議

通過一個自定義協(xié)議來實現(xiàn)服務(wù)端與客戶端之間的通訊，協(xié)議中有如下幾個字段：

*
* <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>
*

每個字段的含義

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

Packet 的類型將會決定到達服務(wù)端的字節(jié)流將被反序列化為何種 Packet，也決定了該 Packet 將會被哪個 PacketHandler 進行處理。

內(nèi)容長度將會解決 Packet 的拆包與粘包問題，服務(wù)端在解析字節(jié)流時，將會等到字節(jié)的長度達到內(nèi)容的長度時，才進行字節(jié)的讀取。

除此之外，Packet 中還會存儲一個 sync 字段，該字段將指定服務(wù)端在處理該 Packet 的數(shù)據(jù)時是否需要使用異步的業(yè)務(wù)線程池來處理。

健康檢查

服務(wù)端與客戶端各自維護了一個健康檢查的服務(wù)，即 Netty 為我們提供的 IdleStateHandler，通過繼承該類，并且實現(xiàn) channelIdle 方法即可實現(xiàn)連接 “空閑” 時的邏輯處理，當出現(xiàn)空閑時，目前我們只關(guān)心讀空閑，我們既可以認為這條鏈接出現(xiàn)問題了。

那么只需要在鏈接出現(xiàn)問題時，將這條鏈接關(guān)閉即可，如下所示：

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

}

另外，客戶端需要額外再維護一個健康檢查器，正常情況下他負責定時向服務(wù)端發(fā)送心跳，當鏈接的狀態(tài)變成 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();
    }

}

業(yè)務(wù)線程池

我們知道，Netty 中維護著兩個 IO 線程池，一個 boss 主要負責鏈接的建立，另外一個 worker 主要負責鏈接上的數(shù)據(jù)讀寫，我們不應(yīng)該使用 IO 線程來處理我們的業(yè)務(wù)，因為這樣很可能會對 IO 線程造成阻塞，導(dǎo)致新鏈接無法及時建立或者數(shù)據(jù)無法及時讀寫。

為了解決這個問題，我們需要在業(yè)務(wù)線程池中來處理我們的業(yè)務(wù)邏輯，但是這并不是絕對的，如果我們要執(zhí)行的邏輯很簡單，不會造成太大的阻塞，則可以直接在 IO 線程中處理，比如客戶端發(fā)送一個 Ping 服務(wù)端回復(fù)一個 Pong，這種情況是沒有必要在業(yè)務(wù)線程池中進行處理的，因為處理完了最終還是要交給 IO 線程去寫數(shù)據(jù)。但是如果一個業(yè)務(wù)邏輯需要查詢數(shù)據(jù)庫或者讀取文件，這種操作往往比較耗時間，所以就需要將這些操作封裝起來交給業(yè)務(wù)線程池去處理。

服務(wù)端允許客戶端在傳輸?shù)?Packet 中指定采用何種方式進行業(yè)務(wù)的處理，服務(wù)端在將字節(jié)流解碼成 Packet 之后，會根據(jù) 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);
        }
    }
}

不止是IM框架

bitchat?除了可以作為 IM 框架之外，還可以作為一個通用的通訊框架。

Packet 作為通訊的載體，通過繼承 AbstractPacket 即可快速實現(xiàn)自己的業(yè)務(wù)，搭配 PacketHandler 作為數(shù)據(jù)處理器即可實現(xiàn)客戶端與服務(wù)端的通訊。