Netty的分布式聊天系统

基于Netty的分布式聊天系统

Gitee地址：https://gitee.com/yuyuuyuy/micro-mall

---

---

前言

IM 全称是『Instant Messaging』，中文名是即时通讯。在这个高度信息化的移动互联网时代，生活中 IM 类产品已经成为必备品，比较有名的如钉钉、微信、QQ 等以 IM 为核心功能的产品。本文探讨IM系统的架构以及具体实现。使用到的技术有：gateway,nacos,netty,redis,rabbitmq,sharding-sphere,mybatis,mysql

---

一、IM系统架构的探讨

最简单的架构，就是客户端直接连接Netty服务器，然后通过服务器来通讯。

然而，netty服务器是动态变化的，客户端也不好确定到底连入哪一个服务器。因此，引入网关层和注册中心，每个Netty服务器在注册中心注册，用户请求Netty服务器时，先通过网关，网关经过负载均衡算法，从注册中心的Netty服务器中选出一个给用户使用，并将客户端与Netty服务器的路由信息保存到redis。

以上架构只能支持用户在线即时通讯，要是有一方离线，那离线用户就收不到离线消息了，因此，考虑把离线消息存入reids集群，用户上线后，先从redis中拉取离线消息。这种架构不支持消息漫游，因此，要把数据持久化，这里，考虑用mysql实现聊天消息的持久化。然而，高并发下，mysql很可能撑不住，从而出现消息的丢失问题。因此，引入消息队列rabbitmq,确保消息可靠传输。另外，一般来说，聊天消息的数据量是巨大的，并发量也是巨大的，为提高数据库读写效率和确保存储巨大的数据量，这里用sharding-sphere分库分表。完整的架构图如下：

二、效果展示

1.在线聊天

在线聊天：
https://www.bilibili.com/video/BV1wR4y1K7KH/

2.离线消息

如果消息接收者不在线，则将离线消息保存到redis中，用户上线后，从redis拉取离线消息
https://www.bilibili.com/video/BV1AR4y1K7tV/

3.消息漫游

如果消息接收者长期不在线，则将保存在redis中的消息删除，以免占用redis宝贵的资源。离线消息在redis中过期后，会自动生成一个标识，告诉该用户在mysql中有未读的消息，用户上线后，从mysql中拉取离线消息，并可自主选择拉取历史消息。
https://www.bilibili.com/video/BV1AR4y1K7tV/

4.sharding-sphere分库分表

通过分库分表实现数据库的横向拓展，减轻单表单数据库的读写压力
https://www.bilibili.com/video/BV1DF411M7NX/?spm_id_from=autoNext

---

三、代码实现

1.netty服务器

netty抽象出两组线程池，BossGroup专门负责接收客户端的连接，WorkerGroup专门负责网络的读写

//设置BossGroup和WorkerGroup
 NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
 NioEventLoopGroup workerGroup = new NioEventLoopGroup();
 

接下来基本上就是模板代码，设置一下服务器的各种参数，比如channel，pipeline，handler等，handler用来监听各种事件，netty基于事件驱动的特性就是通过handler来实现的。这里注意一下，我是把netty放到web容器中启动的，启动的时候需要手动把netty服务器注册到nacos注册中心。

    public void run() throws InterruptedException {
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        try {
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childHandler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel channel) {
                            //获取到pipeline
                            channel.pipeline().addLast(new HttpServerCodec());
                            channel.pipeline().addLast(new ChunkedWriteHandler());
                            channel.pipeline().addLast(new HttpObjectAggregator(8192));
                            channel.pipeline().addLast(new WebSocketServerProtocolHandler("/chat"));
                            //下面的WebSocketHandler是我自定义的handler
                            //这个handler用来监听channel上的各种事件
                            //比如用户上线，离线，发消息等
                            channel.pipeline().addLast(new WebSocketHandler());
                        }
                    });

            System.out.println("================服务器启动================");
            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
            System.out.println("================注册到nacos================");
            NamingService namingService = nacosServiceManager.getNamingService(nacosDiscoveryProperties.getNacosProperties());
            namingService.registerInstance(name, ip, port);
            channelFuture.channel().closeFuture().sync();
        } catch (NacosException ignored) {
            System.out.println("ip获取失败");
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

handler在netty中就相当于是springmvc中的controller层，下面就是自定义的WebSocketHandler监听并处理各种事件的代码
其中channelMap是一个静态变量，用来保存用户id到对应的channel的映射。
通过它，我们可以找到目标用户在哪个channel,从而通过该channel把消息正确地发给目标用户。其中ChannelHandlerContext类可以获取各种上下文信息，比如用户的ip地址，所在的channel,pipeline等等，还可以通过channel的attr方法设置自定义的属性，比如给channel设置用户id属性，这样拿到channel，就知道这个channel是属于哪个用户的了

 //绑定用户id和channel
    public static ConcurrentHashMap<String, Channel> channelMap = new ConcurrentHashMap<>();

//接收并处理消息
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame msg) throws Exception {
        System.out.println("服务器端收到消息："   msg.text());
        Channel channel = channelHandlerContext.channel();
        ChatMessage chatMessage = new ChatMessage();
        JSONObject json = JSON.parseObject(msg.text());
        Integer msgType = json.getInteger("msgType");
        String jwt = json.getString("msg");
        Long sender = json.getLong("sender");
        Long receiver = json.getLong("receiver");
        //初始化，绑定用户和channel，并注册到channelMap
        if (msgType == 1) {
            Claims claims = JwtUtil.parseJWT(jwt);
            String id = String.valueOf(claims.get("sub"));
            if (id != null) {
                messageUtil.online(channel, id);
                channelMap.put(id, channel);
            } else {
                System.out.println("token错误");
            }
        }
        //聊天消息
        else if (msgType == 2) {
            String message = json.getString("msg");
            chatMessage.setMsgType(2);
            chatMessage.setSender(sender);
            chatMessage.setReceiver(receiver);
            chatMessage.setMessage(message);
            //确认是本人发的消息，不是伪造的
            if (channelMap.get(String.valueOf(sender)) == channel) {
                //根据接收者的id获取接收者的channel
                Channel toChannel = getChannelByUserId(String.valueOf(receiver));
                messageUtil.sendMessage(chatMessage, toChannel);
            }
        }
    }

    //将当前channel加入到channelGroup
    //有用户上线
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        // TODO 将该客户加入聊天的信息推送给其他在线的客户端
         System.out.println("客户端："   channel.remoteAddress()   "加入聊天"   DateNow);
    }

    //用户离线
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
//        System.out.println("channel是：" ctx.channel());
        Channel channel = ctx.channel();
        //获取该channel的用户id
        AttributeKey<String> key = AttributeKey.valueOf("user");
        String userId = channel.attr(key).get();
        //redis中删除该用户的路由
        messageUtil.remove(userId);
        //channelMap中删除该用户
        channelMap.remove(userId);
        //redis中删除该用户
        String redisKey = "chat:route:"   userId;
        redisTemplate.delete(redisKey);
        System.out.println("handlerAdded被调用"   ctx.channel().id().asLongText());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.out.println("发生异常");
        System.out.println(cause);
        ctx.close();
    }
   // 根据用户id获取该用户的通道
    public Channel getChannelByUserId(String userId) {
        return channelMap.get(userId);
    }

下面是的代码是我自定义的消息工具类MessageUtil，在netty中相当于springmvc中的service层，用来处理具体的业务逻辑。

      //上线一个用户
    
    public void online(Channel channel, String userId) throws UnknownHostException {
        //绑定channel和用户的id
        AttributeKey<String> key = AttributeKey.valueOf("user");
        channel.attr(key).set(userId);
        String ip4 = String.valueOf(Inet4Address.getLocalHost());
        String ip = ip4.substring(ip4.lastIndexOf("/")   1);
        //redis中设置user到服务器的路由
        redisTemplate.opsForValue().set("chat:route:"   userId, ip   ":"   port);
        //TODO 从redis中取 chat:message:expired:" uid，判断是否存在，存在说明该用户有过期离线消息，去MYSQL中查找
        //TODO 并且，不要直接连MYSQL查找，不然会增加MYSQL的压力，而且做不到异步
        boolean hasKey = Boolean.TRUE.equals(redisTemplate.hasKey("chat:message:expired:"   userId));
        if (hasKey) {
            //告诉rabbitmq需要从Mysql拉取离线消息的用户id
            //message微服务监听mq中需要拉取离线消息的用户id,然后从mysql读取该用户的离线消息，然后发送回消息队列，该服务器监听消息队列，如收到
            // 离线消息则把消息发送给该用户
            rabbitTemplate.convertAndSend("message-pull-exchange", "message.users.pull.message", userId);
        }
        // 从redis中拉取离线消息
        Map<Object, Object> map = redisTemplate.opsForHash().entries("chat:message:"   userId);
        for (Object value : map.values()) {
            System.out.println(value);
            JSONObject chatMessage = JSONObject.parseObject((String) value);
            Long id = chatMessage.getLongValue("id");
            channel.writeAndFlush(new TextWebSocketFrame((String) value));
            //TODO 发送到消息队列告诉mysql该消息已签收
            System.out.println("发送到消息队列，签收消息"   id);
            rabbitTemplate.convertAndSend("message-signed-exchange", "message.signed.message", id);
        }
        //拉取完离线消息后从redis中删除消息
        redisTemplate.delete("chat:message:"   userId);
    }

    public void sendMessage(ChatMessage chatMessage, Channel toChannel) {
        Long userId = chatMessage.getReceiver();
        String msg = chatMessage.getMessage();
        String msgId = String.valueOf(chatMessage.getId());
        String chatMessageJson = JSONObject.toJSONString(chatMessage);
        //判断一下该用户是否连接的是本服务器
        if (toChannel != null) {
            //连接的是本服务器就直接发送就可以了
            toChannel.writeAndFlush(new TextWebSocketFrame(LocalDateTime.now()   ":     "   chatMessageJson));
            //消息设置为已接受，然后存入数据库
            chatMessage.setSigned(1);
            rabbitTemplate.convertAndSend("message-save-exchange", "message.save.message", chatMessage);
        } else {
            //判断一下该用户是否连接的是其他服务器，且在线
            boolean hasKey = Boolean.TRUE.equals(redisTemplate.hasKey("chat:route:"   userId));
            if (hasKey) {
                //目标用户已上线且在其他服务器中，通过存在redis里的路由信息得知目标用户所在的服务器，通过mq转发到该服务器
                //发送到mq中，然后数据库mysql消费消息，把所有的消息都持久化
                rabbitTemplate.convertAndSend("message-save-exchange", "message.save.message", chatMessage);
                //TODO 每台服务器都创建并监听自己的队列，通过存在redis里的路由信息得知目标用户所在的服务器，发送到rabbitmq相应的队列中，让目标用户所在的服务器接收并转发消息
            } else {
                //目标用户已离线
                // 发送到redis中，该用户上线后自动拉取离线消息
                long currentTimeMillis = System.currentTimeMillis();
                HashMap<String, String> chatMessageHashMap = new HashMap<>();
                chatMessageHashMap.put(msgId, chatMessageJson);
                redisTemplate.opsForHash().putAll("chat:message:"   userId, chatMessageHashMap);
                //给离线消息设置过期时间,10天后自动过期，用户想再得到离线消息就要从MYSQL中取
                redisTemplate.expire("chat:message:"   userId, 2, TimeUnit.MINUTES);
                //发送到mq中，然后数据库mysql消费消息，把所有的消息都持久化
                rabbitTemplate.convertAndSend("message-save-exchange", "message.save.message", chatMessage);
            }
        }
    }
    public void remove(String userId) {
        //redis中删除该用户的路由
        redisTemplate.delete("chat:route:"   userId);
    }

2.消息处理微服务

该模块用来对用户的聊天消息进行各种处理，比如消息签收，消息拉取，消息持久化等。通过消息队列收发各种消息，并开启手动确认，确保消息不丢失，实现了对其他微服务的异步解耦。

把消息保存到mysql

    //聊天消息持久化到MYSQL
    @RabbitHandler
    public void listener(ChatMessage chatMessage, Channel channel, Message message) {
        System.out.println("mysql收到消息："   chatMessage);
        //TODO 存入mysql数据库
        try {
            messageService.save(chatMessage);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            //TODO 把错误写入日志
            System.out.println("消息："   chatMessage   "写入数据库失败");
            System.out.println(e.getMessage());
        }
    }

签收消息

    //确认消息已签收
    @RabbitHandler
    public void signMessage(Long id, Channel channel, Message message) {
        System.out.println("mysql中消息："   id   "已确认签收");
        //TODO 存入mysql数据库
        try {
            messageService.signMessage(id);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            //TODO 把错误写入日志
            System.out.println("消息："   id   "确认失败");
            System.out.println(e.getMessage());
        }
    }

从mysql中拉取离线消息并发送回消息队列，等待netty服务器签收消息

    //从mysql中拉取离线消息
    @RabbitHandler
    public void getUsersPullMessage(String id, Channel channel, Message message) {
        System.out.println("用户："   id   "需要从mysql中拉取离线消息");
        Long uid = Long.valueOf(id);
        //TODO 存入mysql数据库
        try {
            //从mysql中拉取离线消息并发送回消息队列
            messageService.pullAndSendMessage(uid);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            //TODO 把错误写入日志
            System.out.println("消息："   id   "确认失败");
            System.out.println(e.getMessage());
        }
    }

3.使用sharding-sphere分库分表

使用sharding-sphere分库分表最核心的就是确定分库分表策略，并配置好sharding-sphere,接下来就可以像访问一个数据库一张表一样来访问多个数据库多个表的数据。这里简单地通过消息id和用户id来分库分表

以下是我的sharding-sphere配置

schemaName: sharding_chat
#
dataSources:
  ds_0:
    url: jdbc:mysql://127.0.0.1:3306/chat?useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai&useSSL=false
    username: root
    password: admin
    connectionTimeoutMilliseconds: 30000
    idleTimeoutMilliseconds: 60000
    maxLifetimeMilliseconds: 1800000
    maxPoolSize: 50
    minPoolSize: 1
  ds_1:
    url: jdbc:mysql://192.168.231.136:3306/chat?useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai&useSSL=false
    username: root
    password: admin
    connectionTimeoutMilliseconds: 30000
    idleTimeoutMilliseconds: 60000
    maxLifetimeMilliseconds: 1800000
    maxPoolSize: 50
    minPoolSize: 1
#
rules:
- !SHARDING
  tables:
    chat_message:
      actualDataNodes: ds_${0..1}.chat_message_${0..1}
      tableStrategy:
        standard:
          shardingColumn: id
          shardingAlgorithmName: chat_message_inline
      keyGenerateStrategy:
        column: id
        keyGeneratorName: snowflake

  defaultDatabaseStrategy:
    standard:
      shardingColumn: sender
      shardingAlgorithmName: chat_database_inline

  shardingAlgorithms:
    chat_database_inline:
      type: INLINE
      props:
        algorithm-expression: ds_${sender % 2}
    chat_message_inline:
      type: INLINE
      props:
        algorithm-expression: chat_message_${id % 2}

总结

这篇好文章是转载于：学新通技术网