基于Netty,從零開發(fā)IM(二):編碼實(shí)踐篇(im單聊功能)

本文由作者“大白菜”分享,個人博客 cmsblogs.cn,有較多修訂和改動。注意:本系列是給IM初學(xué)者的文章,IM老油條們還望海涵,勿噴!
1、引言
接上篇《IM系統(tǒng)設(shè)計(jì)篇》,本篇主要講解的是通過實(shí)戰(zhàn)編碼實(shí)現(xiàn)IM的單聊功能,內(nèi)容涉及技術(shù)原理、編碼實(shí)踐。

補(bǔ)充說明:因?yàn)楸鞠盗形恼轮饕康氖且龑?dǎo)IM初學(xué)者在基于Netty的情況下,如何一步一步從零寫出IM的邏輯和思維能力,因而為了簡化編碼實(shí)現(xiàn),本系列中編碼實(shí)現(xiàn)的客戶端都是基于控制臺實(shí)現(xiàn)的(希望不要被嫌棄),因?yàn)槔斫饧夹g(shù)的本質(zhì)顯然比炫酷的外在表現(xiàn)形式更為重要。
學(xué)習(xí)交流:
- 移動端IM開發(fā)入門文章:《新手入門一篇就夠:從零開發(fā)移動端IM》
- 開源IM框架源碼:https://github.com/JackJiang2011/MobileIMSDK(備用地址點(diǎn)此)
(本文已同步發(fā)布于:http://www.52im.net/thread-3974-1-1.html)
2、寫在前面
建議你在閱讀本文之前,務(wù)必先讀本系列首篇《IM系統(tǒng)設(shè)計(jì)篇》,在著重理解IM系統(tǒng)的理論設(shè)計(jì)思路之后,再來讀實(shí)戰(zhàn)代碼則效果更好。
最后,在開始本文之前,請您務(wù)必提前了解Netty的相關(guān)基礎(chǔ)知識,可從本系列首篇《IM系統(tǒng)設(shè)計(jì)篇》中的“知識準(zhǔn)備”一章開始。

3、系列文章
本文是系列文章的第2篇,以下是系列目錄:
《基于Netty,從零開發(fā)IM(一):IM系統(tǒng)設(shè)計(jì)篇》
《基于Netty,從零開發(fā)IM(二):編碼實(shí)踐篇(單聊功能)》(* 本文)
《基于Netty,從零開發(fā)IM(三):編碼實(shí)踐篇(群聊功能)》(稍后發(fā)布..?)
《基于Netty,從零開發(fā)IM(四):編碼實(shí)踐篇(系統(tǒng)優(yōu)化)》(稍后發(fā)布..?)
4、運(yùn)行效果
本篇我們主要來實(shí)現(xiàn)的是IM單聊功能,具體就是:模擬IM聊天的兩個用戶分別登陸各自的賬號,然后可以互相發(fā)送聊天消息。
我們提前看一下本篇要實(shí)現(xiàn)的功能運(yùn)行效果。
客戶端 1 登陸效果:

客戶端 2 登陸效果:

客戶端 1 發(fā)送消息效果圖:

客戶端 2 接受消息效果圖:

5、技術(shù)原理
5.1 概述
上節(jié)中,可以看到此次實(shí)戰(zhàn)的運(yùn)行效果是一個基于 console 控制臺的聊天,根據(jù)上篇《IM系統(tǒng)設(shè)計(jì)篇》的思路設(shè)計(jì),我們也知道主要核心是服務(wù)端保存一份關(guān)系映射,通過接受人 ID 找到對應(yīng)的通道進(jìn)行消息發(fā)送。
但是,我們要想實(shí)現(xiàn)具體的功能,則需要大體上做一個核心技術(shù)實(shí)現(xiàn)步驟的拆解,具體拆分成以下三步。
5.2 第一步: 編碼和解碼的實(shí)現(xiàn)
針對IM單聊功能,有兩個核心技術(shù)點(diǎn):
1)一是序列化和反序列化;
2)二是通訊協(xié)議實(shí)現(xiàn)。
客戶端和服務(wù)端之間的數(shù)據(jù)通訊,我們是基于實(shí)體對象去交互的,這樣數(shù)據(jù)格式更加的方便。
對于實(shí)體對象的序列化和反序列化,推薦使用?Fastjson?框架去實(shí)現(xiàn),而不是Netty官方示例所使用的對象流。
同時為了更加規(guī)范的管理不同業(yè)務(wù)實(shí)體,我們需要定義一個實(shí)體基類,所有的業(yè)務(wù)實(shí)體都繼承它(下面的章節(jié)會詳細(xì)講解)。
5.3 第二步: 登錄和消息發(fā)送兩個業(yè)務(wù)點(diǎn)的實(shí)現(xiàn)
登錄主要是為了讓用戶 ID 和通道(就是Netty中的Channel,也即網(wǎng)絡(luò)連接)進(jìn)行綁定。
在登錄成功之后為 Channel 通過?attr()?方法綁定該用戶 ID,主要目的有三個:
1)客戶端A在發(fā)送消息時,服務(wù)端可以通過 Channel 獲取消息發(fā)送者的用戶ID,以便知道消息是“誰”發(fā)過來的;
2)服務(wù)端在收到客戶端A發(fā)過來的消息時,通過消息中的接收者用戶ID,可以獲取接收者的Channel,以便知道消息該發(fā)給“誰”;
3)在 Channel 斷開的時候,服務(wù)端可以監(jiān)聽到 Channel,并且獲取 Channel 的屬性,從而刪除對應(yīng)的用戶ID和Chennel映射關(guān)系。
對于業(yè)務(wù)處理來說,用戶登錄和消息發(fā)送是兩個不同的業(yè)務(wù)點(diǎn),一般來說需要定義多個 Handler 來分別處理,但是這里為了減少 Handler 的數(shù)量,統(tǒng)一一個 Handler 處理。
* 友情提示:用戶ID和Chennel的綁定,可以參考成熟的開源IM工程?MobileIMSDK??中的實(shí)現(xiàn)邏輯?OnlineProcessor.java,以便通過更接近IM產(chǎn)品級實(shí)踐進(jìn)行學(xué)習(xí)。
5.4 第三步: 映射關(guān)系的實(shí)現(xiàn)
前面也分析過了,服務(wù)端需要保存一份用戶ID和Channel映射關(guān)系,這個映射關(guān)系只需要使用一個 Map 進(jìn)行存儲即可,即?Map<Integer,Channel>,其中:key 是用戶 ID、value 是 Channel(Channel也就是客戶端的網(wǎng)絡(luò)連接對象啦)。
這部分需要交待的不多,理解清楚用戶ID和Channel的關(guān)系就足夠了。
接下來就是具體的編碼實(shí)戰(zhàn)了。。。
6、實(shí)體定義實(shí)戰(zhàn)
實(shí)體的設(shè)計(jì),主要從兩個方面進(jìn)行入手考慮:
1)通訊協(xié)議的規(guī)則,分別是協(xié)議標(biāo)識符、業(yè)務(wù)指令、數(shù)據(jù)長度、數(shù)據(jù)四個部分;
2)不同業(yè)務(wù)對應(yīng)不同的字段屬性。
具體如下圖所示:

基礎(chǔ)實(shí)體:
主要定義 tag 字段,標(biāo)識協(xié)議的標(biāo)識符,code () 抽象方法,主要表示的是業(yè)務(wù)指令,不同的業(yè)務(wù)對應(yīng)不同的指令。
@Data
public abstract class BaseBean implements Serializable {
????private Integer tag=1;//固定值,標(biāo)識的是一個協(xié)議類型,不同協(xié)議對應(yīng)不同的值
????public abstract Byte code();//業(yè)務(wù)指令抽象方法
}
登錄請求實(shí)體:
@Data
public class LoginReqBean extends BaseBean implements Serializable {
????private Integer userid;//用戶ID
????private String username;//用戶名稱
????public Byte code() {
????????return 1;//業(yè)務(wù)指令
????}
}
登錄響應(yīng)實(shí)體:
@Data
public class LoginResBean extends BaseBean implements Serializable {
????private Integer status;//響應(yīng)狀態(tài),0登錄成功,1登錄失敗
????private String msg;//響應(yīng)信息
????private Integer userid;//用戶ID
?????public Byte code() {
????????return 2;//業(yè)務(wù)指令
????}
}
消息發(fā)送實(shí)體:
public class MsgReqBean extends BaseBean implements Serializable {
????private Integer fromuserid;//發(fā)送人ID
????private Integer touserid;//接受人ID
????private String msg;//發(fā)送消息
?????public Byte code() {
????????return 3;//業(yè)務(wù)指令
????}
}
消息響應(yīng)響應(yīng):
public class MsgResBean extends BaseBean implements Serializable {
????private Integer status;//響應(yīng)狀態(tài),0發(fā)送成功,1發(fā)送失敗
????private String msg;//響應(yīng)信息
?????public Byte code() {
????????return 4;//業(yè)務(wù)指令
????}
}
消息接受實(shí)體:
public class MsgRecBean extends BaseBean implements Serializable {
????private Integer fromuserid;//發(fā)送人ID
????private String msg;//消息
?
????public Byte code() {
????????return 5;//業(yè)務(wù)指令
????}
}
7、編碼和解碼實(shí)戰(zhàn)
7.1 依賴坐標(biāo)
<dependency>
????<groupId>com.alibaba</groupId>
????<artifactId>fastjson</artifactId>
????<version>1.2.47</version>
</dependency>
7.2 編碼實(shí)現(xiàn)
public class MyEncoder extends MessageToByteEncoder<BaseBean> {
????protected void encode(
????????ChannelHandlerContext channelHandlerContext,
????????BaseBean baseBean,
????????ByteBuf byteBuf) throws Exception {
?
????????//1.把實(shí)體序列化成字節(jié)數(shù)字
????????byte[] bytes= JSON.toJSONBytes(baseBean);
?
????????//2.根據(jù)協(xié)議組裝數(shù)據(jù)
????????byteBuf.writeInt(baseBean.getTag());//標(biāo)識(4個字節(jié))
????????byteBuf.writeByte(baseBean.code());//指令(1個字節(jié))
????????byteBuf.writeInt(bytes.length);//長度(4個字節(jié))
????????byteBuf.writeBytes(bytes);//
????}
}
7.3 解碼實(shí)現(xiàn)
public class MyDecoder extends ByteToMessageDecoder {
????protected void decode(
????????ChannelHandlerContext channelHandlerContext,
????????ByteBuf byteBuf,
????????List<Object> list) throws Exception {
?
????????//1.根據(jù)協(xié)議取出數(shù)據(jù)
????????int tag=byteBuf.readInt();//標(biāo)識符
????????byte code=byteBuf.readByte();//獲取指令
????????int len=byteBuf.readInt();//獲取數(shù)據(jù)長度
????????byte[] bytes=new byte[len];
????????byteBuf.readBytes(bytes);
?
????????//2.根據(jù)code獲取類型
????????Class<? extendsBaseBean> c= MapUtils.getBean(code);
?
????????//3.反序列化
????????BaseBean baseBean=JSON.parseObject(bytes,c);
????????list.add(baseBean);
????}
}
7.4 指令和實(shí)體關(guān)系
為什么需要這么一個工具類呢?指令表示的是業(yè)務(wù)類型,不同的業(yè)務(wù)對應(yīng)不同的實(shí)體,那么解碼的時候,怎么知道反序列化成什么樣的實(shí)體呢?思路是獲取到的指令,再根據(jù)指令找到對應(yīng)的實(shí)體即可。
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,專門管理指令和實(shí)體的關(guān)系
????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. 根據(jù)指令獲取對應(yīng)的實(shí)體
????public static Class<? extends BaseBean> getBean(Byte code){
????????try{
????????????return map.get(code);
????????}catch(Exception e){
????????????throw new RuntimeException(e.getMessage());
????????}
????}
}
8、客戶端代碼實(shí)戰(zhàn)
8.1 Pipeline 管理鏈表
.handler(new ChannelInitializer<SocketChannel>() {
????@Override
????public void initChannel(SocketChannel ch) {
????????//1.拆包器
????????ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,5,4));
????????//2.自定義解碼器
????????ch.pipeline().addLast(new MyDecoder());
????????//3.自定義業(yè)務(wù)
????????ch.pipeline().addLast(new ClientChatHandler());
????????//4.自定義編碼器
????????ch.pipeline().addLast(new MyEncoder());
????}
});
8.2 業(yè)務(wù) Handler
所有的業(yè)務(wù)處理在同一個 Handler 里面進(jìn)行處理,通過判斷實(shí)體類型來區(qū)分不同的業(yè)務(wù)處理。
public class ClientChatHandler extends ChannelInboundHandlerAdapter {
????@Override
????public void channelActive(ChannelHandlerContext ctx) throws Exception {
????????//通道就緒時,發(fā)起登錄請求
????????login(ctx.channel());
????}
?
????@Override
????public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
????????//根據(jù)msg做類型判斷,不同的業(yè)務(wù)做不同的處理
????????if(msg instanceof LoginResBean){
????????????//1.登錄結(jié)果響應(yīng)
????????????LoginResBean res=(LoginResBean) msg;
????????????System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>登錄響應(yīng):"+res.getMsg());
????????????if(res.getStatus()==0){
????????????????//1.登錄成功,則給通道綁定屬性
????????????????ctx.channel().attr(AttributeKey.valueOf("userid")).set(res.getUserid());
????????????????//2.調(diào)用發(fā)送消息方法
????????????????sendMsg(ctx.channel());
????????????}else{
????????????????//1.登錄失敗,調(diào)用登錄方法
????????????????login(ctx.channel());
????????????}
?
????????}elseif(msg instanceof MsgResBean){
????????????//1.發(fā)送消息結(jié)果響應(yīng)
????????????MsgResBean res=(MsgResBean)msg;
????????????System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>發(fā)送響應(yīng):"+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);
????}
?
????//發(fā)送消息方法
????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(">>聊天內(nèi)容:");
????????????????????String msg=scanner.next();
?
????????????????????MsgReqBean bean=new MsgReqBean();
????????????????????//從通道屬性獲取發(fā)送人ID
????????????????????Integer fromuserid=(Integer) channel.attr(
????????????????????????AttributeKey.valueOf("userid")
????????????????????).get();
????????????????????//發(fā)送人ID
????????????????????bean.setFromuserid(fromuserid);
????????????????????//接受人ID
????????????????????bean.setTouserid(touserid);
????????????????????//發(fā)送消息
????????????????????bean.setMsg(msg);
????????????????????channel.writeAndFlush(bean);
????????????????}
????????????}
????????}).start();
????}
}
9、服務(wù)端代碼實(shí)戰(zhàn)
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.業(yè)務(wù)Handler
????????ch.pipeline().addLast(new ServerChatHandler());
????????//4.自定義編碼器
????????ch.pipeline().addLast(new MyEncoder());
????}
});
9.2 業(yè)務(wù) 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.發(fā)送消息請求
????????????sendMsg((MsgReqBean)msg,ctx.channel());
????????}
????}
????//登錄處理方法
????private void login(LoginReqBean bean, Channel channel){
????????LoginResBean res=new LoginResBean();
?
????????//從map里面根據(jù)用戶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.響應(yīng)
????????????res.setStatus(0);
????????????res.setMsg("登錄成功");
????????????res.setUserid(bean.getUserid());
????????????channel.writeAndFlush(res);
????????}else{
????????????//通道不為空,證明該用戶已經(jīng)在線了
?
????????????res.setStatus(1);
????????????res.setMsg("該賬戶目前在線");
????????????channel.writeAndFlush(res);
????????}
????}
????//消息發(fā)送處理方法
????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、本篇小結(jié)
本篇主要編碼實(shí)戰(zhàn)了IM的單聊功能,實(shí)現(xiàn)思路相對還是稍微有點(diǎn)小復(fù)雜。
大家主要核心掌握以下幾點(diǎn)思路就可以了:
1)不同的業(yè)務(wù)可設(shè)置不同的實(shí)體和指令,需要把指令和實(shí)體的關(guān)系管理起來,方便反序列化的時候,可以根據(jù)指令來反序列化到具體實(shí)體;
2)需要把用戶 ID 和通道的關(guān)系管理起來(方便根據(jù)用戶ID找到Channel通道,反過來也一樣),并且靈活運(yùn)用 Channel 的 attr () 方法,該方法可以綁定屬性值,非常的有用。
11、參考資料
[1]?手把手教你用Netty實(shí)現(xiàn)心跳機(jī)制、斷線重連機(jī)制
[2]?自已開發(fā)IM很難?手把手教你擼一個Andriod版IM
[3]?基于Netty,從零開發(fā)一個IM服務(wù)端
[4]?拿起鍵盤就是干,教你徒手開發(fā)一套分布式IM系統(tǒng)
[5]?正確理解IM長連接、心跳及重連機(jī)制,并動手實(shí)現(xiàn)
[6]?手把手教你用Go快速搭建高性能、可擴(kuò)展的IM系統(tǒng)
[7]?手把手教你用WebSocket打造Web端IM聊天
[8]?萬字長文,手把手教你用Netty打造IM聊天
[9]?基于Netty實(shí)現(xiàn)一套分布式IM系統(tǒng)
[10]?基于Netty,搭建高性能IM集群(含技術(shù)思路+源碼)
[11]?SpringBoot集成開源IM框架MobileIMSDK,實(shí)現(xiàn)即時通訊IM聊天功能
(本文已同步發(fā)布于:http://www.52im.net/thread-3974-1-1.html)