Update RM

Author: xuxueli

doc/XXL-MQ官方文档.md

@@ -633,51 +633,71 @@ transaction | 事务开关，开启消息事务性保证只会成功执行一次
 ### 5.7 版本 v1.4.0 Release Notes[迭代中]
 
 ### Tmp
-
+```
 1、特性：
     - 特性：存算分离、水平扩展、高性能（TPS：Mysql单机1W/Blade 10W）、海量消息（Mysql日百万/Blade日十亿）、消息轨迹、多消费模式（分片/串行/广播）、延迟消息、失败重试（固定/增长/指数）、
     - 其他：失败告警、AccessToken、容器化；
-2、流程：
-    - pub->broker：根据group生产，服务端处理。
-    - sub-> broker：根据cId分片消费，服务端分片；服务端滚动预热，抗并发。
-    - broker ：无状态，维护topic注册节点。
-    - appname：应用服务，注册节点uuid
-    - topic：关联appname，分片切分。清理及归档配置。
-    - message：topic，group，shardId
-    - 归档数据：定时归档，自动清理
-3、设计：
+2、设计：
 - Broker：
     - Manage：
-      - User：服务授权
+        - User：服务授权
         - AccessToken：能力
-        - AppName：能力（注册、节点动态更新；用于Topic数据分片；）；【AppName维度，在线实例信息：实例总数 = instanceNum；序号 = instanceIndex；partitionScope = 3～5；】
+        - AppName：
+            - 能力（注册、节点动态更新；用于Topic数据分片；）；【AppName维度，在线实例信息：实例总数 = instanceNum；序号 = instanceIndex；partitionScope = 3～5；】
             - 模型：Instance注册：字段（appname + uuid + register_heartbeat）；app维度20s汇总一次，同步至app表；时钟打平，从0开始每20s一次；
-        - Topic：能力（定义管理 + 查看注册节点 / 节点分片分配情况；）；【Topic】【名称】【负责人】【告警邮箱】【状态】【存储策略】【partition数量】【优先级】【重试次数】【重试间隔策略】【归档策略】
-            - 模型：topic + store + partition + level + author + alarm_email + timeout
-        - Message：查看 + 管理（增 + 该状态 + 归档）；消息队列，物理消息队列；【10min一次，自动数据归档；】
+        - Topic：
+            - 能力（定义管理 + 查看注册节点 / 节点分片分配情况；）；【Topic】【名称】【负责人】【告警邮箱】【状态】【存储策略】【partition数量】【优先级】【重试次数】【重试间隔策略】【归档策略】
+            - 模型：topic + store + partitionCount + level + author + alarm_email + timeout
+        - Message：
+            - 能力：查看 + 管理（增 + 该状态 + 归档）；消息队列，物理消息队列；【10min一次，自动数据归档；】
             - 模型：msgid + msgbody + topic + group + partitionId + status + retryCount + intervalTime + effectTime + consume_log;
             - 属性：
                 - topic：关联 消息主题；
                 - group：数据广播；【topic向上；广播；】
                 - uuid序号：并行处理
                 - partitionKey：分区Key进行hashcode取模，会转成分区ID，限制 [0-10000] 之内；结合Consumer在线列表，匹配消费分片范围，实现并行分片消费消息；【topic向下；并行；同sId保障顺序；根据消费者 partition 信息计算 范围；】
-         - MessageArchive：归档消息，同 message；
-    - Registry：提供 Consumer 注册、动态发现能力；消息分片消费时使用；
-    - Broker Server：提供消息存储、读写能力；
+         - MessageArchive：
+            - 能力：定期讲终止态消息，同步归档，清理原始表；
+            - 模型：同 message；
+    - Registry：
+         - 能力：提供 Consumer 注册、动态发现能力；消息分片消费时使用；
+    - Broker 
+         - 能力：Server：提供消息存储、读写能力；
     - OpenAPI：统一“Token验证”（http+gson；借助 xxl-tool 实现通用 http-rpc 能力；）
         - a 、注册：app+topic初始化 + 节点心跳注册/摘除；
-          - 数据格式：
-            - instance01（UUID）：
-              - topic01 / group01
-              - topic02 / group02
-        - b、生产：异步队列，批量写入；处理group广播。
-        - c、批量查询（锁定）：pullAndLock，多topic并行查询能力，根据node分配10条，同时锁定。分配不到直接返回。
-          - 分片数据：
-            - topic01：
-              - group01：2/3；
-          - 逻辑：pullAndLock（topics，group +节点）【生成唯一标识，lock时写入；根据标识判断锁定值；】
-            - 分片查询：topic + group + 分片范围（分片序号计算；动态计算）；默认每个topic取100条数据；
-            - 数据所动：查询出的数据，锁定执行状态；根基Topic自定义超时时间（默认10min）超时释放；
+                  - 数据格式：
+                        - app01 : 
+                        - instanceUuid01：
+                        - topicList：
+                              - topic01 : default
+                              - topic02 : group02
+                  - RegistryData：存app表；
+                        - app01:
+                              - instanceList：
+                                    - instance01：partitionScope: [0-5000]
+                                    - instance02：partitionScope：[5001-10000]
+                              - topicList：
+                                    - topic01 : [default、group01、group02]
+                                    - topic02 : [default]
+                  - RegistryCache：30s一次；
+                        - topic01：
+                          - info：配置信息（app）；
+                          - instanceList：
+                                    - instance01：partitionScope: [0-5000]
+                                    - instance02：partitionScope：[5001-10000]
+                          - groupList：[default、group01、group02]
+                        - app01：基础信息；
+        - b、生产：
+                  - 能力：异步队列，批量写入；处理group广播。
+                  - 数据格式：topic + group(null广播)  + partitionKey（null随机；partitionId [0, 10000]） + msgBody
+        - c、批量查询（锁定）：
+                - 能力：多topic并行查询；单topic分片查询；能力，根据node分配10条，同时锁定。分配不到直接返回。
+                - pullAndLock：（topics，group +节点）【生成唯一标识，lock时写入；根据标识判断锁定值；】
+                      - 数据格式：
+                          - topic01 + group01 + clientId（计算partitionScope）
+                          - topic02 + group01 +
+                      - 查询逻辑：多topic并行查询 + 单topic分片查询 + 每topic每次取10个；
+                      - 锁定逻辑：针对查询出的数据，更新锁定态；注意超时释放；
         - d、消费消息：异步队列，批量更新消费结果；
 - Client：
     - Registry 组件：
@@ -687,9 +707,9 @@ transaction | 事务开关，开启消息事务性保证只会成功执行一次
         - 功能：直连Broker；发起消息生产；
         - 性能：内存queue，批量异步推送；异常，写本地磁盘；（xxl-tool）
         - 要点：
-            - 并行消息：指定 topic + group（固定） + shardingid（随机），生产单条消息；借助 shardingid 分片消费；
-            - 串行消息：指定 topic + group（固定） + shardingid（固定>0），生产单条消息；固定 shardingid 绑定固定节点消费；
-            - 广播消息：指定 topic + group（uuid） + shardingid（随机）；根据在线 group 列表生产多条消息；
+            - 并行消息：指定 topic + group（固定） + partitionKey（随机），生产单条消息；借助 partitionKey 分片消费；
+            - 串行消息：指定 topic + group（固定） + partitionKey（固定>0），生产单条消息；固定 partitionKey 绑定固定节点消费；
+            - 广播消息：指定 topic + group（uuid） + partitionKey（随机）；根据在线 group 列表生产多条消息；
     - Consumer 组件：
       - 功能：查询消息，消费消息，回调消息；
       - 性能：批量查询、批量回调；
@@ -700,7 +720,7 @@ transaction | 事务开关，开启消息事务性保证只会成功执行一次
     - ConsumerInvokeThread 组件：
       - 功能：消费业务逻辑，执行线程；
       - 要点：单Topic单线程；不同Topic隔离；
-
+```
 
 ### TODO
 - 会考虑移除 mysql 强依赖的，迁移 jpa 进一步提升通用型。

doc/db/tables_xxl_mq.sql

@@ -8,65 +8,67 @@ use `xxl_mq`;
 SET NAMES utf8mb4;
 
 
-## —————————————————————— config data ——————————————————
+## —————————————————————— topic and message ——————————————————
 
 CREATE TABLE `xxl_mq_topic`(
-    `id`          int(11)      NOT NULL AUTO_INCREMENT,
-    `topic`       varchar(255) NOT NULL COMMENT '消息主题Topic',
-    `desc`        varchar(100) NOT NULL COMMENT '消息主题名称',
-    `store_type`  tinyint(4)   NOT NULL COMMENT '存储类型：0-通用存储，2-单独存储',
-    `level`       int(11)      NOT NULL COMMENT '优先级',
-    `owner`       varchar(50)  NOT NULL COMMENT '负责人',
-    `alarm_email` varchar(255) DEFAULT NULL COMMENT '告警配置（邮箱）',
-    `add_time`    datetime     NOT NULL COMMENT '新增时间',
-    `update_time` datetime     NOT NULL COMMENT '更新时间',
+    `id`                    bigint(20)   NOT NULL AUTO_INCREMENT,
+    `topic`                 varchar(255) NOT NULL COMMENT '消息主题Topic',
+    `name`                  varchar(100) NOT NULL COMMENT '消息主题名称',
+    `owner`                 varchar(50)  NOT NULL COMMENT '负责人',
+    `alarm_email`           varchar(255) DEFAULT NULL COMMENT '告警配置（邮箱）',
+    `status`                tinyint(4)   NOT NULL COMMENT '状态：0-正常、1-禁用',
+    `store_strategy`        tinyint(4)   NOT NULL COMMENT '存储策略：0-统一存储，2-隔离存储',
+    `partition_strategy`    tinyint(4)   NOT NULL COMMENT '分区策略：0-Hash分区，1-随机分区，2-轮询分区',
+    `level`                 int(11)      NOT NULL COMMENT '优先级',
+    `retry_type`            varchar(100) NOT NULL COMMENT '重试策略（固定；增长；指数；不重试；）',
+    `retry_count`           int(11)      NOT NULL COMMENT '重试次数',
+    `retry_interval`        int(11)      NOT NULL COMMENT '重试间隔，单位秒（3s；3/6/9；3/9/27）',
+    `execution_timeout`     int(11)      NOT NULL COMMENT '执行超时时间',
+    `add_time`              datetime     NOT NULL COMMENT '新增时间',
+    `update_time`           datetime     NOT NULL COMMENT '更新时间',
     PRIMARY KEY (`id`),
     UNIQUE KEY `uni_topic` (`topic`) USING BTREE
 ) ENGINE = InnoDB  DEFAULT CHARSET = utf8mb4 COMMENT ='消息主题';
 
 CREATE TABLE `xxl_mq_message`(
-    `id`             bigint(20)   NOT NULL AUTO_INCREMENT,
-    `data`           text         NOT NULL COMMENT '消息数据',
-    `topic`          varchar(255) NOT NULL COMMENT '消息主题Topic',
-    `group`          varchar(255) NOT NULL COMMENT '消息分组',
-    `sharding_id`    int(11)      NOT NULL COMMENT '消息分片序号',
-    `status`         tinyint(4)   NOT NULL COMMENT '状态：0-正常、1-运行中、2-成功、3-失败',
-    `retry_count`    int(11)      NOT NULL COMMENT '重试次数',
-    `retry_type`     varchar(100) NOT NULL COMMENT '重试策略（固定；增长；指数）',
-    `retry_interval` int(11)      NOT NULL COMMENT '重试间隔（3s；2/4/6；2/4/8）',
-    `effect_time`    datetime     NOT NULL COMMENT '生效时间',
-    `consume_log`    text DEFAULT NULL COMMENT '消费地址',
-    `add_time`       datetime     NOT NULL COMMENT '新增时间',
-    `update_time`    datetime     NOT NULL COMMENT '更新时间',
+    `id`                    bigint(20)   NOT NULL AUTO_INCREMENT,
+    `data`                  text         NOT NULL COMMENT '消息数据',
+    `topic`                 varchar(255) NOT NULL COMMENT '消息主题Topic',
+    `group`                 varchar(255) NOT NULL COMMENT '消息主题分组',
+    `partition_id`          int(11)      NOT NULL COMMENT '消息分片ID',
+    `status`                tinyint(4)   NOT NULL COMMENT '状态：0-正常、1-执行中、2-成功、3-失败、4-超时失败',
+    `effect_time`           datetime     NOT NULL COMMENT '生效时间',
+    `consume_log`           text         DEFAULT NULL COMMENT '消费日志',
+    `consume_instance_uuid` varchar(50)  DEFAULT NULL COMMENT '消费实例实例唯一标识',
+    `add_time`              datetime     NOT NULL COMMENT '新增时间',
+    `update_time`           datetime     NOT NULL COMMENT '更新时间',
     PRIMARY KEY (`id`),
-    KEY `i_t_g_1` (`topic`, `group`, `sharding_id`)
+    KEY `i_t_g_p` (`topic`, `group`, `partition_id`)
 ) ENGINE = InnoDB  DEFAULT CHARSET = utf8mb4 COMMENT ='消息数据表';
 
 CREATE TABLE `xxl_mq_message_archive` (
-    `id`             bigint(20)   NOT NULL AUTO_INCREMENT,
-    `data`           text         NOT NULL COMMENT '消息数据',
-    `topic`          varchar(255) NOT NULL COMMENT '消息主题Topic',
-    `group`          varchar(255) NOT NULL COMMENT '消息分组',
-    `sharding_id`    int(11)      NOT NULL COMMENT '消息分片序号',
-    `status`         tinyint(4)   NOT NULL COMMENT '状态：0-正常、1-运行中、2-成功、3-失败',
-    `retry_count`    int(11)      NOT NULL COMMENT '重试次数',
-    `retry_type`     varchar(100) NOT NULL COMMENT '重试策略（固定；增长；指数）',
-    `retry_interval` int(11)      NOT NULL COMMENT '重试间隔（3s；2/4/6；2/4/8）',
-    `effect_time`    datetime     NOT NULL COMMENT '生效时间',
-    `consume_log`    text DEFAULT NULL COMMENT '消费日志',
-    `add_time`       datetime     NOT NULL COMMENT '新增时间',
-    `update_time`    datetime     NOT NULL COMMENT '更新时间',
+    `id`                    bigint(20)   NOT NULL AUTO_INCREMENT,
+    `data`                  text         NOT NULL COMMENT '消息数据',
+    `topic`                 varchar(255) NOT NULL COMMENT '消息主题Topic',
+    `group`                 varchar(255) NOT NULL COMMENT '消息主题分组',
+    `partition_id`          int(11)      NOT NULL COMMENT '消息分片ID',
+    `status`                tinyint(4)   NOT NULL COMMENT '状态：0-正常、1-执行中、2-成功、3-失败、4-超时失败',
+    `effect_time`           datetime     NOT NULL COMMENT '生效时间',
+    `consume_log`           text         DEFAULT NULL COMMENT '消费日志',
+    `consume_instance_uuid` varchar(50)  DEFAULT NULL COMMENT '消费实例实例唯一标识',
+    `add_time`              datetime     NOT NULL COMMENT '新增时间',
+    `update_time`           datetime     NOT NULL COMMENT '更新时间',
     PRIMARY KEY (`id`),
-    KEY `i_t_g_1` (`topic`, `group`, `sharding_id`)
+    KEY `i_t_g_p` (`topic`, `group`, `partition_id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='消息数据归档表';
 
 ## —————————————————————— registry ——————————————————
 
 CREATE TABLE `xxl_mq_instance` (
     `id`                 bigint(20)      NOT NULL AUTO_INCREMENT,
     `appname`            varchar(50)     NOT NULL COMMENT 'AppName（服务唯一标识）',
-    `uuid`               varchar(50)     NOT NULL COMMENT '节点唯一标识',
-    `register_heartbeat` datetime        DEFAULT NULL COMMENT '节点最后心跳时间，动态注册时判定是否过期',
+    `uuid`               varchar(50)     NOT NULL COMMENT '实例唯一标识',
+    `register_heartbeat` datetime        DEFAULT NULL COMMENT '实例最后心跳时间，动态注册时判定是否过期',
     `add_time`           datetime        NOT NULL COMMENT '新增时间',
     `update_time`        datetime        NOT NULL COMMENT '更新时间',
     PRIMARY KEY (`id`),
@@ -81,7 +83,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_instance`  text COMMENT '在线节点列表，数据JSON',
+    `registry_data`  text COMMENT '在线节点列表，数据JSON',
     `add_time`      datetime        NOT NULL COMMENT '新增时间',
     `update_time`   datetime        NOT NULL COMMENT '更新时间',
     PRIMARY KEY (`id`),

xxl-mq-admin2/src/main/java/com/xxl/mq/admin/mapper/InstanceMapper.java

@@ -0,0 +1,47 @@
+package com.xxl.mq.admin.mapper;
+
+import com.xxl.mq.admin.model.entity.Instance;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Mapper;
+
+import java.util.List;
+
+/**
+* Instance Mapper
+*
+* Created by xuxueli on '2025-03-16 12:29:49'.
+*/
+@Mapper
+public interface InstanceMapper {
+
+    /**
+    * 新增
+    */
+    public int insert(@Param("instance") Instance instance);
+
+    /**
+    * 删除
+    */
+    public int delete(@Param("ids") List<Integer> ids);
+
+    /**
+    * 更新
+    */
+    public int update(@Param("instance") Instance instance);
+
+    /**
+    * Load查询
+    */
+    public Instance load(@Param("id") int id);
+
+    /**
+    * 分页查询Data
+    */
+	public List<Instance> pageList(@Param("offset") int offset, @Param("pagesize") int pagesize);
+
+    /**
+    * 分页查询Count
+    */
+    public int pageListCount(@Param("offset") int offset, @Param("pagesize") int pagesize);
+
+}

xxl-mq-admin2/src/main/java/com/xxl/mq/admin/model/entity/Instance.java

@@ -0,0 +1,93 @@
+package com.xxl.mq.admin.model.entity;
+
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+*  Instance Entity
+*
+*  Created by xuxueli on '2025-03-16 12:28:40'.
+*/
+public class Instance implements Serializable {
+    private static final long serialVersionUID = 42L;
+
+    /**
+    * id
+    */
+    private long id;
+
+    /**
+    * AppName（服务唯一标识）
+    */
+    private String appname;
+
+    /**
+    * 节点唯一标识
+    */
+    private String uuid;
+
+    /**
+    * 节点最后心跳时间，动态注册时判定是否过期
+    */
+    private Date registerHeartbeat;
+
+    /**
+    * 新增时间
+    */
+    private Date addTime;
+
+    /**
+    * 更新时间
+    */
+    private Date updateTime;
+
+
+    public long getId() {
+        return id;
+    }
+
+    public void setId(long id) {
+        this.id = id;
+    }
+
+    public String getAppname() {
+        return appname;
+    }
+
+    public void setAppname(String appname) {
+        this.appname = appname;
+    }
+
+    public String getUuid() {
+        return uuid;
+    }
+
+    public void setUuid(String uuid) {
+        this.uuid = uuid;
+    }
+
+    public Date getRegisterHeartbeat() {
+        return registerHeartbeat;
+    }
+
+    public void setRegisterHeartbeat(Date registerHeartbeat) {
+        this.registerHeartbeat = registerHeartbeat;
+    }
+
+    public Date getAddTime() {
+        return addTime;
+    }
+
+    public void setAddTime(Date addTime) {
+        this.addTime = addTime;
+    }
+
+    public Date getUpdateTime() {
+        return updateTime;
+    }
+
+    public void setUpdateTime(Date updateTime) {
+        this.updateTime = updateTime;
+    }
+
+}