# shanks-mq-component

**Repository Path**: packagejava/mq-componet

## Basic Information

- **Project Name**: shanks-mq-component
- **Description**: 基于rocketmq + 本地消息表实现的分布式事务消息组件
- **Primary Language**: Java
- **License**: Apache-2.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 4
- **Created**: 2025-05-19
- **Last Updated**: 2025-05-19

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

<div id="top"></div>

<!-- 目录 -->

<details>
  <summary>目录</summary>
  <ol>
    <li>
      <a href="#关于本项目">关于本项目</a>
      <ul>
        <li><a href="#构建工具">构建工具</a></li>
      </ul>
    </li>
    <li>
      <a href="#开始">开始</a>
      <ul>
        <li><a href="#依赖">依赖</a></li>
        <li><a href="#安装">安装</a></li>
      </ul>
    </li>
    <li><a href="#使用方法">使用方法</a></li>
    <li><a href="#路线图">路线图</a></li>
    <li><a href="#贡献">贡献</a></li>
    <li><a href="#许可证">许可证</a></li>
    <li><a href="#联系我们">联系我们</a></li>
    <li><a href="#致谢">致谢</a></li>
  </ol>
</details>



<!-- 关于本项目 -->
## 关于本项目

[![产品截图][product-screenshot]](https://example.com)

本项目是一个分布式事务最终一致性解决方案的框架，使用本地事务+本地消息表的方式实现事务的最终一致性



<p align="right">(<a href="#top">返回顶部</a>)</p>



### 技术栈

* [Maven]([https://maven.apache.org/)
* [Java JDK17以上](https://www.oracle.com/java/technologies/javase/jdk17-archive-downloads.html)
* [SpringBoot 3.0](https://docs.spring.io/spring-boot/index.html)
* [RocketMQ 5.0](https://rocketmq.apache.org/)
* [Mybatis-Plus 3.5.6](https://baomidou.com/)
* [MySQL 8.0](https://www.mysql.com/cn/)

<p align="right">(<a href="#top">返回顶部</a>)</p>



<!-- 开始 -->
## 开始

### 使用方法

本地数据库增加表

````sql
CREATE TABLE `message_shedlock` (
                                             `name` VARCHAR(255) NOT NULL COMMENT '锁的唯一标识',
                                             `lock_until` DATETIME NOT NULL COMMENT '锁的有效期时间',
                                             `locked_at` DATETIME NOT NULL COMMENT '锁被获取的时间',
                                             `locked_by` VARCHAR(255) NOT NULL COMMENT '持有锁的节点',
                                             PRIMARY KEY (`name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='分布式锁表';

CREATE TABLE `message_consume_record` (
                                   `message_consume_id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键',
                                   `msg_id` varchar(255) DEFAULT NULL COMMENT '消费时包装的消息id',
                                   `msg_topic` varchar(255) DEFAULT NULL COMMENT '消息主题',
                                   `msg_tag` varchar(255) DEFAULT NULL COMMENT '消息标签',
                                   `msg_key` varchar(255) DEFAULT NULL COMMENT '消息键',
                                   `checkcode` varchar(255) DEFAULT NULL COMMENT '校验码',
                                   `consumer_source` varchar(255) DEFAULT NULL COMMENT '消费者',
                                   `producer_source` varchar(255) DEFAULT NULL COMMENT '生产者',
                                   `mq_zone` varchar(255) DEFAULT NULL COMMENT '消息泳道',
                                   `message_json` text COMMENT '消息Json',
                                   `consume_time` datetime DEFAULT NULL COMMENT '消费时间',
                                   `create_time` datetime DEFAULT NULL COMMENT '创建时间',
                                   PRIMARY KEY (`message_consume_id`),
                                   UNIQUE KEY `unidx_msg_id` (`msg_id`),
                                   KEY `idx_create_time` (`create_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='消息消费表';

CREATE TABLE `message_consume_exception_record` (
                                                    `message_consume_exception_id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键',
                                                    `msg_id` varchar(255) DEFAULT NULL COMMENT '消息 ID',
                                                    `msg_topic` varchar(255) DEFAULT NULL COMMENT '消息主题',
                                                    `msg_tag` varchar(255) DEFAULT NULL COMMENT '消息标签',
                                                    `msg_key` varchar(255) DEFAULT NULL COMMENT '消息键',
                                                    `consumer` varchar(255) DEFAULT NULL COMMENT '消费者',
                                                    `consume_exception_time` datetime DEFAULT NULL COMMENT '消费异常时间',
                                                    `delete_time` datetime DEFAULT NULL COMMENT '删除时间',
                                                    `delete_flag` tinyint(1) DEFAULT NULL COMMENT '删除标志',
                                                    PRIMARY KEY (`message_consume_exception_id`),
                                                    KEY `idx_msg_id` (`msg_id`,`consume_exception_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='消息消费异常记录表';

CREATE TABLE `message_record` (
                                  `message_id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键',
                                  `message_uuid` varchar(255) DEFAULT NULL COMMENT '消息UUID,由messageJson + topic md5而成,数据库唯一索引防止重复推送',
                                  `message_json` text COMMENT '消息Json',
                                  `message_topic` varchar(255) DEFAULT NULL COMMENT '消息topic',
                                  `message_tag` varchar(255) DEFAULT NULL COMMENT '消息tag',
                                  `message_record_status` varchar(255) DEFAULT NULL COMMENT '消息状态',
                                  `retry_count` int DEFAULT NULL COMMENT '重试次数',
                                  `biz_type` varchar(255) DEFAULT NULL COMMENT '业务类型',
                                  `biz_uuid` varchar(255) DEFAULT NULL COMMENT '业务uuid',
                                  `mq_zone` varchar(255) DEFAULT NULL COMMENT 'mq',
                                  `producer_source` varchar(255) DEFAULT NULL COMMENT '生产者来源',
                                  `producer_key` varchar(255) DEFAULT NULL COMMENT '生产者key',
                                  `consume_sourece` varchar(255) DEFAULT NULL COMMENT '消费者来源',
                                  `sending_time` datetime DEFAULT NULL COMMENT '发送中时间',
                                  `sent_time` datetime DEFAULT NULL COMMENT '已发送时间',
                                  `ack_time` datetime DEFAULT NULL COMMENT 'ack时间',
                                  `create_time` datetime DEFAULT NULL COMMENT '创建时间',
                                  PRIMARY KEY (`message_id`),
                                  UNIQUE KEY `idx_message_uuid` (`message_uuid`),
                                  KEY `idx_create_time` (`create_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='消息记录表';


````



clone项目至本地打包成maven依赖后，导入如下依赖

```java
<dependency>
    <groupId>com.v</groupId>
    <artifactId>shanks-mq-rocketmq</artifactId>
    <version>1.0.0-SNAPSHOT</version>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>com.v</groupId>
    <artifactId>shanks-mq-data-mybaits</artifactId>
    <version>1.0.0-SNAPSHOT</version>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>com.mysql</groupId>
    <artifactId>mysql-connector-j</artifactId>
    <version>8.0.33</version>
</dependency>
```

yml配置

```yaml
spring:
  rocketMQConsumer:
    enable: true    
rocketmq:
  namesrvAddr: 192.168.145.130:9876
  default-producer:
    groupName: mq-test-system%default_producer
  default-consumer:
    groupName: mq-test-system%default_consumer
```

项目配置生产者消费者

```java
import com.shanks.framework.mq.data.mybatis.config.MybatisMQDataConfiguration;
import com.shanks.framework.mq.demo.constant.MQConstant;
import com.shanks.framework.mq.demo.processor.DemoProcessor;
import com.shanks.framework.mq.exception.MQException;
import com.shanks.framework.mq.producer.BaseMQProducer;
import com.shanks.framework.mq.rocketmq.config.RocketMQConfiguration;
import com.shanks.framework.mq.rocketmq.consumer.RocketMQConsumerAttribute;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;

@SpringBootConfiguration
@Import(MybatisMQDataConfiguration.class)
public class RocketMQConfig extends RocketMQConfiguration {

    /**
     * 配置生产者
     * @return
     * @throws MQException
     */
    @Bean
    public BaseMQProducer defaultRocketMQProducer() throws MQException {
        return createAndRegisterMQProducer(MQConstant.DEFAULT_PRODUCER, null);
    }

    /**
     * 配置生产者、消费者绑定订阅关系
     * @return
     * @throws MQException
     */
    @Bean
    @ConditionalOnProperty(name = "spring.rocketMQConsumer.enable", havingValue = "true")
    public DefaultMQPushConsumer defaultMQPushConsumer() throws MQException {
        /**
         * 配置消费者并且绑定消费关系
         */
        RocketMQConsumerAttribute rocketMQConsumerAttribute = newConsumerAttribute(MQConstant.DEFAULT_CONSUMER, null);
        rocketMQConsumerAttribute.subscribe(MQConstant.TOPIC.TOPIC_CHANGE_ORDER_STATUS, MQConstant.TAG.TAG_CHANGE_ORDER_STATUS, demoProcessor());

        /**
         * 生产者方配置消息确认topic
         */
        rocketMQConsumerAttribute.subscribeAckMsg(MQConstant.TOPIC.TOPIC_CHANGE_ORDER_STATUS, MQConstant.TAG.TAG_CHANGE_ORDER_STATUS);
        return createAndStartMQConsumer(rocketMQConsumerAttribute);
    }

    /**
     * 消费者处理器注入
     * @return
     * @throws MQException
     */
    @Bean
    public DemoProcessor demoProcessor() throws MQException{
        return new DemoProcessor(defaultRocketMQProducer());
    }
}

```

消费处理器重写

```java
import com.alibaba.fastjson.JSON;
import com.shanks.framework.mq.domain.MQMessage;
import com.shanks.framework.mq.enums.MQStatusEnum;
import com.shanks.framework.mq.processor.AbstractMQProcessor;
import com.shanks.framework.mq.producer.BaseMQProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class DemoProcessor extends AbstractMQProcessor {

    public DemoProcessor(BaseMQProducer baseMQProducer){
        setMqProducer(baseMQProducer);
    }

    private static final Logger logger = LoggerFactory.getLogger(DemoProcessor.class);
    @Override
    protected MQStatusEnum consumeMessage(MQMessage message) throws Exception {
        logger.info("接收到消息为{}", JSON.toJSON(message));
        return MQStatusEnum.SUCCESS;
    }

    @Override
    protected String getMessageDesc(MQMessage message) {
        return "TEST";
    }
}
```


### 原理

![消息原理](/image/消息原理.svg)


<p align="right">(<a href="#top">返回顶部</a>)</p>



<!-- 项目目录结构 -->

## 项目目录结构

````java
├─shanks-mq-core       // 消息流转核心实现
│  ├─src
│  │  └─main
│  │      ├─java
│  │      │  ├─com
│  │      │  │  └─shanks
│  │      │  │      └─framework
│  │      │  │          └─mq
│  │      │  │              ├─common
│  │      │  │              │  ├─constant
│  │      │  │              │  └─util
│  │      │  │              ├─configuration
│  │      │  │              ├─domain
│  │      │  │              ├─enums
│  │      │  │              ├─exception
│  │      │  │              ├─helper
│  │      │  │              ├─job
│  │      │  │              ├─model
│  │      │  │              ├─monitor
│  │      │  │              ├─processor
│  │      │  │              ├─producer
│  │      │  │              ├─queue
│  │      │  │              ├─repository
│  │      │  │              └─service
│  │      │  │                  └─impl
│  │      │  └─net
│  │      │      └─javacrumbs
│  │      │          └─shedlock
│  │      │              └─support
│  │      └─resources
│      │  └─net             // 调度任务分布式锁
│      │      └─javacrumbs
│      │          └─shedlock
│      │              └─support
├─shanks-mq-data-mybaits    // 本地消息持久层mybtis-plus实现
│  ├─src
│  │  └─main
│  │      ├─java
│  │      │  └─com
│  │      │      └─shanks
│  │      │          └─framework
│  │      │              └─mq
│  │      │                  └─data
│  │      │                      └─mybatis
│  │      │                          ├─config
│  │      │                          ├─mapper
│  │      │                          ├─repository
│  │      │                          └─service
│  │      └─resources 
├─shanks-mq-rocketmq    // 本地消息rocketmq实现
│  ├─src
│  │  └─main
│  │      └─java
│  │          └─com
│  │              └─shanks
│  │                  └─framework
│  │                      └─mq
│  │                          └─rocketmq
│  │                              ├─config
│  │                              ├─consumer
│  │                              ├─listener
│  │                              └─producer
└─shanks-mq-springboot-demo    // 使用demo
    ├─src
    │  ├─main
    │  │  ├─java
    │  │  │  └─com
    │  │  │      └─shanks
    │  │  │          └─framework
    │  │  │              └─mq
    │  │  │                  └─demo
    │  │  │                      ├─builder
    │  │  │                      ├─config
    │  │  │                      ├─constant
    │  │  │                      ├─controller
    │  │  │                      ├─entity
    │  │  │                      ├─processor
    │  │  │                      └─service
    │  │  │                          └─impl
    │  │  └─resources
    │  └─test             
    │      └─java
    │          └─com
    │              └─shanks
    │                  └─framework
    │                      └─mq
    │                          └─demo
````



<p align="right">(<a href="#top">返回顶部</a>)</p>



<!-- 许可证 -->
## 许可证

根据 MIT 许可证分发。打开 [LICENSE.txt](LICENSE.txt) 查看更多内容。


<p align="right">(<a href="#top">返回顶部</a>)</p>



<!-- 联系我们 -->
## 联系我们

邮箱 - [1649919188@qq.com](https://wx.mail.qq.com/)



<p align="right">(<a href="#top">返回顶部</a>)</p>



<!-- 致谢 -->
## 致谢

* 觉得有用麻烦点个star，谢谢

<p align="right">(<a href="#top">返回顶部</a>)</p>