限时领|《AI 百问百答》专栏课+实体书(包邮)! 了解详情
写点什么

Event Sourcing 和 CQRS 落地(七):服务优化

  • 2019-07-08
  • 本文字数:4221 字

    阅读完需:约 14 分钟

Event Sourcing 和 CQRS落地(七):服务优化

在本系列的上一篇文章中,主要介绍了如何实现可靠消息,本文主要介绍如何进行服务优化。

服务优化

失败消息的补偿机制

由于消息存在发送失败的情况,比如 broker 临时下线或者不可用了,尽管这种情况很少,我们最好做一个机制可以定期或者手动检查,并且尝试自己发送,这里我们就来实现这个机制。

提供未发送的 event 查询

  1. CustomDomainEventEntryRepository中加入:


    /**     * 查找未发送的事件     *     * @param pageable     *     * @return     */    Page<CustomDomainEventEntry> findBySentFalse(Pageable pageable);        /**     * 查询未发送事件的数量     * @return     */    Long countBySentFalse();
复制代码


  1. 将未发送事件的数量集成到 actuator,让我们可以事实看到失败消息的数量:


@Component@AllArgsConstructorpublic class EventHealthContributor implements InfoContributor {
private final CustomDomainEventEntryRepository customDomainEventEntryRepository;
@Override public void contribute(Info.Builder builder) { Long count = customDomainEventEntryRepository.countBySentFalse();
builder.withDetail("failedMessage", count); }}
复制代码


打开 http://localhost:8080/actuator/info 应该就可以看到我们的失败消息数量。

定期检查并且自动发送

建立对应的 service 和 controller:


@Service@Slf4j@AllArgsConstructorpublic class ScheduleService {
private final CustomDomainEventEntryRepository customDomainEventEntryRepository; private final ContractEventPublisher contractEventPublisher;
@Scheduled(cron = "0 0 12 * * ?") @SchedulerLock(name = "failedMessageDiscoveryTask") public void failedMessageDiscovery() {
Integer page = 0; PageRequest request = PageRequest.of(page, 1000);
Page<CustomDomainEventEntry> results = customDomainEventEntryRepository.findBySentFalse(request); log.warn(MessageFormat.format("发现 [{0}] 条失败消息,尝试重新发送", results.getTotalElements())); sendFailedMessage(results.getContent()); while (results.hasNext()) { request = PageRequest.of(page + 1, 1000); results = customDomainEventEntryRepository.findBySentFalse(request); sendFailedMessage(results.getContent()); } log.info("所有失败消息尝试发送完毕"); } private void sendFailedMessage(Collection<CustomDomainEventEntry> failedEvents) {
failedEvents.forEach(e -> { contractEventPublisher.sendEvent(e); e.setSent(true); customDomainEventEntryRepository.save(e); }); }}
复制代码


有些时候我们可能需要自己触发一个修复操作,可以把这个写成 API:


/** * 用于修复 view 视图和 aggregate 的不一致性,以及未发送消息的重试 */@Slf4j@AllArgsConstructor@RestController@RequestMapping("/repair")public class DataRepairController {
private final ScheduleService scheduleService; private static final String SECRET = "e248b98418db4cdcb069e8a1c08f6bb7";
@GetMapping("/message") @Async public void repairMessage(@RequestParam("secret") String secret) { if (!StringUtils.equals(secret, SECRET)) { return; }
scheduleService.failedMessageDiscovery(); }}

复制代码


pom 中加入:



<dependency> <groupId>net.javacrumbs.shedlock</groupId> <artifactId>shedlock-spring</artifactId> <version>1.0.0</version> </dependency> <dependency> <groupId>net.javacrumbs.shedlock</groupId> <artifactId>shedlock-provider-jdbc-template</artifactId> <version>1.0.0</version> </dependency>
复制代码


数据库中执行脚本:


CREATE TABLE shedlock(    name VARCHAR(64),     lock_until TIMESTAMP(3) NULL,     locked_at TIMESTAMP(3) NULL,     locked_by  VARCHAR(255),     PRIMARY KEY (name)) 
复制代码


使用SchedulerLock定义了每晚 12 点开始检查并自动发送,由于 SchedulerLock 在集群下的问题,这里使用了 shedlock 加锁,使得只有一个实例会执行该代码。这样就完成了自动修复与手动修复的接口暴露。

view 和 aggregate 之间的不一致补偿

在并发比较高的时候,可能会出现同时 update 一条记录的情况,这个时候需要上锁,JPA 可以自动创建并管理乐观锁,乐观锁会在 update 同一条记录的时候直接返回一个错误,我们只需要在 entity 上加上 version 字段即可:


@Entity@Getter@Setter@AllArgsConstructor@NoArgsConstructorpublic class ContractView implements ContractInterface {
@Id @Column(length = 64) private Long id;
private String name;
private String partyA;
private String partyB;
private boolean deleted = false;
private String industryName;
private long sequenceNumber;
@Version private Long version;}
复制代码


某些情况下我们可能还需要手动干预 aggregateview 之间的一致性问题,比如在线上运行中莫名其妙有一个 id 为 xxx 的数据出现了不一致的情况,可能是在某些极端情况下消费端出现了逻辑错误导致了数据错误,但是又很难找到这批数据的特征。这个时候其实可以写一个 DataRepair 的 API,专门重新生成 view,或者说范围性的重新生成 view(因为真正在生产环境全部重新生成 view,必定是很耗时的一件事情),所以如果之前的 view 层代码是通过读取 aggregate 状态之后做映射的话,这里就会方便很多。比如:


@Service@AllArgsConstructorpublic class ContractViewService {
private final QueryGateway queryGateway; private final ContractViewRepository contractViewRepository;
public void updateViewFromAggregateById(Long aggregateIdentifier, Instant time) {
QueryContractCommand command = new QueryContractCommand(aggregateIdentifier, time); ContractAggregate aggregate = queryGateway.query(command, ContractAggregate.class).join(); ContractView view = contractViewRepository.findById(aggregateIdentifier).orElse(new ContractView());
ContractAggregateViewMapper.mapAggregateToView(aggregate, view); contractViewRepository.save(view); }}
@Component@AllArgsConstructor@Transactionalpublic class ContractEventHandler {
private final ContractViewService contractViewService;
@StreamEventHandler(types = ChannelDefinition.CONTRACTS_INPUT) public void handle(ContractCreatedEvent event, DomainEvent<ContractCreatedEvent, HashMap> domainEvent) { contractViewService.updateViewFromAggregateById(event.getIdentifier(), domainEvent.getTimestamp()); }}
@Slf4j@AllArgsConstructor@RestController@RequestMapping("/repair")public class DataRepairController {
private final ScheduleService scheduleService;
private final ContractViewService contractViewService;
private static final String SECRET = "e248b98418db4cdcb069e8a1c08f6bb7";
@GetMapping("/message") @Async public void repairMessage(@RequestParam("secret") String secret) { if (!StringUtils.equals(secret, SECRET)) { return; }
scheduleService.failedMessageDiscovery(); }
@PostMapping("/aggregate") @Async public void repairAggregate(@RequestParam("secret") String secret, Long aggregateIdentifier) { if (!StringUtils.equals(secret, SECRET)) { return; } contractViewService.updateViewFromAggregateById(aggregateIdentifier, Instant.now()); }}
复制代码

事件失败之后的补偿

updateContractView的时候可能会出现各种异常,由于 view 是消费消息处理的,所以重试机制就在 SCSrabbitmq 这里了,默认是重试三次。

实现分布式 CommandBus

为什么需要分布式

前面我们也提到过,为了防止资源争夺等问题的出现,最好尽可能的保证同一 aggregate 的内容让同一个 service 去处理。这个时候就需要分布式 command bus 了,我们知道 Spring Cloud 都是以 http 通讯的,这种一般请求的分布不受我们控制,好在 Axon 框架为我们提供了和 Spring Cloud 融合的功能,下面就看看具体怎么实现。

Spring Cloud Connector 实现

Spring Cloud Connector 实际上就是在每个节点用 ServiceInstance.Metadata 记录了自己的 routing 规则来让别的节点知道如何去做 routing ,但是在某些服务发现的实现下 ServiceInstance.Metadata 是不可编辑的,这个时候就会在生成一个 API 来返回策略(理论上效率应该低很多,毕竟要走下请求)。Axon 对配置支持的也比较好:


    <dependency>      <groupId>org.axonframework.extensions.springcloud</groupId>      <artifactId>axon-springcloud</artifactId>      <version>4.1</version>    </dependency>
复制代码


然后配置文件中将分布式 command 打开就好了:


axon:  serializer:    general: jackson  distributed:    enabled: true    spring-cloud:      fallback-to-http-get: true      fallback-url: /axon-routing
复制代码


完整的例子 - branch session8


作者介绍:


周国勇,目前就职于杭州匠人网络创业,致力于楼宇资产管理的 SaaS 化,负责后端业务架构设计、项目管理,喜欢对业务模型的分析,热衷新技术的探索和实践,经常在踩坑的路上越走越远。


相关文章:


《Event Sourcing 和 CQRS 落地(一):UID-Generator 实现》


《Event Sourcing 和 CQRS 落地(二):Event-Sourcing 实现》


《Event Sourcing 和 CQRS 落地(三):CQRS 实现》


《Event Sourcing 和 CQRS 落地(四):深入使用 -Axon》


《Event Sourcing 和 CQRS 落地(五):Spring-Cloud-Stream 优化》


《Event Sourcing 和 CQRS 落地(六):实现可靠消息》


2019-07-08 10:308037

评论

发布
暂无评论
发现更多内容

流动的智慧:开创集成资产管理新局面 ——华为云ROMA Connect资产中心

轶天下事

雅菲奥朗 FinOps 认证培训:开启企业云财务管理转型之路

雅菲奥朗

云计算 FinOps FinOps 认证 FinOps 考试 FinOps 培训

MySQL亿级数据平滑迁移实战

vivo互联网技术

MySQL spring mybatis 数据迁移

从零开始探索人工智能:Python与PyTorch的奇妙之旅

代码忍者

反欺诈技术发展的现状与未来预测

芯盾时代

金融 反欺诈

LED双色屏的特性与受欢迎的原因

Dylan

升级 LED display LED显示屏 色彩 颜色

巧用PDF转Markdown插件,在扣子(Coze)手搓一个有趣好玩的AI Bot

合合技术团队

人工智能 PDF 科技

华为云全域Serverless技术创新:全球首创通用Serverless平台被ACM SIGCOMM录用

轶天下事

AICon 全球人工智能开发与应用大会参会有感

三掌柜

AICon

AI 应用实战营 - 毕业总结

德拉古蒂洛维奇

【第3期】2024 搜索客 Meetup | Elasticsearch 的代码结构和写入查询流程的解读 - 下篇

极限实验室

「模型量化技术」可视化指南:A Visual Guide to Quantization

Baihai IDP

AI 模型量化 白海科技 LLMs 企业号 8 月 PK 榜

Photoshop 2024最新版详细安装教程 附mac版和win版安装包

Rose

SSL证书验证失败的原因

德迅云安全杨德俊

引领未来商业潮流:AI赋能商品管理软件的革新应用与趋势

第七在线

Microsoft 365 有什么用?microsoft 365永久激活密钥

Rose

华为云Serverless可观测性解决方案打造高效、可靠的云原生应用

轶天下事

汇聚行业实践,树立应用典范——《Serverless应用实践案例集》重磅发布

轶天下事

操作系统笔记 day4

万里无云万里天

操作系统

极兔速递2024上半年净利润转正 中国市场首次盈利

财见

淘宝商品销量数据接口:获取与利用全攻略

tbapi

淘宝商品销量接口 淘宝商品销量API 淘宝月销量接口 淘宝商品销量数据采集

【ASPLOS2024】RECom:通过编译器技术加速推荐模型推理,论文中选并获得荣誉奖项!

阿里云大数据AI技术

人工智能 阿里云 论文 ASPLOS

推荐6款好用的低代码开发平台

秃头小帅oi

游戏行业大模型陪练实战:3分钟快速配置对练剧本

中关村科金

人工智能 大模型 智能陪练

Event Sourcing 和 CQRS落地(七):服务优化_文化 & 方法_周国勇_InfoQ精选文章