重构 IM 部分的代码设计 (#18)

* 优化代码设计

* 重构IM部分的代码

* fix compile

* fix

* fix-2

* fix-3

* 升版

---------

Co-authored-by: 姚世权 <yaoshiquan@zwcad.com>

Author: passerbyYSQ

pom.xml

@@ -11,13 +11,17 @@
     </parent>
 	<groupId>top.ysqorz</groupId>
 	<artifactId>forum</artifactId>
-	<version>1.1.0</version>
+	<version>1.1.1</version>
 	<!-- 打包方式为war包	-->
 	<packaging>jar</packaging>
 	<name>forum</name>
 	<description>旨在打造一个功能齐全的论坛。It is not a small project!</description>
 
 	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+		<maven.compiler.encoding>UTF-8</maven.compiler.encoding>
+		<argLine>-Dfile.encoding=UTF-8</argLine>
 		<java.version>1.8</java.version>
         <netty.version>4.1.50.Final</netty.version>
 		<!-- 版本过高，连接时会出现两次报错，然后才连接成功 -->

src/main/java/top/ysqorz/forum/controller/IMController.java

@@ -1,6 +1,5 @@
 package top.ysqorz.forum.controller;
 
-import org.springframework.stereotype.Component;
 import org.springframework.stereotype.Controller;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
@@ -9,9 +8,8 @@
 import top.ysqorz.forum.common.StatusCode;
 import top.ysqorz.forum.im.entity.MsgModel;
 import top.ysqorz.forum.im.entity.MsgType;
-import top.ysqorz.forum.im.handler.MsgCenter;
+import top.ysqorz.forum.im.handler.MsgCenterImpl;
 import top.ysqorz.forum.service.IMService;
-import top.ysqorz.forum.shiro.ShiroUtils;
 import top.ysqorz.forum.utils.JsonUtils;
 
 import javax.annotation.Resource;
@@ -44,7 +42,7 @@ public StatusCode sendMsg(@NotBlank String msgJson, @NotBlank String channelId)
         if (MsgType.isFunctionalType(MsgType.valueOf(msg.getMsgType()))) { // 如果非法type会抛出异常
             return StatusCode.NOT_SUPPORT_FUNC_TYPE;
         }
-        MsgCenter.getInstance().remoteDispatch(msg, channelId, ShiroUtils.getToken());
+        imService.handleMsg(msg, channelId);
         return StatusCode.SUCCESS;
     }
 
@@ -53,7 +51,7 @@ public StatusCode sendMsg(@NotBlank String msgJson, @NotBlank String channelId)
      */
     @PostMapping("/push")
     public StatusCode pushMsg(@RequestBody MsgModel msg, String channelId) { // source channel
-        MsgCenter.getInstance().push(msg, channelId);
+        MsgCenterImpl.getInstance().push(msg, channelId);
         return StatusCode.SUCCESS;
     }
 

src/main/java/top/ysqorz/forum/im/MsgCenter.java

@@ -0,0 +1,21 @@
+package top.ysqorz.forum.im;
+
+import top.ysqorz.forum.im.entity.ChannelMap;
+import top.ysqorz.forum.im.handler.BindEventCallback;
+import top.ysqorz.forum.im.handler.BindMsgHandler;
+
+/**
+ * @author passerbyYSQ
+ * @create 2025-02-01 22:24
+ */
+public interface MsgCenter extends MsgOperator, BindEventCallback {
+    /**
+     * 添加消息处理器
+     */
+    MsgCenter addHandler(MsgHandler<?> handler);
+
+    /**
+     * 获取对应类型的所有通道
+     */
+    ChannelMap getChannelMap(String channelType);
+}

src/main/java/top/ysqorz/forum/im/MsgHandler.java

@@ -0,0 +1,43 @@
+package top.ysqorz.forum.im;
+
+import io.netty.channel.Channel;
+import top.ysqorz.forum.im.entity.ChannelMap;
+import top.ysqorz.forum.im.entity.ChannelType;
+import top.ysqorz.forum.im.entity.MsgModel;
+import top.ysqorz.forum.im.entity.MsgType;
+
+import java.util.concurrent.Executor;
+
+
+/**
+ * @author passerbyYSQ
+ * @create 2025-02-01 22:32
+ */
+public interface MsgHandler<DataType> extends MsgOperator {
+    /**
+     * 当前处理器是否支持处理此消息
+     */
+    boolean support(MsgModel msg, Channel channel);
+
+    /**
+     * 消息的POJO类
+     */
+    Class<DataType> getDataClass();
+
+    /**
+     * 当前消息处理器支持的消息类型
+     */
+    MsgType getMsgType();
+
+    /**
+     * 当前消息处理器支持的通道类型
+     */
+    ChannelType getChannelType();
+
+    /**
+     * 设置异步线程池
+     */
+    void setExecutor(Executor executor);
+
+    void setChannelMap(ChannelMap channelMap);
+}

src/main/java/top/ysqorz/forum/im/MsgHandlerPipeline.java

@@ -0,0 +1,37 @@
+package top.ysqorz.forum.im;
+
+import java.util.Iterator;
+
+/**
+ * ...
+ *
+ * @author yaoshiquan
+ * @date 2025/1/27
+ */
+public interface MsgHandlerPipeline extends Iterable<MsgHandler<?>>, Iterator<MsgHandler<?>> {
+    default MsgHandlerPipeline addHandler(MsgHandler<?> handler) {
+        return addHandlerAtTail(handler);
+    }
+
+    MsgHandlerPipeline addHandlerAtHead(MsgHandler<?> handler);
+
+    MsgHandlerPipeline addHandlerAtTail(MsgHandler<?> handler);
+
+    MsgHandlerPipeline addHandlerAtIndex(int index, MsgHandler<?> handler);
+
+    MsgHandler<?> getHeadHandler();
+
+    MsgHandler<?> getTailHandler();
+
+    MsgHandler<?> getHandler(int index);
+
+    void removeHandler(MsgHandler<?> handler);
+
+    void removeHandler(int index);
+
+    boolean exist(MsgHandler<?> handler);
+
+    boolean isEmpty();
+
+    int size();
+}

src/main/java/top/ysqorz/forum/im/MsgOperator.java

@@ -0,0 +1,20 @@
+package top.ysqorz.forum.im;
+
+import io.netty.channel.Channel;
+import top.ysqorz.forum.im.entity.MsgModel;
+
+/**
+ * @author passerbyYSQ
+ * @create 2025-02-01 22:54
+ */
+public interface MsgOperator {
+    /**
+     * 处理消息
+     */
+    boolean handle(MsgModel msg, Channel channel);
+
+    /**
+     * 在当前服务节点，推送消息到客户端
+     */
+    boolean push(MsgModel msg, String channelId);
+}

src/main/java/top/ysqorz/forum/im/entity/AsyncInsertTask.java

@@ -7,9 +7,9 @@
  * @create 2022-01-16 1:03
  */
 public class AsyncInsertTask<T> implements Runnable { // T：PO的类型
-    private BaseMapper<T> mapper;
-    private T data;
-    private int retryCount;
+    private final BaseMapper<T> mapper;
+    private final T data;
+    private final int retryCount;
 
     public AsyncInsertTask(BaseMapper<T> mapper, T data) {
         this(mapper, data, 3);

src/main/java/top/ysqorz/forum/im/entity/ChannelMap.java

@@ -7,6 +7,7 @@
 
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -33,10 +34,10 @@ public void bind(String token, String groupId, Channel channel) {
         channel.attr(IMUtils.GROUP_ID_KEY).set(groupId);
         channel.attr(IMUtils.CHANNEL_TYPE_KEY).set(channelType.name());
         // 双重检测锁
-        Set<Channel> channels = channelMap.get(groupId);
-        if (channels == null) {
+        Set<Channel> channels;
+        if (Objects.isNull(channels = channelMap.get(groupId))) {
             synchronized (this) {
-                if (channels == null) {
+                if (Objects.isNull(channels = channelMap.get(groupId))) {
                     channels = new HashSet<>();
                     channelMap.put(groupId, channels);
                 }
@@ -76,6 +77,9 @@ public void pushToGroup(MsgType msgType, Object data, String sourceChannelId, St
 
     public boolean isBound(Channel channel) {
         String token = IMUtils.getTokenFromChannel(channel);
+        if (channel instanceof FakeChannel && Objects.nonNull(token)) {
+            return true;
+        }
         String channelType = IMUtils.getChannelTypeFromChannel(channel);
         String groupId = IMUtils.getGroupIdFromChannel(channel);
         if (token == null || channelType == null || groupId == null) {