基於Netty，從零開發IM(二)：編碼實踐篇（im單聊功能）
本文由作者“大白菜”分享，個人部落格 cmsblogs.cn，有較多修訂和改動。注意：本系列是給IM初學者的文章，IM老油條們還望海涵，勿噴！
1、引言
接上篇《IM系統設計篇》，本篇主要講解的是通過實戰編碼實現IM的單聊功能，內容涉及技術原理、編碼實踐。
補充說明：因為本系列文章主要目的是引導IM初學者在基於Netty的情況下，如何一步一步從零寫出IM的邏輯和思維能力，因而為了簡化編碼實現，本系列中編碼實現的客戶端都是基於控制檯實現的（希望不要被嫌棄），因為理解技術的本質顯然比炫酷的外在表現形式更為重要。
2、寫在前面
建議你在閱讀本文之前，務必先讀本系列首篇《IM系統設計篇》，在著重理解IM系統的理論設計思路之後，再來讀實戰程式碼則效果更好。
最後，在開始本文之前，請您務必提前瞭解Netty的相關基礎知識，可從本系列首篇《IM系統設計篇》中的“知識準備”一章開始。
3、系列文章
本文是系列文章的第2篇，以下是系列目錄：
- 《基於Netty，從零開發IM(一)：IM系統設計篇》
- 《基於Netty，從零開發IM(二)：編碼實踐篇（單聊功能）》（* 本文）
- 《基於Netty，從零開發IM(三)：編碼實踐篇（群聊功能）》（稍後釋出.. ）
- 《基於Netty，從零開發IM(四)：編碼實踐篇（系統優化）》（稍後釋出.. ）
4、執行效果
本篇我們主要來實現的是IM單聊功能，具體就是：模擬IM聊天的兩個使用者分別登陸各自的賬號，然後可以互相傳送聊天訊息。
我們提前看一下本篇要實現的功能執行效果。
客戶端 1 登陸效果：
客戶端 2 登陸效果：
客戶端 1 傳送訊息效果圖：
客戶端 2 接受訊息效果圖：
5、技術原理
5.1 概述
上節中，可以看到此次實戰的執行效果是一個基於 console 控制檯的聊天，根據上篇《IM系統設計篇》的思路設計，我們也知道主要核心是服務端儲存一份關係對映，通過接受人 ID 找到對應的通道進行訊息傳送。
但是，我們要想實現具體的功能，則需要大體上做一個核心技術實現步驟的拆解，具體拆分成以下三步。
5.2 第一步： 編碼和解碼的實現
針對IM單聊功能，有兩個核心技術點：
- 1）一是序列化和反序列化；
- 2）二是通訊協議實現。
客戶端和服務端之間的資料通訊，我們是基於實體物件去互動的，這樣資料格式更加的方便。
對於實體物件的序列化和反序列化，推薦使用 Fastjson 框架去實現，而不是Netty官方示例所使用的物件流。
同時為了更加規範的管理不同業務實體，我們需要定義一個實體基類，所有的業務實體都繼承它（下面的章節會詳細講解）。
5.3 第二步： 登入和訊息傳送兩個業務點的實現
登入主要是為了讓使用者 ID 和通道（就是Netty中的Channel，也即網路連線）進行繫結。
在登入成功之後為 Channel 通過 attr() 方法繫結該使用者 ID，主要目的有三個：
- 1）客戶端A在傳送訊息時，服務端可以通過 Channel 獲取訊息傳送者的使用者ID，以便知道訊息是“誰”發過來的；
- 2）服務端在收到客戶端A發過來的訊息時，通過訊息中的接收者使用者ID，可以獲取接收者的Channel，以便知道訊息該發給“誰”；
- 3）在 Channel 斷開的時候，服務端可以監聽到 Channel，並且獲取 Channel 的屬性，從而刪除對應的使用者ID和Chennel對映關係。
對於業務處理來說，使用者登入和訊息傳送是兩個不同的業務點，一般來說需要定義多個 Handler 來分別處理，但是這裡為了減少 Handler 的數量，統一一個 Handler 處理。
* 友情提示：使用者ID和Chennel的繫結，可以參考成熟的開源IM工程 MobileIMSDK 中的實現邏輯 OnlineProcessor.java，以便通過更接近IM產品級實踐進行學習。
5.4 第三步： 對映關係的實現
前面也分析過了，服務端需要儲存一份使用者ID和Channel對映關係，這個對映關係只需要使用一個 Map 進行儲存即可，即 Map<Integer,Channel>，其中：key 是使用者 ID、value 是 Channel（Channel也就是客戶端的網路連線物件啦）。
這部分需要交待的不多，理解清楚使用者ID和Channel的關係就足夠了。
接下來就是具體的編碼實戰了。。。
6、實體定義實戰
實體的設計，主要從兩個方面進行入手考慮：
- 1）通訊協議的規則，分別是協議識別符號、業務指令、資料長度、資料四個部分；
- 2）不同業務對應不同的欄位屬性。
具體如下圖所示：
基礎實體：
主要定義 tag 欄位，標識協議的識別符號，code () 抽象方法，主要表示的是業務指令，不同的業務對應不同的指令。
public abstract class BaseBean implements Serializable {
private Integer tag=1;//固定值，標識的是一個協議型別，不同協議對應不同的值
public abstract Byte code();//業務指令抽象方法
}
登入請求實體：
public class LoginReqBean extends BaseBean implements Serializable {
private Integer userid;//使用者ID
private String username;//使用者名稱稱
public Byte code() {
return 1;//業務指令
}
}
登入響應實體：
public class LoginResBean extends BaseBean implements Serializable {
private Integer status;//響應狀態，0登入成功，1登入失敗
private String msg;//響應資訊
private Integer userid;//使用者ID
public Byte code() {
return 2;//業務指令
}
}
訊息傳送實體：
public class MsgReqBean extends BaseBean implements Serializable {
private Integer fromuserid;//傳送人ID
private Integer touserid;//接受人ID
private String msg;//傳送訊息
public Byte code() {
return 3;//業務指令
}
}
訊息響應響應：
public class MsgResBean extends BaseBean implements Serializable {
private Integer status;//響應狀態，0傳送成功，1傳送失敗
private String msg;//響應資訊
public Byte code() {
return 4;//業務指令
}
}
訊息接受實體：
public class MsgRecBean extends BaseBean implements Serializable {
private Integer fromuserid;//傳送人ID
private String msg;//訊息
public Byte code() {
return 5;//業務指令
}
}
7、編碼和解碼實戰
7.1 依賴座標
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.47</version>
</dependency>
7.2 編碼實現
public class MyEncoder extends MessageToByteEncoder<BaseBean> {
protected void encode(
ChannelHandlerContext channelHandlerContext,
BaseBean baseBean,
ByteBuf byteBuf) throws Exception {
//1.把實體序列化成位元組數字
byte[] bytes= JSON.toJSONBytes(baseBean);
//2.根據協議組裝資料
byteBuf.writeInt(baseBean.getTag());//標識（4個位元組）
byteBuf.writeByte(baseBean.code());//指令（1個位元組）
byteBuf.writeInt(bytes.length);//長度（4個位元組）
byteBuf.writeBytes(bytes);//
}
}
7.3 解碼實現
public class MyDecoder extends ByteToMessageDecoder {
protected void decode(
ChannelHandlerContext channelHandlerContext,
ByteBuf byteBuf,
List<Object> list) throws Exception {
//1.根據協議取出資料
int tag=byteBuf.readInt();//識別符號
byte code=byteBuf.readByte();//獲取指令
int len=byteBuf.readInt();//獲取資料長度
byte[] bytes=new byte[len];
byteBuf.readBytes(bytes);
//2.根據code獲取型別
Class<? extendsBaseBean> c= MapUtils.getBean(code);
//3.反序列化
BaseBean baseBean=JSON.parseObject(bytes,c);
list.add(baseBean);
}
}
7.4 指令和實體關係
為什麼需要這麼一個工具類呢？指令表示的是業務型別，不同的業務對應不同的實體，那麼解碼的時候，怎麼知道反序列化成什麼樣的實體呢？思路是獲取到的指令，再根據指令找到對應的實體即可。
public class MapUtils {
//1. 自定義指令
private static Byte codeLoginReq=1;
private static Byte codeLoginRes=2;
private static Byte codeMsgReq=3;
private static Byte codeMsgRes=4;
private static Byte codeMsgRec=5;
//2. 自定義一個Map，專門管理指令和實體的關係
private static Map<Byte, Class<? extends BaseBean>> map=new HashMap<Byte,Class<? extends BaseBean>>();
//3. 初始化
static{
map.put(codeLoginReq, LoginReqBean.class);
map.put(codeLoginRes, LoginResBean.class);
map.put(codeMsgReq, MsgReqBean.class);
map.put(codeMsgRes, MsgResBean.class);
map.put(codeMsgRec, MsgRecBean.class);
}
//4. 根據指令獲取對應的實體
public static Class<? extends BaseBean> getBean(Byte code){
try{
return map.get(code);
}catch(Exception e){
throw new RuntimeException(e.getMessage());
}
}
}
8、客戶端程式碼實戰
8.1 Pipeline 管理連結串列
.handler(new ChannelInitializer<SocketChannel>() {
public void initChannel(SocketChannel ch) {
//1.拆包器
ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,5,4));
//2.自定義解碼器
ch.pipeline().addLast(new MyDecoder());
//3.自定義業務
ch.pipeline().addLast(new ClientChatHandler());
//4.自定義編碼器
ch.pipeline().addLast(new MyEncoder());
}
});
8.2 業務 Handler
所有的業務處理在同一個 Handler 裡面進行處理，通過判斷實體型別來區分不同的業務處理。
public class ClientChatHandler extends ChannelInboundHandlerAdapter {
public void channelActive(ChannelHandlerContext ctx) throws Exception {
//通道就緒時，發起登入請求
login(ctx.channel());
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//根據msg做型別判斷，不同的業務做不同的處理
if(msg instanceof LoginResBean){
//1.登入結果響應
LoginResBean res=(LoginResBean) msg;
System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>登入響應："+res.getMsg());
if(res.getStatus()==0){
//1.登入成功，則給通道繫結屬性
ctx.channel().attr(AttributeKey.valueOf("userid")).set(res.getUserid());
//2.呼叫傳送訊息方法
sendMsg(ctx.channel());
}else{
//1.登入失敗，呼叫登入方法
login(ctx.channel());
}
}elseif(msg instanceof MsgResBean){
//1.傳送訊息結果響應
MsgResBean res=(MsgResBean)msg;
System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>傳送響應："+res.getMsg());
}else if(msg instanceof MsgRecBean){
//2.接受訊息
MsgRecBean res=(MsgRecBean)msg;
System.out.println("fromuserid="+res.getFromuserid()+",msg="+res.getMsg());
}
}
//登入方法
private void login(Channel channel){
Scanner scanner=new Scanner(System.in);
System.out.println(">>使用者ID：");
Integer userid=scanner.nextInt();
System.out.println(">>使用者名稱稱：");
String username=scanner.next();
LoginReqBean bean=new LoginReqBean();
bean.setUserid(userid);
bean.setUsername(username);
channel.writeAndFlush(bean);
}
//傳送訊息方法
private void sendMsg(finalChannel channel){
final Scanner scanner=new Scanner(System.in);
new Thread(new Runnable() {
public void run() {
while(true){
System.out.println(">>接收人ID：");
Integer touserid=scanner.nextInt();
System.out.println(">>聊天內容：");
String msg=scanner.next();
MsgReqBean bean=new MsgReqBean();
//從通道屬性獲取傳送人ID
Integer fromuserid=(Integer) channel.attr(
AttributeKey.valueOf("userid")
).get();
//傳送人ID
bean.setFromuserid(fromuserid);
//接受人ID
bean.setTouserid(touserid);
//傳送訊息
bean.setMsg(msg);
channel.writeAndFlush(bean);
}
}
}).start();
}
}
9、服務端程式碼實戰
9.1 Pipeline 管理連結串列
.childHandler(new ChannelInitializer<NioSocketChannel>() {
protected void initChannel(NioSocketChannel ch) {
//1.拆包器
ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,5,4));
//2.自定義解碼器
ch.pipeline().addLast(new MyDecoder());
//3.業務Handler
ch.pipeline().addLast(new ServerChatHandler());
//4.自定義編碼器
ch.pipeline().addLast(new MyEncoder());
}
});
9.2 業務 Handler
public class ServerChatHandler extends ChannelInboundHandlerAdapter{
//1.定義一個Map（key是使用者ID，value是連線通道）
private static Map<Integer, Channel> map=new HashMap<Integer, Channel>();
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if(msg instanceof LoginReqBean){
//1.登入請求
login((LoginReqBean) msg,ctx.channel());
}else if(msg instanceof MsgReqBean){
//2.傳送訊息請求
sendMsg((MsgReqBean)msg,ctx.channel());
}
}
//登入處理方法
private void login(LoginReqBean bean, Channel channel){
LoginResBean res=new LoginResBean();
//從map裡面根據使用者ID獲取連線通道
Channel c=map.get(bean.getUserid());
if(c==null){
//通道為空，證明該使用者沒有線上
//1.新增到map
map.put(bean.getUserid(),channel);
//2.給通道賦值
channel.attr(AttributeKey.valueOf("userid")).set(bean.getUserid());
//3.響應
res.setStatus(0);
res.setMsg("登入成功");
res.setUserid(bean.getUserid());
channel.writeAndFlush(res);
}else{
//通道不為空，證明該使用者已經線上了
res.setStatus(1);
res.setMsg("該賬戶目前線上");
channel.writeAndFlush(res);
}
}
//訊息傳送處理方法
private void sendMsg(MsgReqBean bean,Channel channel){
Integer touserid=bean.getTouserid();
Channel c=map.get(touserid);
if(c==null){
MsgResBean res=new MsgResBean();
res.setStatus(1);
res.setMsg(touserid+",不線上");
channel.writeAndFlush(res);
}else{
MsgRecBean res=new MsgRecBean();
res.setFromuserid(bean.getFromuserid());
res.setMsg(bean.getMsg());
c.writeAndFlush(res);
}
}
}
10、本篇小結
本篇主要編碼實戰了IM的單聊功能，實現思路相對還是稍微有點小複雜。
大家主要核心掌握以下幾點思路就可以了：
- 1）不同的業務可設定不同的實體和指令，需要把指令和實體的關係管理起來，方便反序列化的時候，可以根據指令來反序列化到具體實體；
- 2）需要把使用者 ID 和通道的關係管理起來（方便根據使用者ID找到Channel通道，反過來也一樣），並且靈活運用 Channel 的 attr () 方法，該方法可以繫結屬性值，非常的有用。
11、參考資料
[2] 自已開發IM很難？手把手教你擼一個Andriod版IM
[6] 手把手教你用Go快速搭建高效能、可擴充套件的IM系統
[10] 基於Netty，搭建高效能IM叢集（含技術思路+原始碼）
[11] SpringBoot整合開源IM框架MobileIMSDK，實現即時通訊IM聊天功能
學習交流：
- 移動端IM開發入門文章：《新手入門一篇就夠：從零開發移動端IM》
- 開源IM框架原始碼：http://github.com/JackJiang2011/MobileIMSDK（備用地址點此）
（本文已同步釋出於：http://www.52im.net/thread-3974-1-1.html）
- 微信Windows端IM訊息資料庫的優化實踐：查詢慢、體積大、檔案損壞等
- IM聊天系統安全手段之傳輸內容端到端加密技術
- 基於Netty，從零開發IM(二)：編碼實踐篇（im單聊功能）
- 基於Netty，從零開發IM(二)：編碼實踐篇（im單聊功能）
- 基於Netty，徒手擼IM(一)：IM系統設計篇
- 基於Netty，徒手擼IM(一)：IM系統設計篇
- 一套十萬級TPS的IM綜合訊息系統的架構實踐與思考
- 一套十萬級TPS的IM綜合訊息系統的架構實踐與思考
- 解密抖音春節紅包背後的技術設計與實踐
- B站基於微服務的API閘道器從0到1的演進之路
- 微信團隊分享：微信後臺在海量併發請求下是如何做到不崩潰的
- 微信團隊分享：微信後臺在海量併發請求下是如何做到不崩潰的
- 視訊直播技術乾貨：一文讀懂主流視訊直播系統的推拉流架構、傳輸協議等
- 不為人知的網路程式設計(十一)：從底層入手，深度分析TCP連線耗時的祕密
- 萬字長文：手把手教你實現一套高效的IM長連線自適應心跳保活機制
- 萬字長文：手把手教你實現一套高效的IM長連線自適應心跳保活機制
- SpringBoot整合開源IM框架MobileIMSDK，實現即時通訊IM聊天功能
- 直播技術乾貨分享：千萬級直播系統後端架構設計的方方面面
- IM全文檢索技術專題(四)：微信iOS端的最新全文檢索技術優化實踐
- 網路程式設計懶人入門(十四)：到底什麼是Socket？一文即懂！