DDD 架构分层,MQ消息要放到那一层处理?

这篇具有很好参考价值的文章主要介绍了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文章来源地址https://www.toymoban.com/news/detail-654413.html

@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消息了。

到了这里,关于DDD 架构分层,MQ消息要放到那一层处理?的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Java架构师设计模式分层架构

    想学习架构师构建流程请跳转:Java架构师系统架构设计 设计模式的分层架构是一种常见的软件设计模式,它将应用程序划分为不同的层次,以便更好地组织和管理代码。每个层次

    2024年02月01日
    浏览(33)
  • Java开发中的分层开发和整洁架构

    maven多模块开发项目管理. 可以利用这种管理功能,实现一个项目的多层次模块开发–分层开发. 比如,当前项目HelloController依赖HelloService 这样做目的: 复杂开发过程.解耦(不调整依赖关系,无法解耦). 分层开发(横向拆分)和纵向拆分的区别在于,拆出多层,最终运行也是一个项目. 代

    2024年02月15日
    浏览(54)
  • 【Java Web】005 -- 请求响应 & 分层解耦(Postman、三层架构、IOC、DI、注解)

    目录 前言(SpringBoot程序请求响应流程) 一、请求 1、Postman(接口测试工具) ①、介绍 ②、安装 2、简单参数 ①、原始方式 ②、SpringBoot方法 ③、小结 3、实体参数 ①、简单实体对象 ②、复杂实体对象 ③、小结 4、数组集合参数 ①、数组 ②、集合 ③、小结 5、日期参数

    2024年02月11日
    浏览(51)
  • MQ消息队列详解以及MQ重复消费问题

    https://blog.csdn.net/qq_44240587/article/details/104630567 核心的就是:解耦、异步、削锋 现有ABCDE五个系统,最初的时候BCD三个系统都要调用A系统的接口获取数据,一切都很正常,但是突然,D系统说:我不要了,你不用给我传数据了,A系统无奈,只能修改代码,将调用D系统的代码删除

    2024年04月13日
    浏览(52)
  • 【消息中间件】详解mq消息积压

    作者简介 目录 1.产生原因 2.解决办法 2.1.事前处理机制 2.2.事中处理机制 2.3.事后处理机制 消息积压(Message Backlog)指的是在消息队列(MQ)系统中等待被处理的消息数量超过了正常的处理速度,导致消息在队列中积压堆积的情况。 消息积压的常见表现: 系统资源使用率上升

    2024年02月07日
    浏览(44)
  • mq常见问题:消息丢失、消息重复消费、消息保证顺序

    mq常见问题:消息丢失、消息重复消费、消息保证顺序 消息丢失问题 拿rabbitmq举例来说,出现消息丢失的场景如下图 从图中可以看到一共有以下三种可能出现消息丢失的情况: 1 生产者丢消息 生产者在将数据发送到MQ的时候,可能由于网络等原因造成消息投递失败 2MQ自身丢

    2024年02月09日
    浏览(55)
  • 【架构】领域驱动设计(DDD)的几种典型架构介绍

    我们生活中都听说了DDD,也了解了DDD,那么怎么将一个新项目从头开始按照DDD的过程进行划分与架构设计呢? 各种服务 IAAS:基础设施服务,Infrastructure-as-a-service PAAS:平台服务,Platform-as-a-service SAAS:软件服务,Software-as-a-service 从图中已经可以很容易看出架构的演进过程,

    2024年02月11日
    浏览(35)
  • DDD架构为什么应该首选六边形架构?

    分层架构的一个重要原则是:每层只能与位于其下方的层发生耦合。 分层架构分两种:一种是严格分层架构,规定某层只能与直接位于其下方的层发生耦合;另一种是松散分层架构,允许任意上方层与任意下方层发生耦合。 下图是一个典型的DDD传统分层架构。 以上分层架构

    2024年02月16日
    浏览(51)
  • MQ消息队列篇:三大MQ产品的必备面试种子题

    MQ(消息队列)是一种FIFO(先进先出)的数据结构,主要用于实现异步通信、削峰平谷和解耦等功能。它通过将生产者生成的消息发送到队列中,然后由消费者进行消费。这样,生产者和消费者之间就不存在直接的耦合关系。 其中,MQ的优势主要体现在以下几个方面: 异步通

    2024年02月14日
    浏览(40)
  • 消息队列MQ

    MQ的原理可以简单概括为生产者将消息发送到队列中,消费者从队列中获取消息进行处理。具体来说,MQ的原理包括以下几个方面: 生产者:生产者将消息发送到MQ服务器中,消息可以是文本、对象、文件等形式。生产者可以使用API或者其他工具将消息发送到MQ服务器,同时可

    2024年02月03日
    浏览(38)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包