Event Sourcing 和 CQRS落地（七）：服务优化_文化 & 方法_周国勇

2025 AI基础设施风向标，不看必后悔！#AI基础设施峰会了解详情 



 写点什么

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

服务优化

失败消息的补偿机制

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

提供未发送的 event 查询

在CustomDomainEventEntryRepository中加入：

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

复制代码

将未发送事件的数量集成到 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;}

复制代码

某些情况下我们可能还需要手动干预 aggregate 和 view 之间的一致性问题，比如在线上运行中莫名其妙有一个 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 是消费消息处理的，所以重试机制就在 SCS 和 rabbitmq 这里了，默认是重试三次。

实现分布式 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 落地（二）：Event-Sourcing 实现》

《Event Sourcing 和 CQRS 落地（三）：CQRS 实现》

《Event Sourcing 和 CQRS 落地（四）：深入使用 -Axon》

《Event Sourcing 和 CQRS 落地（五）：Spring-Cloud-Stream 优化》

《Event Sourcing 和 CQRS 落地（六）：实现可靠消息》