可靠消息投递demo

news2025/3/22 5:36:28

以下是一个基于 Spring Boot + RocketMQ 的完整分布式事务实战 Demo，包含事务消息、本地事务、自动重试、死信队列（DLQ） 等核心机制。代码已充分注释，可直接运行。

一、项目结构

src/main/java
├── com.example.rocketmq
│   ├── controller
│   │   └── OrderController.java
│   ├── model
│   │   ├── Order.java
│   │   ├── OrderRequest.java
│   ├── repository
│   │   ├── OrderRepository.java
│   ├── service
│   │   ├── InventoryService.java
│   │   ├── OrderService.java
│   │   ├── PaymentService.java
│   ├── listener
│   │   └── OrderConsumer.java
│   └── RocketMQConfig.java
├── application.yml
└── pom.xml

在这里插入图片描述

二、依赖配置（pom.xml）

<dependencies>
    <!-- Spring Boot Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    
    <!-- RocketMQ Spring Boot Starter -->
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.11.0</version>
    </dependency>
    
    <!-- MySQL Driver -->
    <dependency>
        <groupId>com.mysql</groupId>
        <artifactId>mysql-connector-j</artifactId>
        <scope>runtime</scope>
    </dependency>
    
    <!-- Spring Data JPA -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
</dependencies>

三、配置文件（application.yml）

# RocketMQ配置
rocketmq:
  producer:
    name-server: localhost:9876
    default-topic: order_topic
  consumer:
    name-server: localhost:9876
    default-topic: order_topic
    consumer-group: order_consumer_group
    acknowledge-mode: AUTO
    max-reconsume-times: 5 # 最大重试次数
  broker:
    role: SYNC_MASTER # 同步复制模式
    store-path-commit-log: /data/rocketmq/commitlog
    store-path-consume-queue: /data/rocketmq/consumequeue

# 数据库配置
spring:
  datasource:
    url: jdbc:mysql://localhost:3306/rocketmq_db?useSSL=false&serverTimezone=UTC
    username: root
    password: root
    driver-class-name: com.mysql.cj.jdbc.Driver
    hikari:
      maximum-pool-size: 20
  jpa:
    hibernate:
      ddl-auto: update
    show-sql: true

四、核心代码实现

1. 实体类（Order.java）
@Entity
@Table(name = "orders")
@Data
public class Order {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String userId;
    private BigDecimal amount;
    private String sku;
    private Integer status; // 0-待支付，1-已支付，2-已发货
}
2. 生产者代码（OrderService.java）
@Service
public class OrderService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private InventoryService inventoryService;

    @Autowired
    private PaymentService paymentService;

    @Transactional // 本地事务
    public void createOrder(OrderRequest request) {
        // 1. 扣减库存（本地事务）
        inventoryService.deduct(request.getSku());
        
        // 2. 发送事务消息（与本地事务绑定）
        rocketMQTemplate.sendMessageInTransaction(
            "order_topic", 
            request, 
            () -> { // 事务回滚回调
                System.out.println("本地事务回滚，消息未发送！");
                return null;
            }
        );
    }
}
3. 消费者代码（OrderConsumer.java）
@Service
public class OrderConsumer {

    @Autowired
    private OrderRepository orderRepository;

    @Autowired
    private PaymentService paymentService;

    @Autowired
    private InventoryService inventoryService;

    @RocketMQListener(
        topics = "order_topic",
        consumerGroup = "order_consumer_group",
        acknowledge-mode = AcknowledgeMode.AUTO
    )
    public void listen(OrderRequest request) {
        try {
            // 1. 生成订单记录
            Order order = new Order();
            order.setUserId(request.getUserId());
            order.setAmount(request.getAmount());
            order.setSku(request.getSku());
            orderRepository.save(order);

            // 2. 扣款（外部服务调用）
            paymentService.charge(request.getUserId(), request.getAmount());

            // 3. 发送物流通知（模拟成功）
            System.out.println("物流已通知，订单号: " + order.getId());
        } catch (Exception e) {
            // 4. 异常处理：触发重试或补偿
            System.out.println("处理失败，触发重试! 订单号: " + request.getOrderNo());
            throw new RuntimeException("订单处理失败", e);
        }
    }
}
4. 支付服务（PaymentService.java）
@Service
public class PaymentService {

    @Autowired
    private PaymentRepository paymentRepository;

    public void charge(String userId, BigDecimal amount) {
        // 模拟支付失败（30%概率）
        if (Math.random() < 0.3) {
            throw new RuntimeException("支付失败，用户: " + userId);
        }

        Payment payment = new Payment();
        payment.setUserId(userId);
        payment.setAmount(amount);
        payment.setStatus("SUCCESS");
        paymentRepository.save(payment);
    }
}

五、数据库表设计

1. 订单表（orders）
CREATE TABLE orders (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    user_id VARCHAR(50) NOT NULL,
    amount DECIMAL(10,2) NOT NULL,
    sku VARCHAR(50) NOT NULL,
    status TINYINT DEFAULT 0 COMMENT '0-待支付，1-已支付，2-已发货'
);
2. 支付表（payments）
CREATE TABLE payments (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    user_id VARCHAR(50) NOT NULL,
    amount DECIMAL(10,2) NOT NULL,
    status ENUM('SUCCESS', 'FAILED') DEFAULT 'SUCCESS'
);

六、测试与验证

1. 正常流程

• 步骤：

  1. 发送创建订单请求（扣减库存 + 发送事务消息）。
  2. 消费者处理消息（生成订单 + 扣款 + 物流通知）。
    • 预期结果：
    • 库存减少，订单和支付记录生成，物流通知成功。
2. 异常流程（支付失败）

• 步骤：

  1. 发送创建订单请求。
  2. 消费者处理时支付失败（抛出异常）。
  3. RocketMQ自动重试（默认3次）。
  4. 重试失败后消息转入DLQ。
    • 预期结果：
    • 库存已恢复（通过本地事务回滚）。
    • 订单未生成，支付记录未插入。
3. DLQ处理

• 操作：手动消费DLQ中的消息，排查支付失败原因（如用户余额不足）。
• 代码示例：

@RocketMQListener(
    topics = "order_topic_DLQ",
    consumerGroup = "order_consumer_group_dlq"
)
public void listenDLQ(OrderRequest request) {
    System.out.println("处理死信消息: " + request.getOrderNo());
    // 人工干预逻辑（如短信通知用户）
}

七、关键机制说明

1. 事务消息与本地事务绑定

• 代码示例：sendMessageInTransaction 方法将消息发送与本地事务提交原子化。
• 流程：
• 本地事务成功 → RocketMQ持久化消息。
• 本地事务失败 → RocketMQ丢弃消息。

2. 自动重试与死信队列

• 配置：max-reconsume-times=5 表示最大重试5次。
• DLQ Topic：默认死信队列名称为 order_topic_Retry，可通过 spring.rabbitmq.listener.defaultDLQ 配置。

3. ACK确认机制

• 自动ACK：消费者处理完消息后自动发送确认，RocketMQ删除消息。
• 手动ACK（可选）：通过 AcknowledgeMode.MANUAL 控制。

八、生产环境优化建议

  1. 持久化配置：
    • 确保 storePathCommitLog 和 storePathConsumeQueue 指向持久化磁盘路径。
  2. 多Broker集群：
    • 部署多个Broker节点，配置 brokerRole=SYNC_MASTER 实现高可用。
  3. 监控与报警：
    • 监控 ConsumerLag 和 MessagesPending 指标，阈值报警。
  4. 日志记录：
    • 启用RocketMQ日志（log4j2.xml），记录消息生产、消费详情。

九、总结

通过本Demo，你已掌握以下核心技能：

  1. 事务消息：结合本地事务实现强一致性。
  2. 自动重试：处理临时性故障（如网络抖动）。
  3. 死信队列：隔离无法处理的异常消息。
  4. 监控与运维：通过指标和日志保障系统稳定性。

下一步行动：
• 将Demo部署到Docker容器，模拟高并发场景。
• 结合Seata框架实现更复杂的分布式事务（如订单-库存-支付三阶段）。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2319354.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

Copilot提示词库用法：调整自己想要的，记住常用的，分享该共用的

Copilot提示词库用法：调整自己想要的，记住常用的，分享该共用的

不论你是 Microsoft 365 Copilot 的新用户还是熟练运用的老鸟&#xff0c;不论你是使用copilot chat&#xff0c;还是在office365中使用copilot&#xff0c;copilot提示词库都将帮助你充分使用copilot这一划时代的产品。它不仅可以帮助你记住日常工作中常用的prompt提示词&…
阅读更多...
Python实战（3）-数据库操作

Python实战（3）-数据库操作

前面说过&#xff0c;可用的SQL数据库引擎有很多&#xff0c;它们都有相应的Python模块。这些数据库引擎大都作为服务器程序运行&#xff0c;连安装都需要有管理员权限。为降低Python DB API的使用门槛&#xff0c;我选择了一个名为SQLite的小型数据库引擎。它不需要作为独立的…
阅读更多...
LeetCode 160 Intersection Of Two Linked Lists 相交链表 Java

LeetCode 160 Intersection Of Two Linked Lists 相交链表 Java

题目&#xff1a;找到两个相交列表的起始点&#xff0c;如图c1开始为A和B两个链表的相交点 举例1&#xff1a;8为两个链表的相交点。 注意&#xff1a;相交不止是数值上的相同。 举例2&#xff1a;2为相交点 举例3&#xff1a;没有相交点 解题思路&#xff1a; 相交证明最后一…
阅读更多...
AI Agent中的MCP详解

AI Agent中的MCP详解

一、协议定义与核心价值 MCP（Model Context Protocol，模型上下文协议）是由Anthropic公司于2024年11月推出的开放标准协议，其核心目标是通过建立统一接口规范，解决AI模型与外部系统集成效率低下的行业痛点。该协议通过标准化通信机制，使大型语言模型（LLM）能够无缝对接数…
阅读更多...
win系统上自动化安装配置WSL linux和各种生信工具教程

win系统上自动化安装配置WSL linux和各种生信工具教程

windows系统上自动化安装配置WSL linux系统和各种生信工具教程 高通量测序原始数据的上游分析模块介绍 我开发的OmicsTools软件的这些分析测序原始数据的上游处理分析模块需要使用到linux和linux系统中的一些生信工具&#xff0c;在这里我开发了在windows系统中自动化安装WSL …
阅读更多...
PowerBI纯小白如何驾驭DAX公式一键生成：copilot for fabric

PowerBI纯小白如何驾驭DAX公式一键生成：copilot for fabric

在2025年2月份更新中&#xff0c;powerbi desktop里的copilot功能还新增了一个非常强大的功能&#xff1a;一键生成多个度量值&#xff0c;并直接加载到模型。 直接上示例展示&#xff1a; 打开DAX查询视图&#xff0c;在copilot窗格中直接输入想要生成多个度量值&#xff0c…
阅读更多...
两市总的净流出和净流入来分析情况

两市总的净流出和净流入来分析情况

为了排查数据干扰&#xff0c;只从两市总的净流出和净流入来分析情况。 净流出才对应资金抽离&#xff1a;若净流入为负&#xff08;即净流出&#xff09;&#xff0c;则意味着资金从股市中撤出&#xff0c;例如主动卖出的金额超过主动买入金额。净流入反映市场信心&#xff1…
阅读更多...
如何在SQL中高效使用聚合函数、日期函数和字符串函数：实用技巧与案例解析

如何在SQL中高效使用聚合函数、日期函数和字符串函数：实用技巧与案例解析

文章目录 聚合函数group by子句的使用实战OJ日期函数字符串函数数学函数其它函数 聚合函数 函数说明COUNT([DISTINCT] expr)返回查询到的数据的 数量SUM([DISTINCT] expr)返回查询到的数据的 总和&#xff0c;不是数字没有意义AVG([DISTINCT] expr)返回查询到的数据的 平均值&…
阅读更多...
k8s-coredns-CrashLoopBackOff 工作不正常

k8s-coredns-CrashLoopBackOff 工作不正常

本文作者&#xff1a; slience_me 问题描述 # 问题描述 # rootk8s-node1:/home/slienceme# kubectl get pods --all-namespaces # NAMESPACE NAME READY STATUS RESTARTS AGE # kube-flannel kube-flannel-ds-66bcs …
阅读更多...
【Android性能】Systrace分析

【Android性能】Systrace分析

1&#xff0c;分析工具 1&#xff0c;Systrace新UI网站 Perfetto UI 2&#xff0c;Systrace抓取 可通过android sdk中自带的systrace抓取&#xff0c;路径一般如下&#xff0c;..\AppData\Local\Android\Sdk\platform-tools&#xff0c; 另外需要安装python2.7&#xff0c;…
阅读更多...
Unity导出WebGL，无法显示中文

Unity导出WebGL，无法显示中文

问题&#xff1a;中文无法显示 默认字体无法显示中文 在编辑器中设置了中文和英文的按钮&#xff0c;中文按钮无法显示 导出后无法显示中文 解决办法&#xff1a; 自己添加字体&#xff0c;导入项目&#xff0c;并引用 示例 下载一个字体文件&#xff0c;这里使用的阿里…
阅读更多...
oracle事务的组成

oracle事务的组成

1)数据库事务由以下的部分组成: 一个或多个DML 语句 ; 一个 DDL(Data Definition Language – 数据定义语言) 语句&#xff1b; 一个 DCL(Data Control Language – 数据控制语言)语句&#xff1b; 2)事务的执行开始&#xff1a; 以第一个 DML 语句的执行作为开始 &#xff0c;…
阅读更多...
【如何在OpenWebUI中使用FLUX绘画：基于硅基流动免费API的完整指南】

【如何在OpenWebUI中使用FLUX绘画：基于硅基流动免费API的完整指南】

如何在OpenWebUI中使用FLUX绘画&#xff1a;基于硅基流动免费API的完整指南 注册并获取硅基流动秘钥OpenWebUI中使用函数配置自定义模型-提示词配置效果验证 ) FLUX绘画是一种强大的AI绘图工具&#xff0c;本文将详细介绍如何在OpenWebUI中集成并使用FLUX绘画功能&#xff0c;…
阅读更多...
Facebook 如何影响元宇宙的发展趋势

Facebook 如何影响元宇宙的发展趋势

Facebook 如何影响元宇宙的发展趋势 引言 元宇宙&#xff08;Metaverse&#xff09;这个概念&#xff0c;曾经只存在于科幻小说中&#xff0c;如今正逐渐成为现实。它是一个由多个 3D 虚拟世界组成的网络&#xff0c;用户可以在其中进行社交、游戏、工作等活动。Facebook&…
阅读更多...
1.5.7 掌握Scala内建控制结构 - 变量作用域

1.5.7 掌握Scala内建控制结构 - 变量作用域

本次实战深入理解了Scala中变量作用域的概念&#xff0c;通过两个任务演示了作用域的基本规则。在任务1中&#xff0c;我们创建了一个名为ScopeDemo01的对象&#xff0c;展示了内部作用域能够访问外部作用域的变量。通过在if语句块中访问在外部定义的message变量&#xff0c;我…
阅读更多...
RAID磁盘阵列管理

RAID磁盘阵列管理

一. 什么是RAID RAID是英文Redundant Array of Independent Disks的缩写&#xff0c;中文翻译过来就是“独立冗余磁盘阵列”。简单的说&#xff0c;RAID是一种把多块独立的硬盘&#xff08;物理硬盘&#xff09;按不同的方式组合起来形成一个硬盘组&#xff08;逻辑硬盘&#…
阅读更多...
微博ip属地不发微博会不会变

微博ip属地不发微博会不会变

随着社交媒体的普及&#xff0c;微博作为其中的佼佼者&#xff0c;一直备受关注。而且微博上线了显示用户IP属地的功能&#xff0c;这一功能旨在减少冒充热点事件当事人、恶意造谣、蹭流量等不良行为&#xff0c;确保传播内容的真实性和透明度。然而&#xff0c;这也引发了一些…
阅读更多...
C#从入门到精通（1）

C#从入门到精通（1）

目录 第一章 C#与VS介绍 第二章 第一个C#程序 &#xff08;1&#xff09;C#程序基本组成 1.命名空间 2.类 3.Main方法 4.注释 5.语句 6.标识符及关键字 &#xff08;2&#xff09;程序编写规范 1.代码编写规则 2.程序命名方法 3.元素命名规范 第三章 变量 &…
阅读更多...
仿“东方甄选”直播商城小程序运营平台

仿“东方甄选”直播商城小程序运营平台

在公域直播流量红利趋于饱和、流量成本大幅攀升的当下&#xff0c;私域直播为企业开辟了新的流量聚集和转化渠道&#xff0c;特别是对于那些希望在私域流量领域取得突破的品牌商家来说&#xff0c;直播场景以其独特的高频互动氛围&#xff0c;相比其他运营方式&#xff0c;展现…
阅读更多...
ORACLE 19.8版本数据库环境EXPDP导数据的报错处理

ORACLE 19.8版本数据库环境EXPDP导数据的报错处理

近期用户在做EXPDP导出时&#xff0c;报错异常termination终止;EXPDP本身是简单的功能并且这个环境也是经常做导出的&#xff0c;到底是什么原因导致了这个问题呢&#xff1f; 导出脚本报错&#xff1a; 分析导出日志&#xff0c;当时系统资源充足但是进程启动失败&#xff0c;…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023