IM 单聊实现

回到首页 个人中心 反馈问题 注册登录
下载APP
章节索引 :

IM - 单聊

1. 前言

本节，我们主要来实现单聊功能，分别登陆两个账 号，然后可以互相发送消息。具体效果图如下所示：

客户端 1 登陆效果：
图片描述

客户端 2 登陆效果：
图片描述

客户端 1 发送消息效果图：
图片描述

客户端 2 接受消息效果图：
图片描述

2. 实现流程

以上效果是一个基于 console 控制台的聊天效果，根据上节的思路设计，我们也知道主要核心是服务端保存一份关系映射，通过接受人 ID 找到对应的通道进行消息发送。但是，我们实现具体的功能，需要大体上做一个核心步骤的拆解，具体如下所示：

第一步： 编码和解码的实现。

两个核心功能点，第一是序列化和反序列化，第二通讯协议实现。

客户端和服务端之间的数据通讯，我们是基于实体对象去交互，这样数据格式更加的方便，因此，然而对于实体对象对象的序列化和反序列化推荐使用 Fastjson 框架去实现，而不是前面章节所使用的对象流。同时为了更加规范的管理不同业务的实体，我们需要定义一个实体基类，所有的业务实体都继承它（下面会详细讲解）。

第二步： 登录和消息发送两个业务点的实现。

登录主要是为了让用户 ID 和通道进行绑定，在登录成功之后给 Channel 通过 attr () 方法绑定用户 ID，主要目的有两个：

  1. 在发送消息时，可以通过 Channel 获取用户 ID，发送给服务端，让另外一个客户端接收消息时，知道是谁发送过来的消息；
  2. 在 Channel 断开的时候，服务端可以监听到 Channel，并且获取 Channel 的属性，从而删除对应的映射关系。

业务处理，用户登录和消息发送是两个不同的业务点，一般来说需要定义多个 Handler 来分别处理，但是这里为了减少 Handler 的数量，统一一个 Handler 处理。

第三步： 映射关系的实现。

前面也解析过了，服务端需要保存一份映射关系，只需要使用一个 Map 进行存储即可，Map<Integer,Channel>，key 是用户 ID，value 是 Channel，也就是连接通道。

3. 功能实现

3.1 定义实体

实体的设计，主要从两个方面进行入手考虑，第一： 通讯协议的规则，分别是协议标识符、业务指令、数据长度、数据四个部分；第二： 不同业务对应不同的字段属性。具体如下图所示：

图片描述

基础实体

主要定义 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;//业务指令
    }
}

3.2 编码和解码实现

3.2.1 依赖坐标

<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.47</version>
</dependency>

3.2.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);//
    }
}

3.2.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<? extends BaseBean> c= MapUtils.getBean(code);

        //3.反序列化
        BaseBean baseBean=JSON.parseObject(bytes,c);

        list.add(baseBean);
    }
}

3.2.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());
        }
    }
}

3.3 客户端代码实现

3.3.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());
    }
});

3.3.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());
            }
            
        }else if(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(final Channel 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();
    }
}

3.4 服务端代码实现

3.4.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());
    }
});

3.4.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);
        }
    }
}

4. 小结

本节主要实现了单聊功能，实现思路相对还是稍微有点小复杂，大家主要核心掌握以下几点思路：

  1. 不同业务设置不同的实体和指令，需要把指令和实体的关系管理起来，方便反序列化的时候，可以根据指令来反序列化到具体实体；
  2. 需要把用户 ID 和通道的关系管理起来，并且灵活运用 Channel 的 attr () 方法，该方法可以绑定属性值，非常的有用。
Netty 简介
Netty 简介
Netty 第一应用
Netty 入门案例 Netty Bootstrap 客户端启动类 ServerBootstrap 服务端启动类 Netty ChannelFuture 异步监听 Netty 核心组件
EventLoop 和线程模型
Netty Reactor 模型之单线程模型 Netty Reactor 模型之多线程模型 Netty Reactor 模型之主从多线程模型 Netty 线程模型 Netty EventLoop 事件循环机制
ChannelHandler 和 Pipeline
Netty ChannelPipeline数据管道 Netty ChannelHandler业务处理 Netty ChannelHandler生命周期 Netty ChannelHandler性能优化
数据缓存
Netty ByteBuf传输载体 Netty ByteBuf几种类型
Netty 编解码器
Netty编码和解码 如何自定义编解码器 Netty 内置编解码器 Netty粘包和拆包 Netty 通信协议设计概要 Netty 通信协议功能实现
心跳检测
Netty 心跳检测 Netty 断开重连
常见协议
Netty Http 协议 Netty WebSocket 协议
Netty实战篇
IM 聊天系统设计 IM 单聊实现 IM 群聊实现 IM 优化
IM 单聊实现
29/31
字体 A

A
目录
书签
收藏
A
大小