Event Sourcing 和 CQRS 落地（七）：服务优化

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

@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);    }}

@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);        });    }}

/** * 用于修复 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();    }}

    <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)) 

@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;}

@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());    }}

    <dependency>      <groupId>org.axonframework.extensions.springcloud</groupId>      <artifactId>axon-springcloud</artifactId>      <version>4.1</version>    </dependency>

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