监控易解读（2）：产品架构——实现IT、机房动环与物联网设备的统一监控与管理

使用netty完成websocket聊天

1、Netty配置类

import cn.hutool.extra.spring.SpringUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
@ConditionalOnProperty(prefix = NettyProperties.PREFIX, name = "enable", havingValue = "true")
@ConditionalOnClass({NettyBootstrap.class, NettyProperties.class})
@EnableConfigurationProperties(NettyProperties.class)
public class NettyConfig {
    private final NettyProperties nettyProperties;
    /**
     * 主从调度
     */
    private final EventLoopGroup masterGroup;
    private final EventLoopGroup slaveGroup;
    /**
     * server
     */
    private final ServerBootstrap serverBootstrap;
    private final ThreadPoolTaskExecutor threadPoolTaskExecutor;
    private final ApplicationEventPublisher eventPublisher;


    public NettyConfig(NettyProperties nettyProperties, ThreadPoolTaskExecutor threadPoolTaskExecutor, ApplicationEventPublisher eventPublisher) {
        this.masterGroup = new NioEventLoopGroup();
        this.slaveGroup = new NioEventLoopGroup();
        this.serverBootstrap = new ServerBootstrap();

        this.nettyProperties = nettyProperties;
        this.threadPoolTaskExecutor = threadPoolTaskExecutor;
        this.eventPublisher = eventPublisher;
    }

    @Bean
    @ConditionalOnMissingBean
    public NettyBootstrap nettyBootstrap() {
        return new NettyBootstrap(nettyProperties, masterGroup, slaveGroup, serverBootstrap, eventPublisher);
    }
}

2、Netty属性类

package com.chat.config;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;

/**
 * xx模块 - NettyProperties
 *
 * @author 李子耀
 * @version 1.0.0
 * @className NettyProperties
 * @description NettyProperties
 * @date 2023-8-17 14:39
 */
@Data
@ConfigurationProperties(prefix = NettyProperties.PREFIX)
public class NettyProperties {
    public static final String PREFIX = "netty";
    /**
     * true 是否激活
     */
    private Boolean enable;
    /**
     * 10.10.10.147
     */
    private String ip;
    /**
     * 9002
     */
    private Integer port;
    /**
     * jdz-china-ws
     */
    private String serverName;

    /**
     * dev
     */
    private String namespace;
    /**
     * 192.168.1.208:8848
     */
    private String serverAddr;

    /**
     * 默认参数 请求路径
     */
    private String path = "/";
    /**
     * 默认参数 是否需要认证
     */
    private Boolean auth = false;
    /**
     * 默认参数认证 key
     */
    private String authKey = "token";
    /**
     * 是否启用 评价功能 默认不启动吧，这个是延时队列
     */
    private Boolean enableRating = false;
    /**
     * 用户名
     */
    private String username;
    /**
     * 密码
     */
    private String password;
}

3、Netty启动类

package com.lzy.chat.config;

import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingService;
import com.lzy.chat.handler.AuthenticationHandler;
import com.lzy.chat.handler.MyMessageHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationEventPublisher;

import javax.annotation.PostConstruct;
import java.util.Properties;

/**
 * netty聊天模块 - NettyBootstrap
 *
 * @author 李子耀
 * @version 1.0.0
 * @className NettyBootstrap
 * @description NettyBootstrap
 * @date 2023-8-17 15:47
 */


@Slf4j
public class NettyBootstrap {
    private final NettyProperties nettyProperties;
    /**
     * 主从调度
     */
    private final EventLoopGroup masterGroup;
    private final EventLoopGroup slaveGroup;
    /**
     * server
     */
    private final ServerBootstrap serverBootstrap;

    private final ApplicationEventPublisher eventPublisher;

    public NettyBootstrap(NettyProperties nettyProperties, EventLoopGroup masterGroup, EventLoopGroup slaveGroup, ServerBootstrap serverBootstrap, ApplicationEventPublisher eventPublisher) {
        this.nettyProperties = nettyProperties;
        this.masterGroup = masterGroup;
        this.slaveGroup = slaveGroup;
        this.serverBootstrap = serverBootstrap;
        this.eventPublisher = eventPublisher;
    }

    @PostConstruct
    public void start() {
        // 注册到 Nacos 服务中
        registerNamingService();
        log.info("netty 注册nacos 成功");

        // 启动 Netty 服务器
        startServer();
        log.info("netty 聊天服务9002 加载成功");
        // 等待服务器关闭
        // 这里是等待服务器关闭的代码，可以根据您的实际情况进行编写
        // ...
    }

    /**
     * 注册到 Nacos 服务中
     */
    private void registerNamingService() {
        try {

            Properties properties = new Properties();
            properties.put("namespace", nettyProperties.getNamespace());
            properties.put("serverAddr", nettyProperties.getServerAddr());
            properties.put("username", nettyProperties.getUsername());
            properties.put("password", nettyProperties.getPassword());
            NamingService namingService = NamingFactory.createNamingService(properties);
            namingService.registerInstance(
                    nettyProperties.getServerName(),
                    nettyProperties.getIp(),
                    nettyProperties.getPort());
            log.info("当前netty注册Ip{},port{}", nettyProperties.getIp(), nettyProperties.getPort());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 启动服务
     */
    public void startServer() {
        serverBootstrap.group(masterGroup, slaveGroup).channel(NioServerSocketChannel.class)
                .childHandler(getChannelInitializer())
                .bind(nettyProperties.getPort());

        log.info("netty服务启动完成:{}", nettyProperties.getPort());
    }


    /**
     * 得到通道初始化
     *
     * @return {@link ChannelInitializer}<{@link SocketChannel}>
     */
    private ChannelInitializer<SocketChannel> getChannelInitializer() {
        return new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) {
                ChannelPipeline channelPipeline = ch.pipeline();
                // 添加HTTP编解码器
                channelPipeline.addLast(new HttpServerCodec());
                channelPipeline.addLast(new ChunkedWriteHandler());
                // 添加HTTP对象聚合器，将HTTP消息的多个部分合并成一个完整的HTTP消息
                channelPipeline.addLast(new HttpObjectAggregator(1024 * 64));
                //增加授权配置,设置频道userId并转换路由
                channelPipeline.addLast(new AuthenticationHandler(nettyProperties));
                //增加路由配置。
                channelPipeline.addLast(new WebSocketServerProtocolHandler(nettyProperties.getPath()));
                //业务处理器
                channelPipeline.addLast(new MyMessageHandler(eventPublisher));
            }
        };
    }
}

4、消息处理

package com.chat.handler;

import com.alibaba.fastjson.JSON;
import com.chat.constant.ChatType;
import com.chat.event.ChatRatingEvent;
import com.chat.pojo.dto.ChannelDTO;
import com.chat.socket.*;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationEventPublisher;

import java.util.Map;

/**
 * xx模块 - MessageHandler
 *
 * @author 李子耀
 * @version 1.0.0
 * @className MessageHandler
 * @description MessageHandler
 * @date 2023-8-14 17:15
 */

@Slf4j
public class MyMessageHandler extends SimpleChannelInboundHandler<WebSocketFrame> {

    private final ApplicationEventPublisher applicationEventPublisher;

    public MyMessageHandler(ApplicationEventPublisher applicationEventPublisher) {
        this.applicationEventPublisher = applicationEventPublisher;
    }

    /**
     * 消息处理程序
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) {
        if (frame instanceof TextWebSocketFrame) {
            // 处理文本帧
            TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
            String message = textFrame.text();
            // 处理接收到的文本消息
            processTextMessage(ctx, message);
        } else {
            // 处理其他类型的帧，如二进制帧等
            // ...
            log.info("处理其他类型的帧，如二进制帧等");
        }
    }

    /**
     * 处理文本消息
     *
     * @param ctx     ctx
     * @param message 消息
     */
    private void processTextMessage(ChannelHandlerContext ctx, String message) {
        DefaultMessage msg = JSON.parseObject(message, DefaultMessage.class);
        String channelId = ctx.channel().id().asLongText();
        msg.setChannelId(channelId);
        log.info("当前的chanelID:{}", channelId);
        msg.setSenderId(ctx.channel().attr(Attributes.USER_ID).get());
        String chatType = msg.getChatType();
        if (ChatType.GROUP.equals(chatType)) {
            this.groupChat(msg);
        } else if (ChatType.PRIVATE.equals(chatType)) {
            String messageType = msg.getMessageType();
            if ("10".equals(messageType)) {
                ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(msg)));
            } else {
                //有真实聊天记录
                MessageUtil.sendMsgAndInsertDB(msg);
            }
        }
        Map<String, MessageExtHandler> map = SpringContextHolder.getBeansOfType(MessageExtHandler.class);
        map.forEach((k, v) -> v.handle(ctx, msg));
        ctx.fireChannelRead(msg);
    }

    private void groupChat(Message msg) {
        log.info("群聊消息msg:{}", msg);
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        ChannelGroupManager.add(ctx.channel());
        String channelId = ctx.channel().id().asLongText();
        log.info("客户端已连接，ID是：" + channelId);
        //做一个频道id 鉴权。
        //京东，每次都是一个频道都是 新 的 聊天。
        //新建一个对话记录。 这个 如果有客户聊了天，就是可以评价的.增加评价
        //发布事件，放入延时队列 就一个channelId
        applicationEventPublisher.publishEvent(new ChatRatingEvent(channelId));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ChannelGroupManager.remove(ctx.channel());
        ctx.channel().close();
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        ChannelGroupManager.remove(ctx.channel());
        String channelId = ctx.channel().id().asLongText();
        log.info("客户端已断开连接，ID是：" + channelId);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) {
//            log.info("握手完成，调用回调方法,可以发送消息");
            String channelId = ctx.channel().id().asLongText();
            ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(new ChannelDTO(channelId))));
        } else {
//            log.info("没有握手完成，继续触发用户事件");
            super.userEventTriggered(ctx, evt);
        }
    }
}

5、消息处理类

import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.DefaultEventExecutor;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;


public class ChannelGroupManager {
    private static ChannelGroup group = new DefaultChannelGroup(new DefaultEventExecutor());

    /**
     * 添加
     *
     * @param channel
     * @return
     */
    public static boolean add(Channel channel) {
        if (null == channel) {
            return false;
        }
        return group.add(channel);
    }

    /**
     * 移除
     *
     * @param channel
     * @return
     */
    public static boolean remove(Channel channel) {
        if (null == channel) {
            return false;
        }
        return group.remove(channel);
    }

    /**
     * 根据用户Id 获取channel
     * @param userId
     * @return
     */
    public static List<Channel> channel(String userId) {
        if (StringUtils.isEmpty(userId)) {
            return null;
        }
        List<Channel> reference = new ArrayList<>();
        group.forEach(channel -> {
          String id =  channel.attr(Attributes.USER_ID).get();
          if(userId.equals(id)){
              reference.add(channel);
              return;
          }
        });
        return reference;
    }
}

6、相关的基础类

public interface Message {
    /**
     * 聊天类型 群里：1 私聊：0
     * @return
     */
    String getChatType();

    /**
     * 消息类型 心跳消息：10 文件消息：20 emoji表情：30 HTML文本：40
     * @return
     */
    String getMessageType();

    /**
     * 消息Id，后端自动填充
     * @return
     */
    String getId();

    /**
     * 消息内容，具体格式前端定义就行
     * @return
     */
    String getContent();

    /**
     * 接受者 如果是群聊则是房间id，需要根据房间号找到房间里的人 转换一下
     * @return
     */
    String getReceiverId();

    /**
     * 签收状态 已签收：1 0未签收：0
     * @return
     */
    String getSignState();
}



import io.netty.channel.ChannelHandlerContext;
// 消息扩展处理接口
public interface MessageExtHandler {
    void handle(ChannelHandlerContext ctx, Message message);
}


import lombok.Data;
@Data
public abstract class AbstractMessage implements Message {

    /**
     * @ignore
     */
    private String id;

    /**
     *
     * 消息内容 具体格式前端定义就好
     */
    private String content;

    /**
     * 聊天类型   群里 1 私聊 0
     */
    private String chatType;

    /**
     * 消息类型 心跳消息 10 文件消息 20 emoji 表情 30 HTML文本 40
     */
    private String messageType;

    /**
     * 消息发送者,发送方可以不设值，后端自动设值
     */
    private String senderId;

    /**
     * 消息接受者,如果是群聊则是房间号，需要根据房间号找到房间里的人 转换一下
     * @required
     */
    private String receiverId;

    /**
     * 签收状态 1:签收  0:签收
     */
    private String signState;
}

//消息类型
public abstract class AbstractMessageType {

    /**
     * 心跳消息
     */
    public static final String HEART = "10";

    /**
     * 文件消息
     */
    public static final String FILE = "20";

    /**
     * emoji表情消息
     */
    public static final String EMOJI = "30";

    /**
     * html 文本内容消息
     */
    public static final String HTML = "40";
}



import io.netty.util.AttributeKey;
public class Attributes {

 public static final AttributeKey<String> USER_ID = AttributeKey.newInstance("userId");

 public static final AttributeKey<String> CODE = AttributeKey.newInstance("code");
}

import lombok.Data;
@Data
public class DefaultMessage extends AbstractMessage {
    private String channelId;
}

import com.baomidou.mybatisplus.annotation.FieldFill;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableLogic;
import com.baomidou.mybatisplus.annotation.TableName;
import com.chat.socket.AbstractMessage;
import lombok.Data;

import java.util.Date;

@Data
@TableName(chat_msg")
public class ChatMsgPO extends AbstractMessage {

    private String channelId;
    /**
     * 创建日期
     */
    @TableField(value = "create_date", fill = FieldFill.INSERT)
    protected Date createDate;


    /**
     * 修改操作日期
     */
    @TableField(value = "update_date", fill = FieldFill.UPDATE)
    protected Date updateDate;

    @TableLogic()
    @TableField(fill = FieldFill.INSERT)
    private String enabled;
}


public abstract class ChatType {
    /**
     * 群聊
     */
    public static final String GROUP = "1";

    /**
     * 私聊
     */
    public static final String PRIVATE = "0";
}

public abstract class SignState {
    /**
     * 签收
     */
    public static final String SIGNED = "Y";

    /**
     * 未签收
     */
    public static final String UNSIGNED = "N";
}

7、结果结果导向

import com.alibaba.fastjson.JSON;
import com.chat.pojo.po.ChatMsgPO;
import com.chat.service.ChatMsgService;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.util.StringUtils;

import java.util.List;

public class MessageUtil {
    /**
     * 日志对象
     */
    protected static Logger logger = LoggerFactory.getLogger(MessageUtil.class);

    static ChatMsgService chatMsgService = SpringContextHolder.getBean(ChatMsgService.class);

    /**
     * 发送消息并且存库
     * @param msg
     * @return
     */
    public static boolean sendMsgAndInsertDB(Message msg) {
        checkInit();
        ChatMsgPO chatMsg = new ChatMsgPO();
        BeanUtils.copyProperties(msg,chatMsg);
        chatMsg.setSignState(SignState.UNSIGNED);
        chatMsgService.saveChatMsg(chatMsg);
        return pushMsg(msg);
    }

    /**
     * 检测
     */
    private static void checkInit() {
        if (chatMsgService == null) {
            chatMsgService = SpringContextHolder.getBean(ChatMsgService.class);
        }
    }

    /**
     * 发送聊天消息
     */
    public static boolean pushMsg(Message msg) {
        String receiverId = msg.getReceiverId();
        if (StringUtils.isEmpty(receiverId)) {
            logger.warn("接受者用户Id为空，信息丢弃:{}", JSON.toJSONString(msg));
            return false;
        }
        List<Channel> channels = ChannelGroupManager.channel(receiverId);
        if (channels.isEmpty()) {
            logger.warn("消息接受者不在线：{}", JSON.toJSONString(msg));
            return false;
        }
        channels.forEach(item -> item.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(msg))));
        return true;
    }
}