從事Java已經5年，目前在某互聯網公司做就Java系統架構師，每天都會寫一些技術文章，感興趣的同事請關注我，謝謝。（需要架構資料私信我）

---

bitchat 是一個基于 Netty 的 IM 即時通訊框架
項目地址：https://github.com/all4you/bitchat

快速開始bitchat-example 模塊提供了一個服務端與客戶端的實現示例，可以參照該示例進行自己的業務實現。啟動服務端
要啟動服務端，需要獲取一個 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中心中獲取目標用戶“掛”在哪個服務端下，如果在自己名下，那最簡單直接將消息推送給目標用戶即可，如果在其他服務端，則需要將該請求轉交給目標服務端，讓目標服務端將消息推送給目標用戶。自定義協議

每個字段的含義
所占字節用途1魔數，默認為 0xBC1序列化的算法4Packet 的類型4Packet 的內容長度?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;}@Overrideprotected 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;}@Overridepublic 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);}@Overridepublic 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> {@Overridepublic void channelRead0(ChannelHandlerContext ctx, Packet request) {// if the packet should be handled asyncif (request.getAsync() == AsyncHandle.ASYNC) {EventExecutor channelExecutor = ctx.executor();// create a promisePromise<Packet> promise = new DefaultPromise<>(channelExecutor);// async execute and get a futureFuture<Packet> future = executor.asyncExecute(promise, ctx, request);future.addListener(new GenericFutureListener<Future<Packet>>() {@Overridepublic void operationComplete(Future<Packet> f) throws Exception {if (f.isSuccess()) {Packet response = f.get();writeResponse(ctx, response);}}});} else {// sync execute and get the response packetPacket response = executor.execute(ctx, request);writeResponse(ctx, response);}} }

不止是IM框架bitchat 除了可以作為 IM 框架之外，還可以作為一個通用的通訊框架。
Packet 作為通訊的載體，通過繼承 AbstractPacket 即可快速實現自己的業務，搭配 PacketHandler 作為數據處理器即可實現客戶端與服務端的通訊。

原文：https://mp.weixin.qq.com/s?timestamp=1559194631&src=3&ver=1&signature=lcAYZC239JlrzbvdnNWgZIymFJAtw4y5Gcw7bfo4Qm1D9OB5GevhBdANEsE8-Lt9hb7mr3Y9gVqx34Zi7GltwygWedF7IX9FHzfiJSjK1eTL6Iztc8Yt9uxhCppFGkzcu-nrCwuOEQC5oWH-9ovUmx3xniVNBn4fjpeQvK*gH2k=

總結

以上是生活随笔為你收集整理的netty框架_Netty实战：设计一个IM框架的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。