写点什么

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:307925

评论

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

区域链和数字媒体艺术

dappweb

区块链 ao

众所周知,配置即代码≠基础设置即代码

禅道项目管理

项目管理 DevOps 自动化测试 配置即代码 基础设置即代码

元宇宙与教育

dappweb

ao 元宇宙

华为云空间端云一致体验:重要数据随时随地在手边

最新动态

接入效率提升90%,华为云空间助力开发者高效实现应用数据同步

最新动态

华为云GeminiDB赋能迷你创想:轻松应对亿级月活玩家

最新动态

蓝易云 - docker教程:nginx.conf配置文件示例

百度搜索:蓝易云

nginx Docker 服务器 云服务器 高防服务器

有哪些元宇宙的已知成功案例可以借鉴

dappweb

ao 元宇宙

数字指纹在区块链上如何生成和验证

dappweb

区块链 ao

Java实现零拷贝

极客罗杰

蓝易云 - nginx教程:upstream的两种配置方式

百度搜索:蓝易云

nginx HTTP 云服务器 upstream FastCGI

在版权保护中区块链如何提高效率

dappweb

区块链 ao

2024年全球零信任发展趋势

芯盾时代

网络安全 iam 统一身份认证 零信任

使用 Swift 6 语言模式构建 Swift 包

Swift社区

swift 企业号2024年6月PK榜

蓝易云 - nginx教程:配置项add_header Access-Control-Allow-Origin *的含义

百度搜索:蓝易云

nginx Linux 运维 CORS 云服务器

mac电脑好用的防火墙:Radio Silence for mac 激活版

你的猪会飞吗

mac软件下载 Mac电脑软件

华为大咖说 | 从《庖丁解牛》谈项目经理的三重境界

华为云PaaS服务小智

项目管理 开发 华为云

小浣熊家族 X InfoQ 写作社区有奖征文大赛获奖名单公示

InfoQ写作社区官方

热门活动

作为独立创作者的一点感想

zhumingwu

【测试中的人工智能】自学活动一:理论知识学习

QE_LAB

人工智能 AI 测试

蓝易云 - nginx教程:proxy_pass指令后边有没有“/”的区别

百度搜索:蓝易云

nginx 运维 前端 云服务器 proxy_pass

探索未来网络安全的新篇章:火绒安全与 OpenVINO™ 的协同创新

Geek_2d6073

基于STM32+华为云IOT设计的智能窗帘控制系统

DS小龙哥

6 月 优质更文活动

聊聊元宇宙怎么样

dappweb

ao 元宇宙

元宇宙在元科学习中有哪些独特优势

dappweb

ao 元宇宙

小鲤AI志愿填报助手:一键探索最适合你的大学专业

Geek_1ef48b

DataWorks重磅推出全新资源组2.0,实现低成本灵活付费和动态平滑扩缩容

阿里云大数据AI技术

sql 大数据 阿里云 Serverless Dataworks

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