feat(core): 添加注册功能

xuxueli · xuxueli · commit c798baa73e1e · 2025-04-13T01:58:30.000+08:00
- 在数据库中增加注册数据字段
diff --git a/doc/db/tables_xxl_mq.sql b/doc/db/tables_xxl_mq.sql
@@ -71,6 +71,7 @@ CREATE TABLE `xxl_mq_instance` (
     `appname`            varchar(50)     NOT NULL COMMENT 'AppName（服务唯一标识）',
     `uuid`               varchar(50)     NOT NULL COMMENT '实例唯一标识',
     `register_heartbeat` datetime        DEFAULT NULL COMMENT '实例最后心跳时间，动态注册时判定是否过期',
+    `registry_data`      text            DEFAULT NULL COMMENT '注册数据，JSON',
     `add_time`           datetime        NOT NULL COMMENT '新增时间',
     `update_time`        datetime        NOT NULL COMMENT '更新时间',
     PRIMARY KEY (`id`),
@@ -85,7 +86,7 @@ CREATE TABLE `xxl_mq_application` (
     `appname`       varchar(50)     NOT NULL COMMENT 'AppName（服务唯一标识）',
     `name`          varchar(20)     NOT NULL COMMENT '服务名称',
     `desc`          varchar(100)    NOT NULL COMMENT '服务描述',
-    `registry_data`  text COMMENT '在线节点列表，数据JSON',
+    `registry_data` text            DEFAULT NULL COMMENT '注册数据，JSON',
     `add_time`      datetime        NOT NULL COMMENT '新增时间',
     `update_time`   datetime        NOT NULL COMMENT '更新时间',
     PRIMARY KEY (`id`),
diff --git a/xxl-mq-admin2/src/main/java/com/xxl/mq/admin/model/entity/Application.java b/xxl-mq-admin2/src/main/java/com/xxl/mq/admin/model/entity/Application.java
@@ -31,6 +31,11 @@ public class Application implements Serializable {
     */
     private String desc;
 
+    /**
+     * 注册数据
+     */
+    private String registryData;
+
     /**
     * 新增时间
     */
@@ -74,6 +79,14 @@ public void setDesc(String desc) {
         this.desc = desc;
     }
 
+    public String getRegistryData() {
+        return registryData;
+    }
+
+    public void setRegistryData(String registryData) {
+        this.registryData = registryData;
+    }
+
     public Date getAddTime() {
         return addTime;
     }
diff --git a/xxl-mq-admin2/src/main/java/com/xxl/mq/admin/model/entity/Instance.java b/xxl-mq-admin2/src/main/java/com/xxl/mq/admin/model/entity/Instance.java
@@ -31,6 +31,11 @@ public class Instance implements Serializable {
     */
     private Date registerHeartbeat;
 
+    /**
+     * 注册数据
+     */
+    private String registryData;
+
     /**
     * 新增时间
     */
@@ -74,6 +79,14 @@ public void setRegisterHeartbeat(Date registerHeartbeat) {
         this.registerHeartbeat = registerHeartbeat;
     }
 
+    public String getRegistryData() {
+        return registryData;
+    }
+
+    public void setRegistryData(String registryData) {
+        this.registryData = registryData;
+    }
+
     public Date getAddTime() {
         return addTime;
     }
diff --git a/xxl-mq-admin2/src/main/java/com/xxl/mq/admin/openapi/biz/BrokerServiceImpl.java b/xxl-mq-admin2/src/main/java/com/xxl/mq/admin/openapi/biz/BrokerServiceImpl.java
@@ -1,5 +1,6 @@
 package com.xxl.mq.admin.openapi.biz;
 
+import com.xxl.mq.admin.openapi.config.BrokerFactory;
 import com.xxl.mq.core.openapi.BrokerService;
 import com.xxl.mq.core.openapi.model.*;
 import com.xxl.tool.core.CollectionTool;
@@ -20,11 +21,19 @@ public class BrokerServiceImpl implements BrokerService {
 
     @Override
     public Response<String> registry(RegistryRequest registryRequest) {
-        return null;
+        // valid token
+        if (registryRequest==null || !BrokerFactory.getInstance().validAccessToken(registryRequest.getAccessToken())) {
+            return Response.ofFail("accessToken invalid");
+        }
+
+        // invoke
+        boolean ret = BrokerFactory.getInstance().registry(registryRequest);
+        return ret? Response.ofSuccess() : Response.ofFail();
     }
 
     @Override
     public Response<String> produce(ProduceRequest produceRequest) {
+        // todo
         return null;
     }
 
diff --git a/xxl-mq-admin2/src/main/java/com/xxl/mq/admin/openapi/config/BrokerFactory.java b/xxl-mq-admin2/src/main/java/com/xxl/mq/admin/openapi/config/BrokerFactory.java
@@ -3,8 +3,11 @@
 import com.xxl.mq.admin.constant.enums.AccessTokenStatuEnum;
 import com.xxl.mq.admin.mapper.*;
 import com.xxl.mq.admin.model.entity.AccessToken;
+import com.xxl.mq.admin.model.entity.Instance;
 import com.xxl.mq.core.openapi.BrokerService;
+import com.xxl.mq.core.openapi.model.RegistryRequest;
 import com.xxl.tool.concurrent.CyclicThread;
+import com.xxl.tool.concurrent.MessageQueue;
 import com.xxl.tool.core.CollectionTool;
 import com.xxl.tool.gson.GsonTool;
 import com.xxl.tool.jsonrpc.JsonRpcServer;
@@ -16,6 +19,7 @@
 import org.springframework.context.annotation.Configuration;
 
 import javax.annotation.Resource;
+import java.util.Date;
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
@@ -47,13 +51,9 @@ public static BrokerFactory getInstance() {
     private ApplicationMapper applicationMapper;
     @Resource
     private AccessTokenMapper accessTokenMapper;
+    @Resource
+    private InstanceMapper instanceMapper;
 
-    public ApplicationMapper getApplicationMapper() {
-        return applicationMapper;
-    }
-    public AccessTokenMapper getAccessTokenMapper() {
-        return accessTokenMapper;
-    }
 
     @Override
     public void afterPropertiesSet() throws Exception {
@@ -62,12 +62,14 @@ public void afterPropertiesSet() throws Exception {
 
         // 1、AccessTokenThread
         startAccessTokenThread();
+
+        // 2、Registry MessageQueue
+        startRegistryMessageQueue();
     }
 
     @Override
     public void destroy() throws Exception {
-        // 1、AccessTokenThread
-        stopAccessTokenThread();
+        // stop
     }
 
 
@@ -84,14 +86,16 @@ public JsonRpcServer jsonRpcServer() {
 
     // ---------------------- AccessToken ----------------------
 
+    /**
+     * AccessToken LocalStore
+     */
     private volatile Set<String> accessTokenStore = new ConcurrentSkipListSet<>();
-    private CyclicThread accessTokenThread;
 
     /**
-     * start AccessTokenThread
+     * start AccessTokenThread (will stop with jvm)
      */
     private void startAccessTokenThread() {
-        accessTokenThread = new CyclicThread("accessTokenThread", true, 30 * 1000, new Runnable() {
+        CyclicThread accessTokenThread = new CyclicThread("accessTokenThread", true, 30 * 1000, new Runnable() {
             @Override
             public void run() {
                 try {
@@ -120,15 +124,6 @@ public void run() {
         accessTokenThread.start();
     }
 
-    /*
-     * stop AccessTokenThread
-     */
-    private void stopAccessTokenThread() {
-        if (accessTokenThread != null) {
-            accessTokenThread.stop();
-        }
-    }
-
     /**
      * valid  AccessToken
      *
@@ -139,4 +134,44 @@ public boolean validAccessToken(String accessToken) {
         return accessToken!=null && accessTokenStore.contains(accessToken);
     }
 
+    // ---------------------- Registry ----------------------
+
+    private volatile MessageQueue<RegistryRequest> registryMessageQueue;
+
+    /**
+     * start Registry MessageQueue (will stop with jvm)
+     */
+    private void startRegistryMessageQueue() {
+        registryMessageQueue = new MessageQueue<RegistryRequest>(
+                "registryMessageQueue",
+                1000,
+                messages -> {
+                    for (RegistryRequest registryRequest : messages) {
+                        // write registry
+                        String registryDate = GsonTool.toJson(registryRequest);
+
+                        Instance newInstance = new Instance();
+                        newInstance.setAppname(registryRequest.getAppname());
+                        newInstance.setUuid(registryRequest.getInstanceUuid());
+                        newInstance.setRegisterHeartbeat(new Date());
+                        newInstance.setRegistryData(registryDate);
+
+                        instanceMapper.insert(newInstance); // todo, save or update
+                    }
+                },
+                1,
+                1);
+    }
+
+    /**
+     * registry
+     *
+     * @param registryRequest
+     * @return
+     */
+    public boolean registry(RegistryRequest registryRequest) {
+        return registryMessageQueue.produce(registryRequest);
+    }
+
+
 }
diff --git a/xxl-mq-admin2/src/main/resources/mapper/ApplicationMapper.xml b/xxl-mq-admin2/src/main/resources/mapper/ApplicationMapper.xml
@@ -8,6 +8,7 @@
         <result column="appname" property="appname" />
         <result column="name" property="name" />
         <result column="desc" property="desc" />
+        <result column="registry_data" property="registryData" />
         <result column="add_time" property="addTime" />
         <result column="update_time" property="updateTime" />
     </resultMap>
@@ -17,6 +18,7 @@
         t.`appname`,
         t.`name`,
         t.`desc`,
+        t.`registry_data`,
         t.`add_time`,
         t.`update_time`
     </sql>
@@ -26,13 +28,15 @@
             `appname`,
             `name`,
             `desc`,
+            `registry_data`,
             `add_time`,
             `update_time`
         )
         VALUES(
                   #{application.appname} ,
                   #{application.name} ,
                   #{application.desc} ,
+                  #{application.registryData} ,
                   NOW() ,
                   NOW()
               )
@@ -51,6 +55,7 @@
         SET
             `name` = #{application.name},
             `desc` = #{application.desc},
+            `registry_data` = #{application.registryData},
             `update_time` = NOW()
         WHERE `id` = #{application.id}
     </update>
diff --git a/xxl-mq-admin2/src/main/resources/mapper/InstanceMapper.xml b/xxl-mq-admin2/src/main/resources/mapper/InstanceMapper.xml
@@ -8,6 +8,7 @@
         <result column="appname" property="appname" />
         <result column="uuid" property="uuid" />
         <result column="register_heartbeat" property="registerHeartbeat" />
+        <result column="registry_data" property="registryData" />
         <result column="add_time" property="addTime" />
         <result column="update_time" property="updateTime" />
     </resultMap>
@@ -17,6 +18,7 @@
         t.`appname`,
         t.`uuid`,
         t.`register_heartbeat`,
+        t.`registry_data`,
         t.`add_time`,
         t.`update_time`
     </sql>
@@ -26,13 +28,15 @@
             `appname`,
             `uuid`,
             `register_heartbeat`,
+            `registry_data`,
             `add_time`,
             `update_time`
         )
         VALUES(
               #{instance.appname} ,
               #{instance.uuid} ,
               #{instance.registerHeartbeat} ,
+              #{instance.registryData} ,
               NOW() ,
               NOW()
           )
@@ -52,6 +56,7 @@
             `appname` = #{instance.appname},
             `uuid` = #{instance.uuid},
             `register_heartbeat` = #{instance.registerHeartbeat},
+            `registry_data` = #{instance.registryData},
             `update_time` = NOW()
         WHERE `id` = #{instance.id}
     </update>
diff --git a/xxl-mq-core/src/main/java/com/xxl/mq/core/openapi/model/RegistryRequest.java b/xxl-mq-core/src/main/java/com/xxl/mq/core/openapi/model/RegistryRequest.java
diff --git a/xxl-mq-core/src/test/java/com/xxl/mq/core/test/openapi/BrokerServiceTest.java b/xxl-mq-core/src/test/java/com/xxl/mq/core/test/openapi/BrokerServiceTest.java
