基于Netty,从零开发IM(二):编码实践篇(im单聊功能)

简介: http://www.52im.net/thread-3974-1-1.html

本文由作者“大白菜”分享,个人博客 cmsblogs.cn,有较多修订和改动。注意:本系列是给IM初学者的文章,IM老油条们还望海涵,勿喷!

1、引言

接上篇《IM系统设计篇》,本篇主要讲解的是通过实战编码实现IM的单聊功能,内容涉及技术原理、编码实践。

补充说明:因为本系列文章主要目的是引导IM初学者在基于Netty的情况下,如何一步一步从零写出IM的逻辑和思维能力,因而为了简化编码实现,本系列中编码实现的客户端都是基于控制台实现的(希望不要被嫌弃),因为理解技术的本质显然比炫酷的外在表现形式更为重要。

2、写在前面

建议你在阅读本文之前,务必先读本系列首篇《IM系统设计篇》,在着重理解IM系统的理论设计思路之后,再来读实战代码则效果更好。

最后,在开始本文之前,请您务必提前了解Netty的相关基础知识,可从本系列首篇《IM系统设计篇》中的“知识准备”一章开始。

3、系列文章

本文是系列文章的第2篇,以下是系列目录:

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 () 抽象方法,主要表示的是业务指令,不同的业务对应不同的指令。

@Data

public abstract class BaseBean implements Serializable {

   private Integer tag=1;//固定值,标识的是一个协议类型,不同协议对应不同的值

   public abstract Byte code();//业务指令抽象方法

}

登录请求实体:

@Data

public class LoginReqBean extends BaseBean implements Serializable {

   private Integer userid;//用户ID

   private String username;//用户名称

   public Byte code() {

       return 1;//业务指令

   }

}

登录响应实体:

@Data

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>() {

   @Override

   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 {

   @Override

   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、参考资料

[1] 手把手教你用Netty实现心跳机制、断线重连机制

[2] 自已开发IM很难?手把手教你撸一个Andriod版IM

[3] 基于Netty,从零开发一个IM服务端

[4] 拿起键盘就是干,教你徒手开发一套分布式IM系统

[5] 正确理解IM长连接、心跳及重连机制,并动手实现

[6] 手把手教你用Go快速搭建高性能、可扩展的IM系统

[7] 手把手教你用WebSocket打造Web端IM聊天

[8] 万字长文,手把手教你用Netty打造IM聊天

[9] 基于Netty实现一套分布式IM系统

[10] 基于Netty,搭建高性能IM集群(含技术思路+源码)

[11] SpringBoot集成开源IM框架MobileIMSDK,实现即时通讯IM聊天功能

学习交流:

- 移动端IM开发入门文章:《新手入门一篇就够:从零开发移动端IM

- 开源IM框架源码:https://github.com/JackJiang2011/MobileIMSDK备用地址点此

(本文已同步发布于:http://www.52im.net/thread-3974-1-1.html

目录
相关文章
|
1月前
|
网络协议 程序员 调度
即时通讯技术文集(第33期):IM开发综合技术合集(Part6) [共12篇]
为了更好地分类阅读 52im.net 总计1000多篇精编文章,我将在每周三推送新的一期技术文集,本次是第33 期。
52 0
|
1月前
|
移动开发 编解码 Java
Netty编码器和解码器
Netty从底层Java通道读到ByteBuf二进制数据,传入Netty通道的流水线,随后开始入站处理。在入站处理过程中,需要将ByteBuf二进制类型解码成Java POJO对象。这个解码过程可以通过Netty的Decoder解码器去完成。在出站处理过程中,业务处理后的结果需要从某个Java POJO对象编码为最终的ByteBuf二进制数据,然后通过底层 Java通道发送到对端。在编码过程中,需要用到Netty的Encoder编码器去完成数据的编码工作。
|
1月前
|
Java Unix Linux
【Netty技术专题】「原理分析系列」Netty强大特性之Native transports扩展开发实战
当涉及到网络通信和高性能的Java应用程序时,Netty是一个强大的框架。它提供了许多功能和组件,其中之一是JNI传输。JNI传输是Netty的一个特性,它为特定平台提供了高效的网络传输。 在本文中,我们将深入探讨Netty提供的特定平台的JNI传输功能,分析其优势和适用场景。我们将介绍每个特定平台的JNI传输,并讨论其性能、可靠性和可扩展性。通过了解这些特定平台的JNI传输,您将能够更好地选择和配置适合您应用程序需求的网络传输方式,以实现最佳的性能和可靠性。
86 7
【Netty技术专题】「原理分析系列」Netty强大特性之Native transports扩展开发实战
|
15天前
|
资源调度 JavaScript 前端开发
IM跨平台技术学习(十一):环信基于Electron打包Web IM桌面端的技术实践
这次借着论证 Web IM端 SDK 是否可以在 Electron 生成的桌面端正常稳定使用,我决定把官方新推出的 webim-vue3-demo,打包到桌面端,并记录了这次验证的过程以及所遇到的问题和解决方法。
26 2
|
1天前
|
边缘计算 JSON 网络协议
移动端IM开发者必读(三):爱奇艺移动端跨国弱网通信的优化实践
本次分享的文章内容,基于爱奇艺面向全球用户推出的国际版,在海外跨国网络环境复杂的前提下,针对性地做了一系列弱网优化实践,取得了不错的效果,在此总结分享我们的一些做法和优化思路,希望对你有所帮助。
5 1
|
1月前
|
Android开发 数据安全/隐私保护 iOS开发
ios和安卓测试包发布网站http://fir.im的注册与常用功能
ios和安卓测试包发布网站http://fir.im的注册与常用功能
25 0
ios和安卓测试包发布网站http://fir.im的注册与常用功能
|
1月前
|
移动开发 网络协议 Java
即时通讯技术文集(第38期):IM代码入门实践(Part2) [共15篇]
为了更好地分类阅读 52im.net 总计1000多篇精编文章,我将在每周三推送新的一期技术文集,本次是第38 期。
48 1
|
1月前
|
安全 Java 数据库
即时通讯技术文集(第37期):IM代码入门实践(Part1) [共16篇]
为了更好地分类阅读 52im.net 总计1000多篇精编文章,我将在每周三推送新的一期技术文集,本次是第37 期。
39 2
|
1月前
|
监控 Devops Java
大型IM工程重构实践:企业微信Android端的重构之路
本文将探讨我们在大型IM工程实践中采用的一些行之有效的重构方法和实例,以及如何让一个大型软件系统持续保持活力。
58 0
|
1月前
|
Rust 监控 JavaScript
抖音技术分享:飞鸽IM桌面端基于Rust语言进行重构的技术选型和实践总结
本文将介绍飞鸽IM前端团队如何结合Rust对飞鸽客户端接待能力进行的技术提升,一步步从概念验证、路径分解到分工开发,再到最后上线收益论证,并分享了其中遇到的技术挑战与经验总结等。
85 1