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

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

項目地址：github.com/all4you/bitc

快速開始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;
}
@Override
protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) {
log.warn("[{}] Hasnt 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);
}
}
}

不止是IM框架

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

原文：mp.weixin.qq.com/s?

推薦閱讀：

相關文章