DDD 架构分层，MQ消息要放到那一层处理？

作者：小傅哥
博客：https://bugstack.cn

沉淀、分享、成长，让自己和他人都能有所收获！😄

本文的宗旨在于通过简单干净实践的方式教会读者，使用 Docker 配置 RocketMQ 并在基于 DDD 分层结构的 SpringBoot 工程中使用 RocketMQ 技术。因为大部分 MQ 的发送都是基于特定业务场景的，所以本章节也是基于 《MyBatis 使用教程和插件开发》 章节的扩展。

本章也会包括关于 MQ 消息的发送和接收应该处于 DDD 的哪一层的实践讲解和使用。

本文涉及的工程：

  • xfg-dev-tech-rocketmq：https://gitcode.net/KnowledgePlanet/road-map/xfg-dev-tech-rocketmq
  • RocketMQ Docker 安装：rocketmq-docker-compose-mac-amd-arm.yml
  • 导入测试库表 road-map.sql

一、案例背景

首先我们要知道，MQ 消息的作用是用于；解耦过长的业务流程和应对流量冲击的消峰。如；用户下单支付完成后，拿到支付消息推动后续的发货流程。也可以是我们基于 《MyBatis 使用教程和插件开发》 中的案例场景，给雇员提升级别和薪资的时候，也发送一条MQ消息，用于发送邮件通知给用户。

  • 从薪资调整到邮件发送，这里是2个业务流程，通过 MQ 消息的方式进行连接。
  • 其实MQ消息的使用场景特别多，原来你可能使用多线程的一些操作，现在就扩展为多实例的操作了。发送 MQ 消息出来，让应用的各个实例接收并进行消费。

二、领域事件

因为我们本章所讲解的内容是把 RocketMQ 放入 DDD 架构中进行使用，那么也就引申出领域事件定义。所以我们先来了解下，什么是领域事件。

领域事件，可以说是解耦微服务设计的关键。领域事件也是领域模型中非常重要的一部分内容，用于标示当前领域模型中发生的事件行为。一个领域事件会推进业务流程的进一步操作，在实现业务解耦的同时，也推动了整个业务的闭环。

  • 首先，我们需要在领域模型层，添加一块 event 区域。它的存在是为了定义出于当前领域下所需的事件消息信息。信息的类型可以是model 下的实体对象、聚合对象。
  • 之后，消息的发送是放在基础设置层。本身基础设置层就是依赖倒置于模型层，所以在模型层所定义的 event 对象，可以很方便的在基础设置层使用。而且大部分开发的时候，MQ消息的发送与数据库操作都是关联的，采用的方式是，做完数据落库后，推送MQ消息。所以定义在仓储中实现，会更加得心应手、水到渠成。
  • 最后，就是 MQ 的消息，MQ 的消费可以是自身服务所发出的消息，也可以是外部其他微服务的消息。就在小傅哥所整体讲述的这套简明教程中 DDD 部分的触发器层。

三、环境安装

本案例涉及了数据库和RocketMQ的使用，都已经在工程中提供了安装脚本，可以按需执行。

这里主要介绍 RocketMQ 的安装；

1. 执行 compose yml

文件：docs/rocketmq/rocketmq-docker-compose-mac-amd-arm.yml - 关于安装小傅哥提供了不同的镜像，包括Mac、Mac M1、Windows 可以按需选择使用。

version: '3'
services:
  # https://hub.docker.com/r/xuchengen/rocketmq
  # 注意修改项；
  # 01：data/rocketmq/conf/broker.conf 添加 brokerIP1=127.0.0.1
  # 02：data/console/config/application.properties server.port=9009 - 如果8080端口被占用，可以修改或者添加映射端口
  rocketmq:
    image: livinphp/rocketmq:5.1.0
    container_name: rocketmq
    ports:
      - 9009:9009
      - 9876:9876
      - 10909:10909
      - 10911:10911
      - 10912:10912
    volumes:
      - ./data:/home/app/data
    environment:
      TZ: "Asia/Shanghai"
      NAMESRV_ADDR: "rocketmq:9876"
  • 在 IDEA 中打开 rocketmq-docker-compose-mac-amd-arm.yml 你会看到一个绿色的按钮在左侧侧边栏，点击即可安装。或者你也可以使用命令安装：# /usr/local/bin/docker-compose -f /docs/dev-ops/environment/environment-docker-compose.yml up -d - 比较适合在云服务器上执行。
  • 首次安装可能使用不了，一个原因是 brokerIP1 未配置IP，另外一个是默认的 8080 端口占用。可以按照如下小傅哥说的方式修改。

2. 修改默认配合

  1. 打开 data/rocketmq/conf/broker.conf 添加一条 brokerIP1=127.0.0.1 在结尾
# 集群名称
brokerClusterName = DefaultCluster
# BROKER 名称
brokerName = broker-a
# 0 表示 Master, > 0 表示 Slave
brokerId = 0
# 删除文件时间点，默认凌晨 4 点
deleteWhen = 04
# 文件保留时间，默认 48 小时
fileReservedTime = 48
# BROKER 角色 ASYNC_MASTER为异步主节点，SYNC_MASTER为同步主节点，SLAVE为从节点
brokerRole = ASYNC_MASTER
# 刷新数据到磁盘的方式，ASYNC_FLUSH 刷新
flushDiskType = ASYNC_FLUSH
# 存储路径
storePathRootDir = /home/app/data/rocketmq/store
# IP地址
brokerIP1 = 127.0.0.1
  1. 打开 ``data/console/config/application.properties修改server.port=9009` 端口。
server.address=0.0.0.0
server.port=9009
  • 修改配置后，重启服务。

3. RockMQ登录与配置

3.1 登录

RocketMQ 此镜像，会在安装后在控制台打印登录账号信息，你可以查看使用。

登录：http://localhost:9009/

3.2 创建Topic

  • 也可以使用命令创建：docker exec -it rocketmq sh /home/app/rocketmq/bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t xfg-mq

3.3 创建消费者组

  • 也可以使用命令创建：docker exec -it rocketmq sh /home/app/rocketmq/bin/mqadmin updateSubGroup -n localhost:9876 -c DefaultCluster -g xfg-group

四、工程实现

1. 工程结构

  • MQ 的使用无论是 RocketMQ 还是 Kafka 等，都很简单。但在使用之前，要考虑好怎么在架构中合理的使用。如果最初没有定义好这些，那么胡乱的任何地方都能发送和接收MQ，最后的工程将非常难以维护。
  • 所以这里整个MQ的生产和消费，是按照整个 DDD 领域事件结构进行设计。分为在 domain 使用基础层生产消息，再有 trigger 层接收消息。

2. 配置文件

引入POM

<!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-client-java -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client-java</artifactId>
    <version>5.0.4</version>
</dependency>
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.0</version>
</dependency>

添加配置

# RocketMQ 配置
rocketmq:
  name-server: 127.0.0.1:9876
  consumer:
    group: xfg-group
    # 一次拉取消息最大值，注意是拉取消息的最大值而非消费最大值
    pull-batch-size: 10
  producer:
    # 发送同一类消息的设置为同一个group，保证唯一
    group: xfg-group
    # 发送消息超时时间，默认3000
    sendMessageTimeout: 10000
    # 发送消息失败重试次数，默认2
    retryTimesWhenSendFailed: 2
    # 异步消息重试此处，默认2
    retryTimesWhenSendAsyncFailed: 2
    # 消息最大长度，默认1024 * 1024 * 4(默认4M)
    maxMessageSize: 4096
    # 压缩消息阈值，默认4k(1024 * 4)
    compressMessageBodyThreshold: 4096
    # 是否在内部发送失败时重试另一个broker，默认false
    retryNextServer: false

3. 定义领域事件

源码：cn.bugstack.xfg.dev.tech.domain.salary.event.SalaryAdjustEvent

@EqualsAndHashCode(callSuper = true)
@Data
public class SalaryAdjustEvent extends BaseEvent<AdjustSalaryApplyOrderAggregate> {

    public static String TOPIC = "xfg-mq";

    public static SalaryAdjustEvent create(AdjustSalaryApplyOrderAggregate adjustSalaryApplyOrderAggregate) {
        SalaryAdjustEvent event = new SalaryAdjustEvent();
        event.setId(RandomStringUtils.randomNumeric(11));
        event.setTimestamp(new Date());
        event.setData(adjustSalaryApplyOrderAggregate);
        return event;
    }

}
  • 每个领域的消息，都有领域自己定义。发送的时候再交给基础设施层来发送。

4. 消息发送

源码：cn.bugstack.xfg.dev.tech.infrastructure.event.EventPublisher

@Component
@Slf4j
public class EventPublisher {

    @Setter(onMethod_ = @Autowired)
    private RocketMQTemplate rocketmqTemplate;

    /**
     * 普通消息
     *
     * @param topic   主题
     * @param message 消息
     */
    public void publish(String topic, BaseEvent<?> message) {
        try {
            String mqMessage = JSON.toJSONString(message);
            log.info("发送MQ消息 topic:{} message:{}", topic, mqMessage);
            rocketmqTemplate.convertAndSend(topic, mqMessage);
        } catch (Exception e) {
            log.error("发送MQ消息失败 topic:{} message:{}", topic, JSON.toJSONString(message), e);
            // 大部分MQ发送失败后，会需要任务补偿
        }
    }

    /**
     * 延迟消息
     *
     * @param topic          主题
     * @param message        消息
     * @param delayTimeLevel 延迟时长
     */
    public void publishDelivery(String topic, BaseEvent<?> message, int delayTimeLevel) {
        try {
            String mqMessage = JSON.toJSONString(message);
            log.info("发送MQ延迟消息 topic:{} message:{}", topic, mqMessage);
            rocketmqTemplate.syncSend(topic, MessageBuilder.withPayload(message).build(), 1000, delayTimeLevel);
        } catch (Exception e) {
            log.error("发送MQ延迟消息失败 topic:{} message:{}", topic, JSON.toJSONString(message), e);
            // 大部分MQ发送失败后，会需要任务补偿
        }
    }

}
  • 在基础设施层提供 event 事件的处理，也就是 MQ 消息的发送。

源码：cn.bugstack.xfg.dev.tech.infrastructure.repository.SalaryAdjustRepository

@Resource
private EventPublisher eventPublisher;
    
@Override
@Transactional(rollbackFor = Exception.class, timeout = 350, propagation = Propagation.REQUIRED, isolation = Isolation.DEFAULT)
public String adjustSalary(AdjustSalaryApplyOrderAggregate adjustSalaryApplyOrderAggregate) {
		 
		// ... 省略部分代码 

    eventPublisher.publish(SalaryAdjustEvent.TOPIC, SalaryAdjustEvent.create(adjustSalaryApplyOrderAggregate));
    return orderId;
}

在 SalaryAdjustRepository 仓储的实现中，做完业务流程开始发送 MQ 消息。这里有2点要注意；

  1. 消息发送，不要写在数据库事务中。因为事务一直占用数据库连接，需要快速释放。
  2. 对于一些强MQ要求的场景，需要在发送MQ前，写入一条数据库 Task 记录，发送消息后更新 Task 状态为成功。如果长时间未更新数据库状态或者为失败的，则需要由任务补偿进行处理。

5. 消费消息

源码：cn.bugstack.xfg.dev.tech.trigger.mq.SalaryAdjustMQListener

@Component
@Slf4j
@RocketMQMessageListener(topic = "xfg-mq", consumerGroup = "xfg-group")
public class SalaryAdjustMQListener implements RocketMQListener<String> {

    @Override
    public void onMessage(String s) {
        log.info("接收到MQ消息 {}", s);
    }

}
  • 消费消息，配置消费者组合消费的主题，之后就可以接收到消息了。接收以后你可以做自己的业务，如果抛出异常，消息会进行重新接收处理。

六、测试验证

1. 单独发送消息测试

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class RocketMQTest {

    @Setter(onMethod_ = @Autowired)
    private RocketMQTemplate rocketmqTemplate;

    @Test
    public void test() throws InterruptedException {
        while (true) {
            rocketmqTemplate.convertAndSend("xfg-mq", "我是测试消息");
            Thread.sleep(3000);
        }
    }

}
  • 这里方便你来发送消息，验证流程。

2. 业务流程消息验证

@Test
public void test_execSalaryAdjust() throws InterruptedException {
    AdjustSalaryApplyOrderAggregate adjustSalaryApplyOrderAggregate = AdjustSalaryApplyOrderAggregate.builder()
            .employeeNumber("10000001")
            .orderId("100908977676003")
            .employeeEntity(EmployeeEntity.builder().employeeLevel(EmployeePostVO.T3).employeeTitle(EmployeePostVO.T3).build())
            .employeeSalaryAdjustEntity(EmployeeSalaryAdjustEntity.builder()
                    .adjustTotalAmount(new BigDecimal(100))
                    .adjustBaseAmount(new BigDecimal(80))
                    .adjustMeritAmount(new BigDecimal(20)).build())
            .build();
    String orderId = salaryAdjustApplyService.execSalaryAdjust(adjustSalaryApplyOrderAggregate);
    log.info("调薪测试 req: {} res: {}", JSON.toJSONString(adjustSalaryApplyOrderAggregate), orderId);
    Thread.sleep(Integer.MAX_VALUE);
}

23-07-29.15:40:52.307 [main            ] INFO  HikariDataSource       - HikariPool-1 - Start completed.
23-07-29.15:40:52.445 [main            ] INFO  EventPublisher         - 发送MQ消息 topic:xfg-mq message:{"data":{"employeeEntity":{"employeeLevel":"T3","employeeTitle":"T3"},"employeeNumber":"10000001","employeeSalaryAdjustEntity":{"adjustBaseAmount":80,"adjustMeritAmount":20,"adjustTotalAmount":100},"orderId":"100908977676004"},"id":"98117654515","timestamp":"2023-07-29 15:40:52.425"}
23-07-29.15:40:52.517 [main            ] INFO  ISalaryAdjustApplyServiceTest - 调薪测试 req: {"employeeEntity":{"employeeLevel":"T3","employeeTitle":"T3"},"employeeNumber":"10000001","employeeSalaryAdjustEntity":{"adjustBaseAmount":80,"adjustMeritAmount":20,"adjustTotalAmount":100},"orderId":"100908977676004"} res: 100908977676004
23-07-29.15:40:52.520 [ConsumeMessageThread_1] INFO  SalaryAdjustMQListener - 接收到MQ消息 {"data":{"employeeEntity":{"employeeLevel":"T3","employeeTitle":"T3"},"employeeNumber":"10000001","employeeSalaryAdjustEntity":{"adjustBaseAmount":80,"adjustMeritAmount":20,"adjustTotalAmount":100},"orderId":"100908977676004"},"id":"98117654515","timestamp":"2023-07-29 15:40:52.425"}
  • 当执行一次加薪调整后，就会接收到MQ消息了。

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

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

厦门逗客传媒：抖音本地团购怎么入驻

厦门逗客传媒：抖音本地团购怎么入驻

随着社交媒体的不断发展&#xff0c;短视频平台已经成为了商家推广和营销的热门渠道之一。在这其中&#xff0c;抖音作为全球知名的短视频平台&#xff0c;以其巨大的用户基数和精准的推荐算法吸引了大量商家的关注。而在抖音上&#xff0c;本地团购也成为了一个备受关注的领域…
阅读更多...
【Python】强化学习：原理与Python实战

【Python】强化学习：原理与Python实战

搞懂大模型的智能基因&#xff0c;RLHF系统设计关键问答 RLHF&#xff08;Reinforcement Learning with Human Feedback&#xff0c;人类反馈强化学习&#xff09;虽是热门概念&#xff0c;并非包治百病的万用仙丹。本问答探讨RLHF的适用范围、优缺点和可能遇到的问题&#xff…
阅读更多...
商城-学习整理-高级-商城业务-异步线程池（十三）

商城-学习整理-高级-商城业务-异步线程池（十三）

目录 一、线程1、初始化线程的 4 种方式2、线程池的七大参数3、线程池的运行流程&#xff1a;4、例子5、常见的 4 种线程池6、开发中为什么使用线程池 二、CompletableFuture 异步编排0、业务场景&#xff1a;1、创建异步对象2、计算完成时回调方法3、handle 方法4、线程串行化…
阅读更多...
自动化测试平台seldom-platform部署及使用

自动化测试平台seldom-platform部署及使用

介绍 seldom-platform是一个基于seldom测试框架的测试平台 项目地址&#xff1a;https://github.com/SeldomQA 文档&#xff1a;seldom 语雀 首先&#xff0c;专门为seldom测试框架提供平台化支持。其次&#xff0c;只负责自动化测试项目的解析、执行用例&#xff0c;当然…
阅读更多...
【Django】Task4 序列化及其高级使用、ModelViewSet

【Django】Task4 序列化及其高级使用、ModelViewSet

【Django】Task4 序列化及其高级使用、ModelViewSet Task4主要了解序列化及掌握其高级使用&#xff0c;了解ModelViewSet的作用&#xff0c;ModelViewSet 是 Django REST framework&#xff08;DRF&#xff09;中的一个视图集类&#xff0c;用于快速创建处理模型数据的 API 视…
阅读更多...
十八、深度学习模型30年演化史

十八、深度学习模型30年演化史

1、模型分类 深度学习是解决问题的一系列模型与方法，但深度学习模型不是深度学习领域中唯一的研究方向，且不一定是最重要的研究方向。除了模型之外，比较重要的还有优化算法、损失函数、采样方法等。 1.1 DNN 深度神经网络（Deep Neural Networks， 以下简称DNN）是…
阅读更多...
解决Oracle中XML插入数据时的空格问题

解决Oracle中XML插入数据时的空格问题

&#x1f337;&#x1f341; 博主猫头虎 带您 Go to New World.✨&#x1f341; &#x1f984; 博客首页——猫头虎的博客&#x1f390; &#x1f433;《面试题大全专栏》 文章图文并茂&#x1f995;生动形象&#x1f996;简单易学&#xff01;欢迎大家来踩踩~&#x1f33a; &a…
阅读更多...
windows上ffmpeg如何录制双屏幕中的一个屏幕上的视频

windows上ffmpeg如何录制双屏幕中的一个屏幕上的视频

首先&#xff0c;如何在window上安装ffmpeg自己查找scoop安装ffmpeg. 如题&#xff1a; 如果你有两个屏幕&#xff0c;如何让ffmpeg来录制其中的一个屏幕的视频呢。 很简单&#xff0c;首先你要查看另外一个屏幕的分辨率&#xff1a; 第一步&#xff1a;进入系统中 第二步&am…
阅读更多...
基于OpenCV实战（基础知识一）

基于OpenCV实战（基础知识一）

目录 简介 1.计算机眼中的图像 2.图片的读取、显示与保存 3.视频的读取与显示 简介 OpenCV是一个流行的开源计算机视觉库&#xff0c;由英特尔公司发起发展。它提供了超过2500个优化算法和许多工具包&#xff0c;可用于灰度、彩色、深度、基于特征和运动跟踪等的图像处理和…
阅读更多...
Spring Clould 负载均衡 - Ribbon

Spring Clould 负载均衡 - Ribbon

视频地址&#xff1a;微服务&#xff08;SpringCloudRabbitMQDockerRedis搜索分布式&#xff09; Ribbon-负载均衡原理&#xff08;P14&#xff09; 具体实现时通过LoaBalanced注解实现&#xff0c;表示RestTemplate要被Ribbon拦截处理 orderservice调用user时候&#xff0c…
阅读更多...
用队列实现栈

用队列实现栈

目录 题目题目要求示例 解答方法一、实现思路时间复杂度和空间复杂度代码 方法二、实现思路时间复杂度和空间复杂度代码 方法三、实现思路时间复杂度和空间复杂度代码 总结 题目 用队列实现栈 题目要求 题目链接 示例 解答 方法一、 使用两个队列来实现栈。 实现思路 题…
阅读更多...
科技成果鉴定测试有什么意义?专业CMA、CNAS软件测评公司

科技成果鉴定测试有什么意义?专业CMA、CNAS软件测评公司

科技成果鉴定测试是指通过一系列科学的实验和检测手段&#xff0c;对科技成果进行客观评价和鉴定的过程。通过测试&#xff0c;可以对科技成果的技术优劣进行评估&#xff0c;从而为科技创新提供参考和指导。 一、科技成果鉴定测试的意义 1、帮助客户了解科技产品的性能特点和…
阅读更多...
【Spring】一次性打包学透 Spring | 阿Q送书第五期

【Spring】一次性打包学透 Spring | 阿Q送书第五期

文章目录 如何竭尽可能确保大家学透Spring1. 内容全面且细致2. 主题实用且本土化3. 案例系统且完善4. 知识有趣且深刻 关于作者丁雪丰业内专家推图书热卖留言提前获赠书 不知从何时开始&#xff0c;Spring 这个词开始频繁地出现在 Java 服务端开发者的日常工作中&#xff0c;很…
阅读更多...
nvm安装使用教程

nvm安装使用教程

文章目录 下载配置安装最新稳定版 node安装指定版本查看版本切换版本删除版本 常见问题安装node后 显示拒绝访问的问题使用cnpm会报错的问题降低cnpm版本npm镜像 下载 NVM for Windows 下载地址&#xff1a;https://link.juejin.cn/?targethttps%3A%2F%2Fgithub.com%2Fcoreyb…
阅读更多...
java学习网站_Java程序员必看的学习网站

java学习网站_Java程序员必看的学习网站

哔哩哔哩 B 站动力节点官方视频 https://space.bilibili.com/76542346?spm_id_from333.337.0.0 超多视频免费白嫖 专注 java 培训的动力节点 毕业学员业内高薪就业&#xff0c;逐渐得到了业界广大的好评&#xff0c;被业界誉为 “口口相传的 Java 黄埔军校 “。 网址&#x…
阅读更多...
【计算机网络】【常考问题总结】

【计算机网络】【常考问题总结】

1. ping 127.0.0.1 后会发生什么&#xff1f; ping 127.0.0.1 &#xff1b;ping 0.0.0.0 &#xff1b; ping localhost 面试官问&#xff1a;断网了&#xff0c;还能ping通 127.0.0.1 吗&#xff1f;为什么&#xff1f;_kevin_tech的博客-CSDN博客 2. MTU&#xff0c;MMU是…
阅读更多...
某多多商品平台数据采集

某多多商品平台数据采集

某多多商品平台数据采集 声明逆向目标寻找加密位置代码分析补环境补充内容声明 本文章中所有内容仅供学习交流，严禁用于商业用途和非法用途，否则由此产生的一切后果均与作者 无关，若有侵权，请私信我立即删除! 逆向目标 Anti-Content参数 寻找加密位置 先在控制台全局搜…
阅读更多...
【从零学习python 】69. 网络通信及IP地址分类解析

【从零学习python 】69. 网络通信及IP地址分类解析

文章目录 网络通信的概念IP地址IP地址的分类A类地址B类地址C类地址D类地址E类地址私有地址 进阶案例 网络通信的概念 简单来说&#xff0c;网络是用物理链路将各个孤立的工作站或主机相连在一起&#xff0c;组成数据链路&#xff0c;从而达到资源共享和通信的目的。 使用网络…
阅读更多...
k8s-ingress-context deadline exceeded

k8s-ingress-context deadline exceeded

报错&#xff1a; rancher-rke-01:~/rke # helm install rancher rancher-latest/rancher --namespace cattle-system --set hostnamewww.rancher.local Error: INSTALLATION FAILED: Internal error occurred: failed calling webhook "validate.nginx.ingress.kube…
阅读更多...
leetcode 118.杨辉三角

leetcode 118.杨辉三角

⭐️ 题目描述 &#x1f31f; leetcode链接&#xff1a;https://leetcode.cn/problems/pascals-triangle/description/ 代码&#xff1a; class Solution { public:vector<vector<int>> generate(int numRows) {// 先开空间vector<vector<int>> v;v.…
阅读更多...
最新文章

Copyright @ 2022~2023