init

im-server/src/main/java/com/bx/imserver/IMServerApp.java

@@ -0,0 +1,19 @@
+package com.bx.imserver;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.annotation.EnableScheduling;
+
+@EnableAsync
+@EnableScheduling
+@ComponentScan(basePackages = {"com.bx"})
+@SpringBootApplication
+public class IMServerApp {
+
+    public static void main(String[] args) {
+        SpringApplication.run(IMServerApp.class, args);
+    }
+
+}

im-server/src/main/java/com/bx/imserver/constant/ChannelAttrKey.java

@@ -0,0 +1,21 @@
+package com.bx.imserver.constant;
+
+public final class ChannelAttrKey {
+
+    private ChannelAttrKey() {
+    }
+
+    /**
+     * 用户ID
+     */
+    public static final String USER_ID = "USER_ID";
+    /**
+     * 终端类型
+     */
+    public static final String TERMINAL_TYPE = "TERMINAL_TYPE";
+    /**
+     * 心跳次数
+     */
+    public static final String HEARTBEAT_TIMES = "HEARTBEAT_TIMES";
+
+}

im-server/src/main/java/com/bx/imserver/netty/IMChannelHandler.java

@@ -0,0 +1,99 @@
+package com.bx.imserver.netty;
+
+import com.bx.imcommon.contant.IMRedisKey;
+import com.bx.imcommon.enums.IMCmdType;
+import com.bx.imcommon.model.IMSendInfo;
+import com.bx.imcommon.mq.RedisMQTemplate;
+import com.bx.imserver.constant.ChannelAttrKey;
+import com.bx.imserver.netty.processor.AbstractMessageProcessor;
+import com.bx.imserver.netty.processor.ProcessorFactory;
+import com.bx.imserver.util.SpringContextHolder;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.timeout.IdleState;
+import io.netty.handler.timeout.IdleStateEvent;
+import io.netty.util.AttributeKey;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * WebSocket 长连接下 文本帧的处理器
+ * 实现浏览器发送文本回写
+ * 浏览器连接状态监控
+ */
+@Slf4j
+public class IMChannelHandler extends SimpleChannelInboundHandler<IMSendInfo> {
+
+    /**
+     * 读取到消息后进行处理
+     *
+     * @param ctx      channel上下文
+     * @param sendInfo 发送消息
+     */
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, IMSendInfo sendInfo) {
+        // 创建处理器进行处理
+        AbstractMessageProcessor processor = ProcessorFactory.createProcessor(IMCmdType.fromCode(sendInfo.getCmd()));
+        processor.process(ctx, processor.transForm(sendInfo.getData()));
+    }
+
+    /**
+     * 出现异常的处理 打印报错日志
+     *
+     * @param ctx   channel上下文
+     * @param cause 异常信息
+     */
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+        log.error(cause.getMessage());
+        //关闭上下文
+        //ctx.close();
+    }
+
+    /**
+     * 监控浏览器上线
+     *
+     * @param ctx channel上下文
+     */
+    @Override
+    public void handlerAdded(ChannelHandlerContext ctx) {
+        log.info(ctx.channel().id().asLongText() + "连接");
+    }
+
+    @Override
+    public void handlerRemoved(ChannelHandlerContext ctx) {
+        AttributeKey<Long> userIdAttr = AttributeKey.valueOf(ChannelAttrKey.USER_ID);
+        Long userId = ctx.channel().attr(userIdAttr).get();
+        AttributeKey<Integer> terminalAttr = AttributeKey.valueOf(ChannelAttrKey.TERMINAL_TYPE);
+        Integer terminal = ctx.channel().attr(terminalAttr).get();
+        ChannelHandlerContext context = UserChannelCtxMap.getChannelCtx(userId, terminal);
+        // 判断一下，避免异地登录导致的误删
+        if (context != null && ctx.channel().id().equals(context.channel().id())) {
+            // 移除channel
+            UserChannelCtxMap.removeChannelCtx(userId, terminal);
+            // 用户下线
+            RedisMQTemplate redisTemplate = SpringContextHolder.getBean(RedisMQTemplate.class);
+            String key = String.join(":", IMRedisKey.IM_USER_SERVER_ID, userId.toString(), terminal.toString());
+            redisTemplate.delete(key);
+            log.info("断开连接,userId:{},终端类型:{},{}", userId, terminal, ctx.channel().id().asLongText());
+        }
+    }
+
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        if (evt instanceof IdleStateEvent) {
+            IdleState state = ((IdleStateEvent) evt).state();
+            if (state == IdleState.READER_IDLE) {
+                // 在规定时间内没有收到客户端的上行数据, 主动断开连接
+                AttributeKey<Long> attr = AttributeKey.valueOf("USER_ID");
+                Long userId = ctx.channel().attr(attr).get();
+                AttributeKey<Integer> terminalAttr = AttributeKey.valueOf(ChannelAttrKey.TERMINAL_TYPE);
+                Integer terminal = ctx.channel().attr(terminalAttr).get();
+                log.info("心跳超时，即将断开连接,用户id:{},终端类型:{} ", userId, terminal);
+                ctx.channel().close();
+            }
+        } else {
+            super.userEventTriggered(ctx, evt);
+        }
+
+    }
+}

im-server/src/main/java/com/bx/imserver/netty/IMServer.java

@@ -0,0 +1,10 @@
+package com.bx.imserver.netty;
+
+public interface IMServer {
+
+    boolean isReady();
+
+    void start();
+
+    void stop();
+}

im-server/src/main/java/com/bx/imserver/netty/IMServerGroup.java

@@ -0,0 +1,55 @@
+package com.bx.imserver.netty;
+
+import com.bx.imcommon.contant.IMRedisKey;
+import com.bx.imcommon.mq.RedisMQTemplate;
+import jakarta.annotation.PreDestroy;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+@Slf4j
+@Component
+@AllArgsConstructor
+public class IMServerGroup implements CommandLineRunner {
+
+    public static volatile long serverId = 0;
+
+    private final RedisMQTemplate redisMQTemplate;
+
+    private final List<IMServer> imServers;
+
+    /***
+     * 判断服务器是否就绪
+     *
+     **/
+    public boolean isReady() {
+        for (IMServer imServer : imServers) {
+            if (!imServer.isReady()) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    @Override
+    public void run(String... args) {
+        // 初始化SERVER_ID
+        String key = IMRedisKey.IM_MAX_SERVER_ID;
+        serverId = redisMQTemplate.opsForValue().increment(key, 1);
+        // 启动服务
+        for (IMServer imServer : imServers) {
+            imServer.start();
+        }
+    }
+
+    @PreDestroy
+    public void destroy() {
+        // 停止服务
+        for (IMServer imServer : imServers) {
+            imServer.stop();
+        }
+    }
+}

im-server/src/main/java/com/bx/imserver/netty/UserChannelCtxMap.java

@@ -0,0 +1,46 @@
+package com.bx.imserver.netty;
+
+import io.netty.channel.ChannelHandlerContext;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class UserChannelCtxMap {
+
+    /**
+     *  维护userId和ctx的关联关系，格式:Map<userId,map<terminal，ctx>>
+     */
+    private static Map<Long, Map<Integer, ChannelHandlerContext>> channelMap = new ConcurrentHashMap();
+
+    public static void addChannelCtx(Long userId, Integer channel, ChannelHandlerContext ctx) {
+        channelMap.computeIfAbsent(userId, key -> new ConcurrentHashMap()).put(channel, ctx);
+    }
+
+    public static void removeChannelCtx(Long userId, Integer terminal) {
+        if (userId != null && terminal != null && channelMap.containsKey(userId)) {
+            Map<Integer, ChannelHandlerContext> userChannelMap = channelMap.get(userId);
+            userChannelMap.remove(terminal);
+            if (userChannelMap.isEmpty()) {
+                channelMap.remove(userId);
+            }
+        }
+    }
+
+    public static ChannelHandlerContext getChannelCtx(Long userId, Integer terminal) {
+        if (userId != null && terminal != null && channelMap.containsKey(userId)) {
+            Map<Integer, ChannelHandlerContext> userChannelMap = channelMap.get(userId);
+            if (userChannelMap.containsKey(terminal)) {
+                return userChannelMap.get(terminal);
+            }
+        }
+        return null;
+    }
+
+    public static Map<Integer, ChannelHandlerContext> getChannelCtx(Long userId) {
+        if (userId == null) {
+            return null;
+        }
+        return channelMap.get(userId);
+    }
+
+}

im-server/src/main/java/com/bx/imserver/netty/processor/AbstractMessageProcessor.java

@@ -0,0 +1,18 @@
+package com.bx.imserver.netty.processor;
+
+import io.netty.channel.ChannelHandlerContext;
+
+public abstract class AbstractMessageProcessor<T> {
+
+    public void process(ChannelHandlerContext ctx, T data) {
+    }
+
+    public void process(T data) {
+    }
+
+    public T transForm(Object o) {
+        return (T) o;
+    }
+
+
+}

im-server/src/main/java/com/bx/imserver/netty/processor/GroupMessageProcessor.java

@@ -0,0 +1,71 @@
+package com.bx.imserver.netty.processor;
+
+import cn.hutool.core.util.StrUtil;
+import com.bx.imcommon.contant.IMRedisKey;
+import com.bx.imcommon.enums.IMCmdType;
+import com.bx.imcommon.enums.IMSendCode;
+import com.bx.imcommon.model.IMRecvInfo;
+import com.bx.imcommon.model.IMSendInfo;
+import com.bx.imcommon.model.IMSendResult;
+import com.bx.imcommon.model.IMUserInfo;
+import com.bx.imcommon.mq.RedisMQTemplate;
+import com.bx.imserver.netty.UserChannelCtxMap;
+import io.netty.channel.ChannelHandlerContext;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.Objects;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class GroupMessageProcessor extends AbstractMessageProcessor<IMRecvInfo> {
+
+    private final RedisMQTemplate redisMQTemplate;
+
+    @Override
+    public void process(IMRecvInfo recvInfo) {
+        IMUserInfo sender = recvInfo.getSender();
+        List<IMUserInfo> receivers = recvInfo.getReceivers();
+        log.info("接收到群消息，发送者:{},接收用户数量:{}，内容:{}", sender.getId(), receivers.size(), recvInfo.getData());
+        for (IMUserInfo receiver : receivers) {
+            try {
+                ChannelHandlerContext channelCtx = UserChannelCtxMap.getChannelCtx(receiver.getId(), receiver.getTerminal());
+                if (!Objects.isNull(channelCtx)) {
+                    // 推送消息到用户
+                    IMSendInfo<Object> sendInfo = new IMSendInfo<>();
+                    sendInfo.setCmd(IMCmdType.GROUP_MESSAGE.code());
+                    sendInfo.setData(recvInfo.getData());
+                    channelCtx.channel().writeAndFlush(sendInfo);
+                    // 消息发送成功确认
+                    sendResult(recvInfo, receiver, IMSendCode.SUCCESS);
+
+                } else {
+                    // 消息发送成功确认
+                    sendResult(recvInfo, receiver, IMSendCode.NOT_FIND_CHANNEL);
+                    log.error("未找到channel,发送者:{},接收id:{}，内容:{}", sender.getId(), receiver.getId(), recvInfo.getData());
+                }
+            } catch (Exception e) {
+                // 消息发送失败确认
+                sendResult(recvInfo, receiver, IMSendCode.UNKONW_ERROR);
+                log.error("发送消息异常,发送者:{},接收id:{}，内容:{}", sender.getId(), receiver.getId(), recvInfo.getData());
+            }
+        }
+    }
+
+
+    private void sendResult(IMRecvInfo recvInfo, IMUserInfo receiver, IMSendCode sendCode) {
+        if (recvInfo.getSendResult()) {
+            IMSendResult<Object> result = new IMSendResult<>();
+            result.setSender(recvInfo.getSender());
+            result.setReceiver(receiver);
+            result.setCode(sendCode.code());
+            result.setData(recvInfo.getData());
+            // 推送到结果队列
+            String key = StrUtil.join(":",IMRedisKey.IM_RESULT_GROUP_QUEUE,recvInfo.getServiceName());
+            redisMQTemplate.opsForList().rightPush(key, result);
+        }
+    }
+}

im-server/src/main/java/com/bx/imserver/netty/processor/HeartbeatProcessor.java

@@ -0,0 +1,56 @@
+package com.bx.imserver.netty.processor;
+
+import cn.hutool.core.bean.BeanUtil;
+import com.bx.imcommon.contant.IMConstant;
+import com.bx.imcommon.contant.IMRedisKey;
+import com.bx.imcommon.enums.IMCmdType;
+import com.bx.imcommon.model.IMHeartbeatInfo;
+import com.bx.imcommon.model.IMSendInfo;
+import com.bx.imcommon.mq.RedisMQTemplate;
+import com.bx.imserver.constant.ChannelAttrKey;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.util.AttributeKey;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class HeartbeatProcessor extends AbstractMessageProcessor<IMHeartbeatInfo> {
+
+    private final RedisMQTemplate redisMQTemplate;
+
+    @Override
+    public void process(ChannelHandlerContext ctx, IMHeartbeatInfo beatInfo) {
+        // 响应ws
+        IMSendInfo<Object> sendInfo = new IMSendInfo<>();
+        sendInfo.setCmd(IMCmdType.HEART_BEAT.code());
+        ctx.channel().writeAndFlush(sendInfo);
+        // 设置属性
+        AttributeKey<Long> heartBeatAttr = AttributeKey.valueOf(ChannelAttrKey.HEARTBEAT_TIMES);
+        Long heartbeatTimes = ctx.channel().attr(heartBeatAttr).get();
+        ctx.channel().attr(heartBeatAttr).set(++heartbeatTimes);
+        if (heartbeatTimes % 10 == 0) {
+            // 每心跳10次，用户在线状态续一次命
+            AttributeKey<Long> userIdAttr = AttributeKey.valueOf(ChannelAttrKey.USER_ID);
+            Long userId = ctx.channel().attr(userIdAttr).get();
+            AttributeKey<Integer> terminalAttr = AttributeKey.valueOf(ChannelAttrKey.TERMINAL_TYPE);
+            Integer terminal = ctx.channel().attr(terminalAttr).get();
+            String key = String.join(":", IMRedisKey.IM_USER_SERVER_ID, userId.toString(), terminal.toString());
+            redisMQTemplate.expire(key, IMConstant.ONLINE_TIMEOUT_SECOND, TimeUnit.SECONDS);
+        }
+        AttributeKey<Long> userIdAttr = AttributeKey.valueOf(ChannelAttrKey.USER_ID);
+        Long userId = ctx.channel().attr(userIdAttr).get();
+        log.debug("心跳,userId:{},{}",userId,ctx.channel().id().asLongText());
+    }
+
+    @Override
+    public IMHeartbeatInfo transForm(Object o) {
+        HashMap map = (HashMap) o;
+        return BeanUtil.fillBeanWithMap(map, new IMHeartbeatInfo(), false);
+    }
+}

im-server/src/main/java/com/bx/imserver/netty/processor/LoginProcessor.java

@@ -0,0 +1,83 @@
+package com.bx.imserver.netty.processor;
+
+import cn.hutool.core.bean.BeanUtil;
+import com.alibaba.fastjson.JSON;
+import com.bx.imcommon.contant.IMConstant;
+import com.bx.imcommon.contant.IMRedisKey;
+import com.bx.imcommon.enums.IMCmdType;
+import com.bx.imcommon.model.IMLoginInfo;
+import com.bx.imcommon.model.IMSendInfo;
+import com.bx.imcommon.model.IMSessionInfo;
+import com.bx.imcommon.mq.RedisMQTemplate;
+import com.bx.imcommon.util.JwtUtil;
+import com.bx.imserver.constant.ChannelAttrKey;
+import com.bx.imserver.netty.IMServerGroup;
+import com.bx.imserver.netty.UserChannelCtxMap;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.util.AttributeKey;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class LoginProcessor extends AbstractMessageProcessor<IMLoginInfo> {
+
+    private final RedisMQTemplate redisMQTemplate;
+
+    @Value("${jwt.accessToken.secret}")
+    private String accessTokenSecret;
+
+    @Override
+    public  void process(ChannelHandlerContext ctx, IMLoginInfo loginInfo) {
+        if (!JwtUtil.checkSign(loginInfo.getAccessToken(), accessTokenSecret)) {
+            ctx.channel().close();
+            log.warn("用户token校验不通过，强制下线,token:{}", loginInfo.getAccessToken());
+            return;
+        }
+        String strInfo = JwtUtil.getInfo(loginInfo.getAccessToken());
+        IMSessionInfo sessionInfo = JSON.parseObject(strInfo, IMSessionInfo.class);
+        Long userId = sessionInfo.getUserId();
+        Integer terminal = sessionInfo.getTerminal();
+        log.info("用户登录，userId:{}", userId);
+        ChannelHandlerContext context = UserChannelCtxMap.getChannelCtx(userId, terminal);
+        if (context != null && !ctx.channel().id().equals(context.channel().id())) {
+            // 不允许多地登录,强制下线
+            IMSendInfo<Object> sendInfo = new IMSendInfo<>();
+            sendInfo.setCmd(IMCmdType.FORCE_LOGUT.code());
+            sendInfo.setData("您已在其他地方登录，将被强制下线");
+            context.channel().writeAndFlush(sendInfo);
+            log.info("异地登录，强制下线,userId:{}", userId);
+        }
+        // 绑定用户和channel
+        UserChannelCtxMap.addChannelCtx(userId, terminal, ctx);
+        // 设置用户id属性
+        AttributeKey<Long> userIdAttr = AttributeKey.valueOf(ChannelAttrKey.USER_ID);
+        ctx.channel().attr(userIdAttr).set(userId);
+        // 设置用户终端类型
+        AttributeKey<Integer> terminalAttr = AttributeKey.valueOf(ChannelAttrKey.TERMINAL_TYPE);
+        ctx.channel().attr(terminalAttr).set(terminal);
+        // 初始化心跳次数
+        AttributeKey<Long> heartBeatAttr = AttributeKey.valueOf(ChannelAttrKey.HEARTBEAT_TIMES);
+        ctx.channel().attr(heartBeatAttr).set(0L);
+        // 在redis上记录每个user的channelId，15秒没有心跳，则自动过期
+        String key = String.join(":", IMRedisKey.IM_USER_SERVER_ID, userId.toString(), terminal.toString());
+        redisMQTemplate.opsForValue().set(key, IMServerGroup.serverId, IMConstant.ONLINE_TIMEOUT_SECOND, TimeUnit.SECONDS);
+        // 响应ws
+        IMSendInfo<Object> sendInfo = new IMSendInfo<>();
+        sendInfo.setCmd(IMCmdType.LOGIN.code());
+        ctx.channel().writeAndFlush(sendInfo);
+    }
+
+
+    @Override
+    public IMLoginInfo transForm(Object o) {
+        HashMap map = (HashMap) o;
+        return BeanUtil.fillBeanWithMap(map, new IMLoginInfo(), false);
+    }
+}

im-server/src/main/java/com/bx/imserver/netty/processor/PrivateMessageProcessor.java

@@ -0,0 +1,67 @@
+package com.bx.imserver.netty.processor;
+
+import cn.hutool.core.util.StrUtil;
+import com.bx.imcommon.contant.IMRedisKey;
+import com.bx.imcommon.enums.IMCmdType;
+import com.bx.imcommon.enums.IMSendCode;
+import com.bx.imcommon.model.IMRecvInfo;
+import com.bx.imcommon.model.IMSendInfo;
+import com.bx.imcommon.model.IMSendResult;
+import com.bx.imcommon.model.IMUserInfo;
+import com.bx.imcommon.mq.RedisMQTemplate;
+import com.bx.imserver.netty.UserChannelCtxMap;
+import io.netty.channel.ChannelHandlerContext;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.Objects;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class PrivateMessageProcessor extends AbstractMessageProcessor<IMRecvInfo> {
+
+    private final RedisMQTemplate redisMQTemplate;
+
+    @Override
+    public void process(IMRecvInfo recvInfo) {
+        IMUserInfo sender = recvInfo.getSender();
+        IMUserInfo receiver = recvInfo.getReceivers().get(0);
+        log.info("接收到私聊消息，发送者:{},接收者:{}，内容:{}", sender.getId(), receiver.getId(), recvInfo.getData());
+        try {
+            ChannelHandlerContext channelCtx = UserChannelCtxMap.getChannelCtx(receiver.getId(), receiver.getTerminal());
+            if (!Objects.isNull(channelCtx)) {
+                // 推送消息到用户
+                IMSendInfo<Object> sendInfo = new IMSendInfo<>();
+                sendInfo.setCmd(IMCmdType.PRIVATE_MESSAGE.code());
+                sendInfo.setData(recvInfo.getData());
+                channelCtx.channel().writeAndFlush(sendInfo);
+                // 消息发送成功确认
+                sendResult(recvInfo, IMSendCode.SUCCESS);
+            } else {
+                // 消息推送失败确认
+                sendResult(recvInfo, IMSendCode.NOT_FIND_CHANNEL);
+                log.error("未找到channel，发送者:{},接收者:{}，内容:{}", sender.getId(), receiver.getId(), recvInfo.getData());
+            }
+        } catch (Exception e) {
+            // 消息推送失败确认
+            sendResult(recvInfo, IMSendCode.UNKONW_ERROR);
+            log.error("发送异常，发送者:{},接收者:{}，内容:{}", sender.getId(), receiver.getId(), recvInfo.getData(), e);
+        }
+
+    }
+
+    private void sendResult(IMRecvInfo recvInfo, IMSendCode sendCode) {
+        if (recvInfo.getSendResult()) {
+            IMSendResult<Object> result = new IMSendResult<>();
+            result.setSender(recvInfo.getSender());
+            result.setReceiver(recvInfo.getReceivers().get(0));
+            result.setCode(sendCode.code());
+            result.setData(recvInfo.getData());
+            // 推送到结果队列
+            String key = StrUtil.join(":",IMRedisKey.IM_RESULT_PRIVATE_QUEUE,recvInfo.getServiceName());
+            redisMQTemplate.opsForList().rightPush(key, result);
+        }
+    }
+}

im-server/src/main/java/com/bx/imserver/netty/processor/ProcessorFactory.java

@@ -0,0 +1,19 @@
+package com.bx.imserver.netty.processor;
+
+import com.bx.imcommon.enums.IMCmdType;
+import com.bx.imserver.util.SpringContextHolder;
+
+public class ProcessorFactory {
+
+    public static AbstractMessageProcessor createProcessor(IMCmdType cmd) {
+        return switch (cmd) {
+            case LOGIN->SpringContextHolder.getApplicationContext().getBean(LoginProcessor.class);
+            case HEART_BEAT -> SpringContextHolder.getApplicationContext().getBean(HeartbeatProcessor.class);
+            case PRIVATE_MESSAGE->SpringContextHolder.getApplicationContext().getBean(PrivateMessageProcessor.class);
+            case GROUP_MESSAGE->SpringContextHolder.getApplicationContext().getBean(GroupMessageProcessor.class);
+            case SYSTEM_MESSAGE->SpringContextHolder.getApplicationContext().getBean(SystemMessageProcessor.class);
+            default -> null;
+        };
+    }
+
+}

im-server/src/main/java/com/bx/imserver/netty/processor/SystemMessageProcessor.java

@@ -0,0 +1,67 @@
+package com.bx.imserver.netty.processor;
+
+import cn.hutool.core.util.StrUtil;
+import com.bx.imcommon.contant.IMRedisKey;
+import com.bx.imcommon.enums.IMCmdType;
+import com.bx.imcommon.enums.IMSendCode;
+import com.bx.imcommon.model.IMRecvInfo;
+import com.bx.imcommon.model.IMSendInfo;
+import com.bx.imcommon.model.IMSendResult;
+import com.bx.imcommon.model.IMUserInfo;
+import com.bx.imcommon.mq.RedisMQTemplate;
+import com.bx.imserver.netty.UserChannelCtxMap;
+import io.netty.channel.ChannelHandlerContext;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.Objects;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class SystemMessageProcessor extends AbstractMessageProcessor<IMRecvInfo> {
+
+    private final RedisMQTemplate redisMQTemplate;
+
+    @Override
+    public void process(IMRecvInfo recvInfo) {
+        log.info("接收到系统消息,接收用户数量:{}，内容:{}",  recvInfo.getReceivers().size(), recvInfo.getData());
+        for (IMUserInfo receiver : recvInfo.getReceivers()) {
+            try {
+                ChannelHandlerContext channelCtx =
+                    UserChannelCtxMap.getChannelCtx(receiver.getId(), receiver.getTerminal());
+                if (!Objects.isNull(channelCtx)) {
+                    // 推送消息到用户
+                    IMSendInfo<Object> sendInfo = new IMSendInfo<>();
+                    sendInfo.setCmd(IMCmdType.SYSTEM_MESSAGE.code());
+                    sendInfo.setData(recvInfo.getData());
+                    channelCtx.channel().writeAndFlush(sendInfo);
+                    // 消息发送成功确认
+                    sendResult(recvInfo, IMSendCode.SUCCESS);
+                } else {
+                    // 消息推送失败确认
+                    sendResult(recvInfo, IMSendCode.NOT_FIND_CHANNEL);
+                    log.error("未找到channel，接收者:{}，内容:{}", receiver.getId(), recvInfo.getData());
+                }
+            } catch (Exception e) {
+                // 消息推送失败确认
+                sendResult(recvInfo, IMSendCode.UNKONW_ERROR);
+                log.error("发送异常，,接收者:{}，内容:{}", receiver.getId(), recvInfo.getData(), e);
+            }
+        }
+
+    }
+
+    private void sendResult(IMRecvInfo recvInfo, IMSendCode sendCode) {
+        if (recvInfo.getSendResult()) {
+            IMSendResult<Object> result = new IMSendResult<>();
+            result.setReceiver(recvInfo.getReceivers().get(0));
+            result.setCode(sendCode.code());
+            result.setData(recvInfo.getData());
+            // 推送到结果队列
+            String key = StrUtil.join(":",IMRedisKey.IM_RESULT_SYSTEM_QUEUE,recvInfo.getServiceName());
+            redisMQTemplate.opsForList().rightPush(key, result);
+        }
+    }
+}

im-server/src/main/java/com/bx/imserver/netty/tcp/TcpSocketServer.java

@@ -0,0 +1,98 @@
+package com.bx.imserver.netty.tcp;
+
+import com.bx.imserver.netty.IMChannelHandler;
+import com.bx.imserver.netty.IMServer;
+import com.bx.imserver.netty.tcp.endecode.MessageProtocolDecoder;
+import com.bx.imserver.netty.tcp.endecode.MessageProtocolEncoder;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.*;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.timeout.IdleStateHandler;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * TCP服务器,用于连接非网页的客户端,协议格式： 8字节内容的长度+IMSendInfo的JSON序列化
+ *
+ * @author Blue
+ * @date 2022-11-20
+ */
+@Slf4j
+@Component
+@ConditionalOnProperty(prefix = "tcpsocket", value = "enable", havingValue = "true", matchIfMissing = false)
+public class TcpSocketServer implements IMServer {
+
+    private volatile boolean ready = false;
+
+    @Value("${tcpsocket.port}")
+    private int port;
+
+    private EventLoopGroup bossGroup;
+    private EventLoopGroup workGroup;
+
+    @Override
+    public boolean isReady() {
+        return ready;
+    }
+
+    @Override
+    public void start() {
+        ServerBootstrap bootstrap = new ServerBootstrap();
+        bossGroup = new NioEventLoopGroup();
+        workGroup = new NioEventLoopGroup();
+        // 设置为主从线程模型
+        bootstrap.group(bossGroup, workGroup)
+                // 设置服务端NIO通信类型
+                .channel(NioServerSocketChannel.class)
+                // 设置ChannelPipeline，也就是业务职责链，由处理的Handler串联而成，由从线程池处理
+                .childHandler(new ChannelInitializer<Channel>() {
+                    // 添加处理的Handler，通常包括消息编解码、业务处理，也可以是日志、权限、过滤等
+                    @Override
+                    protected void initChannel(Channel ch) throws Exception {
+                        // 获取职责链
+                        ChannelPipeline pipeline = ch.pipeline();
+                        pipeline.addLast(new IdleStateHandler(120, 0, 0, TimeUnit.SECONDS));
+                        pipeline.addLast("encode", new MessageProtocolEncoder());
+                        pipeline.addLast("decode", new MessageProtocolDecoder());
+                        pipeline.addLast("handler", new IMChannelHandler());
+                    }
+                })
+                // bootstrap 还可以设置TCP参数，根据需要可以分别设置主线程池和从线程池参数，来优化服务端性能。
+                // 其中主线程池使用option方法来设置，从线程池使用childOption方法设置。
+                // backlog表示主线程池中在套接口排队的最大数量，队列由未连接队列（三次握手未完成的）和已连接队列
+                .option(ChannelOption.SO_BACKLOG, 5)
+                // 表示连接保活，相当于心跳机制，默认为7200s
+                .childOption(ChannelOption.SO_KEEPALIVE, true);
+
+        try {
+            // 绑定端口，启动select线程，轮询监听channel事件，监听到事件之后就会交给从线程池处理
+            Channel channel = bootstrap.bind(port).sync().channel();
+            // 就绪标志
+            this.ready = true;
+            log.info("tcp server 初始化完成,端口：{}", port);
+            // 等待服务端口关闭
+            //channel.closeFuture().sync();
+        } catch (InterruptedException e) {
+            log.info("tcp server 初始化异常", e);
+        }
+    }
+
+    @Override
+    public void stop() {
+        if (bossGroup != null && !bossGroup.isShuttingDown() && !bossGroup.isShutdown()) {
+            bossGroup.shutdownGracefully();
+        }
+        if (workGroup != null && !workGroup.isShuttingDown() && !workGroup.isShutdown()) {
+            workGroup.shutdownGracefully();
+        }
+        this.ready = false;
+        log.info("tcp server 停止");
+    }
+
+
+}

im-server/src/main/java/com/bx/imserver/netty/tcp/endecode/MessageProtocolDecoder.java

@@ -0,0 +1,30 @@
+package com.bx.imserver.netty.tcp.endecode;
+
+import com.bx.imcommon.model.IMSendInfo;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ReplayingDecoder;
+import io.netty.util.CharsetUtil;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.List;
+
+@Slf4j
+public class MessageProtocolDecoder extends ReplayingDecoder {
+
+    @Override
+    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
+        if (byteBuf.readableBytes() < 4) {
+            return;
+        }
+        // 获取到包的长度
+        long length = byteBuf.readLong();
+        // 转成IMSendInfo
+        ByteBuf contentBuf = byteBuf.readBytes((int) length);
+        String content = contentBuf.toString(CharsetUtil.UTF_8);
+        ObjectMapper objectMapper = new ObjectMapper();
+        IMSendInfo sendInfo = objectMapper.readValue(content, IMSendInfo.class);
+        list.add(sendInfo);
+    }
+}

im-server/src/main/java/com/bx/imserver/netty/tcp/endecode/MessageProtocolEncoder.java

@@ -0,0 +1,24 @@
+package com.bx.imserver.netty.tcp.endecode;
+
+import com.bx.imcommon.model.IMSendInfo;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToByteEncoder;
+
+import java.nio.charset.StandardCharsets;
+
+public class MessageProtocolEncoder extends MessageToByteEncoder<IMSendInfo> {
+
+    @Override
+    protected void encode(ChannelHandlerContext channelHandlerContext, IMSendInfo sendInfo, ByteBuf byteBuf) throws Exception {
+        ObjectMapper objectMapper = new ObjectMapper();
+        String content = objectMapper.writeValueAsString(sendInfo);
+        byte[] bytes = content.getBytes(StandardCharsets.UTF_8);
+        // 写入长度
+        byteBuf.writeLong(bytes.length);
+        // 写入命令体
+        byteBuf.writeBytes(bytes);
+    }
+
+}

im-server/src/main/java/com/bx/imserver/netty/ws/WebSocketServer.java

@@ -0,0 +1,107 @@
+package com.bx.imserver.netty.ws;
+
+import com.bx.imserver.netty.IMChannelHandler;
+import com.bx.imserver.netty.IMServer;
+import com.bx.imserver.netty.ws.endecode.MessageProtocolDecoder;
+import com.bx.imserver.netty.ws.endecode.MessageProtocolEncoder;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.*;
+import io.netty.channel.nio.NioEventLoopGroup;
+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 io.netty.handler.timeout.IdleStateHandler;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * WS服务器,用于连接网页的客户端,协议格式: 直接IMSendInfo的JSON序列化
+ *
+ * @author Blue
+ * @date 2022-11-20
+ */
+@Slf4j
+@Component
+@ConditionalOnProperty(prefix = "websocket", value = "enable", havingValue = "true", matchIfMissing = false)
+public class WebSocketServer implements IMServer {
+
+    @Value("${websocket.port}")
+    private int port;
+
+    private volatile boolean ready = false;
+
+    private EventLoopGroup bossGroup;
+    private EventLoopGroup workGroup;
+
+
+    @Override
+    public boolean isReady() {
+        return ready;
+    }
+
+    @Override
+    public void start() {
+        ServerBootstrap bootstrap = new ServerBootstrap();
+        bossGroup = new NioEventLoopGroup();
+        workGroup = new NioEventLoopGroup();
+        // 设置为主从线程模型
+        bootstrap.group(bossGroup, workGroup)
+                // 设置服务端NIO通信类型
+                .channel(NioServerSocketChannel.class)
+                // 设置ChannelPipeline，也就是业务职责链，由处理的Handler串联而成，由从线程池处理
+                .childHandler(new ChannelInitializer<Channel>() {
+                    // 添加处理的Handler，通常包括消息编解码、业务处理，也可以是日志、权限、过滤等
+                    @Override
+                    protected void initChannel(Channel ch) {
+                        // 获取职责链
+                        ChannelPipeline pipeline = ch.pipeline();
+                        pipeline.addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS));
+                        pipeline.addLast("http-codec", new HttpServerCodec());
+                        pipeline.addLast("aggregator", new HttpObjectAggregator(65535));
+                        pipeline.addLast("http-chunked", new ChunkedWriteHandler());
+                        pipeline.addLast(new WebSocketServerProtocolHandler("/im"));
+                        pipeline.addLast("encode", new MessageProtocolEncoder());
+                        pipeline.addLast("decode", new MessageProtocolDecoder());
+                        pipeline.addLast("handler", new IMChannelHandler());
+                    }
+                })
+                // bootstrap 还可以设置TCP参数，根据需要可以分别设置主线程池和从线程池参数，来优化服务端性能。
+                // 其中主线程池使用option方法来设置，从线程池使用childOption方法设置。
+                // backlog表示主线程池中在套接口排队的最大数量，队列由未连接队列（三次握手未完成的）和已连接队列
+                .option(ChannelOption.SO_BACKLOG, 5)
+                // 表示连接保活，相当于心跳机制，默认为7200s
+                .childOption(ChannelOption.SO_KEEPALIVE, true);
+
+        try {
+            // 绑定端口，启动select线程，轮询监听channel事件，监听到事件之后就会交给从线程池处理
+            bootstrap.bind(port).sync().channel();
+            // 就绪标志
+            this.ready = true;
+            log.info("websocket server 初始化完成,端口：{}", port);
+            // 等待服务端口关闭
+            //channel.closeFuture().sync();
+        } catch (InterruptedException e) {
+            log.info("websocket server 初始化异常", e);
+        }
+    }
+
+    @Override
+    public void stop() {
+        if (bossGroup != null && !bossGroup.isShuttingDown() && !bossGroup.isShutdown()) {
+            bossGroup.shutdownGracefully();
+        }
+        if (workGroup != null && !workGroup.isShuttingDown() && !workGroup.isShutdown()) {
+            workGroup.shutdownGracefully();
+        }
+        this.ready = false;
+        log.info("websocket server 停止");
+    }
+
+
+}

im-server/src/main/java/com/bx/imserver/netty/ws/endecode/MessageProtocolDecoder.java

@@ -0,0 +1,19 @@
+package com.bx.imserver.netty.ws.endecode;
+
+import com.bx.imcommon.model.IMSendInfo;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToMessageDecoder;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+
+import java.util.List;
+
+public class MessageProtocolDecoder extends MessageToMessageDecoder<TextWebSocketFrame> {
+
+    @Override
+    protected void decode(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame, List<Object> list) throws Exception {
+        ObjectMapper objectMapper = new ObjectMapper();
+        IMSendInfo sendInfo = objectMapper.readValue(textWebSocketFrame.text(), IMSendInfo.class);
+        list.add(sendInfo);
+    }
+}

im-server/src/main/java/com/bx/imserver/netty/ws/endecode/MessageProtocolEncoder.java

@@ -0,0 +1,20 @@
+package com.bx.imserver.netty.ws.endecode;
+
+import com.bx.imcommon.model.IMSendInfo;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToMessageEncoder;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+
+import java.util.List;
+
+public class MessageProtocolEncoder extends MessageToMessageEncoder<IMSendInfo> {
+
+    @Override
+    protected void encode(ChannelHandlerContext channelHandlerContext, IMSendInfo sendInfo, List<Object> list) throws Exception {
+        ObjectMapper objectMapper = new ObjectMapper();
+        String text = objectMapper.writeValueAsString(sendInfo);
+        TextWebSocketFrame frame = new TextWebSocketFrame(text);
+        list.add(frame);
+    }
+}

im-server/src/main/java/com/bx/imserver/task/AbstractPullMessageTask.java

@@ -0,0 +1,23 @@
+package com.bx.imserver.task;
+
+import com.bx.imcommon.mq.RedisMQConsumer;
+import com.bx.imserver.netty.IMServerGroup;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+
+@Slf4j
+public abstract class AbstractPullMessageTask<T> extends RedisMQConsumer<T> {
+
+    @Autowired
+    private IMServerGroup serverGroup;
+
+    @Override
+    public String generateKey() {
+        return String.join(":",  super.generateKey(), IMServerGroup.serverId + "");
+    }
+
+    @Override
+    public Boolean isReady() {
+        return serverGroup.isReady();
+    }
+}

im-server/src/main/java/com/bx/imserver/task/PullGroupMessageTask.java

@@ -0,0 +1,25 @@
+package com.bx.imserver.task;
+
+import com.bx.imcommon.contant.IMRedisKey;
+import com.bx.imcommon.enums.IMCmdType;
+import com.bx.imcommon.model.IMRecvInfo;
+import com.bx.imcommon.mq.RedisMQListener;
+import com.bx.imserver.netty.processor.AbstractMessageProcessor;
+import com.bx.imserver.netty.processor.ProcessorFactory;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+@RedisMQListener(queue = IMRedisKey.IM_MESSAGE_GROUP_QUEUE, batchSize = 10)
+public class PullGroupMessageTask extends AbstractPullMessageTask<IMRecvInfo> {
+
+    @Override
+    public void onMessage(IMRecvInfo recvInfo) {
+        AbstractMessageProcessor processor = ProcessorFactory.createProcessor(IMCmdType.GROUP_MESSAGE);
+        processor.process(recvInfo);
+    }
+
+}

im-server/src/main/java/com/bx/imserver/task/PullPrivateMessageTask.java

@@ -0,0 +1,25 @@
+package com.bx.imserver.task;
+
+import com.bx.imcommon.contant.IMRedisKey;
+import com.bx.imcommon.enums.IMCmdType;
+import com.bx.imcommon.model.IMRecvInfo;
+import com.bx.imcommon.mq.RedisMQListener;
+import com.bx.imserver.netty.processor.AbstractMessageProcessor;
+import com.bx.imserver.netty.processor.ProcessorFactory;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+@RedisMQListener(queue = IMRedisKey.IM_MESSAGE_PRIVATE_QUEUE, batchSize = 10)
+public class PullPrivateMessageTask extends AbstractPullMessageTask<IMRecvInfo> {
+
+    @Override
+    public void onMessage(IMRecvInfo recvInfo) {
+        AbstractMessageProcessor processor = ProcessorFactory.createProcessor(IMCmdType.PRIVATE_MESSAGE);
+        processor.process(recvInfo);
+    }
+
+}

im-server/src/main/java/com/bx/imserver/task/PullSystemMessageTask.java

@@ -0,0 +1,28 @@
+package com.bx.imserver.task;
+
+import com.bx.imcommon.contant.IMRedisKey;
+import com.bx.imcommon.enums.IMCmdType;
+import com.bx.imcommon.model.IMRecvInfo;
+import com.bx.imcommon.mq.RedisMQListener;
+import com.bx.imserver.netty.processor.AbstractMessageProcessor;
+import com.bx.imserver.netty.processor.ProcessorFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author: Blue
+ * @date: 2024-07-16
+ * @version: 1.0
+ */
+@Slf4j
+@Component
+@RedisMQListener(queue = IMRedisKey.IM_MESSAGE_SYSTEM_QUEUE,batchSize = 10)
+public class PullSystemMessageTask extends AbstractPullMessageTask<IMRecvInfo> {
+
+    @Override
+    public void onMessage(IMRecvInfo recvInfo) {
+        AbstractMessageProcessor processor = ProcessorFactory.createProcessor(IMCmdType.SYSTEM_MESSAGE);
+        processor.process(recvInfo);
+    }
+
+}

im-server/src/main/java/com/bx/imserver/util/SpringContextHolder.java

@@ -0,0 +1,39 @@
+package com.bx.imserver.util;
+
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.stereotype.Component;
+
+@Component
+public class SpringContextHolder implements ApplicationContextAware {
+
+    private static ApplicationContext applicationContext;
+
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+        SpringContextHolder.applicationContext = applicationContext;
+    }
+
+    public static ApplicationContext getApplicationContext() {
+        assertApplicationContext();
+        return applicationContext;
+    }
+
+    public static <T> T getBean(String beanName) {
+        assertApplicationContext();
+        return (T) applicationContext.getBean(beanName);
+    }
+
+    public static <T> T getBean(Class<T> requiredType) {
+        assertApplicationContext();
+        return applicationContext.getBean(requiredType);
+    }
+
+    private static void assertApplicationContext() {
+        if (SpringContextHolder.applicationContext == null) {
+            throw new RuntimeException("applicaitonContext属性为null,请检查是否注入了SpringContextHolder!");
+        }
+    }
+
+}