本文的使用宗旨在于通过简单干净实践的方式教会读者,使用 Docker 配置 RocketMQ 并在基于 DDD 分层结构的构M该放 SpringBoot 工程中使用 RocketMQ 技术。因为大部分 MQ 的那层发送都是基于特定业务场景的,所以本章节也是使用基于 《MyBatis 使用教程和插件开发》 章节的扩展。
本章也会包括关于 MQ 消息的构M该放发送和接收应该处于 DDD 的哪一层的实践讲解和使用。
本文涉及的那层工程:
首先我们要知道,使用MQ 消息的构M该放作用是用于;解耦过长的业务流程和应对流量冲击的消峰。如;用户下单支付完成后,那层拿到支付消息推动后续的使用发货流程。也可以是我们基于 《MyBatis 使用教程和插件开发》 中的案例场景,给雇员提升级别和薪资的时候,也发送一条MQ消息,用于发送邮件通知给用户。
图片
因为我们本章所讲解的内容是把 RocketMQ 放入 DDD 架构中进行使用,那么也就引申出领域事件定义。所以我们先来了解下,什么是领域事件。
领域事件,可以说是解耦微服务设计的关键。领域事件也是领域模型中非常重要的一部分内容,用于标示当前领域模型中发生的事件行为。一个领域事件会推进业务流程的进一步操作,在实现业务解耦的同时,也推动了整个业务的闭环。
图片
本案例涉及了数据库和RocketMQ的使用,都已经在工程中提供了安装脚本,可以按需执行。
图片
这里主要介绍 RocketMQ 的安装;
文件: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"
# 集群名称brokerClusterName = DefaultCluster# BROKER 名称brokerName = broker-a# 0 表示 Master, > 0 表示 SlavebrokerId = 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
server.address=0.0.0.0server.port=9009
RocketMQ 此镜像,会在安装后在控制台打印登录账号信息,你可以查看使用。
图片
图片
登录:http://localhost:9009/
图片
图片
图片
引入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
源码:cn.bugstack.xfg.dev.tech.domain.salary.event.SalaryAdjustEvent
图片
@EqualsAndHashCode(callSuper = true)@Datapublic 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; }}
源码:cn.bugstack.xfg.dev.tech.infrastructure.event.EventPublisher
图片
@Component@Slf4jpublic 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发送失败后,会需要任务补偿 } }}
源码:cn.bugstack.xfg.dev.tech.infrastructure.repository.SalaryAdjustRepository
@Resourceprivate 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点要注意;
源码: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); }}
@Slf4j@RunWith(SpringRunner.class)@SpringBootTestpublic class RocketMQTest { @Setter(onMethod_ = @Autowired) private RocketMQTemplate rocketmqTemplate; @Test public void test() throws InterruptedException { while (true) { rocketmqTemplate.convertAndSend("xfg-mq", "我是测试消息"); Thread.sleep(3000); } }}
@Testpublic 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: 10090897767600423-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"}
(责任编辑:综合)
文投控股(600715.SH):北京文创定增基金已减持17.89万股 占公司总股份的0.0096%
《如茵碧草—萋萋护路人》DEMO现可游玩,将于今年晚些时候登陆PC平台