【发布时间】：2021-04-02 06:14:21
【问题描述】：
在我的 Spring Boot 应用中，客户可以提交文件。每个客户的文件都通过每分钟运行的计划任务合并在一起。合并由调度程序执行的事实有许多缺点，例如编写端到端测试很困难，因为在测试中您必须等待调度程序运行才能检索合并结果。
因此，我想改用event-based approach，即
- 客户提交文件
- 已发布包含此客户 ID 的事件
- 合并服务侦听这些事件并在事件对象中为客户执行合并操作
这样做的好处是在有文件可合并后立即触发合并操作。
但是，这种方法存在许多问题，我希望得到一些帮助
并发
合并是一项相当昂贵的操作。最多可能需要 20 秒，具体取决于所涉及的文件数量。因此，合并必须异步发生，即不作为发布合并事件的同一线程的一部分。另外，我不想同时为同一个客户执行多个合并操作，以避免出现以下情况
- 客户 1 保存文件 2 触发文件 1 和文件 2 的合并操作 2
- 不久之后，客户 1 保存了文件 3，触发了文件 1、文件 2 和文件 3 的合并操作 3
- 合并操作3完成保存合并文件3
- Merge operation2 完成用 merge-file2 覆盖 merge-file3
为避免这种情况，我计划使用事件侦听器中的锁按顺序处理同一客户的合并操作，例如
@Component
public class MergeEventListener implements ApplicationListener<MergeEvent> {
private final ConcurrentMap<String, Lock> customerLocks = new ConcurrentHashMap<>();
@Override
public void onApplicationEvent(MergeEvent event) {
var customerId = event.getCustomerId();
var customerLock = customerLocks.computeIfAbsent(customerId, key -> new ReentrantLock());
customerLock.lock();
mergeFileForCustomer(customerId);
customerLock.unlock();
}
private void mergeFileForCustomer(String customerId) {
// implementation omitted
}
}
容错
如果应用程序在合并操作过程中关闭或在合并操作期间发生错误，我该如何恢复？
计划方法的优点之一是它包含隐式重试机制，因为每次运行时它都会查找具有未合并文件的客户。
总结
我怀疑我提出的解决方案可能会（严重）重新实施解决此类问题的现有技术，例如JMS。我建议的解决方案是可取的，还是应该改用 JMS 之类的东西？该应用程序托管在 Azure 上，因此我可以使用它提供的任何服务。
如果我的解决方案是可取的，我应该如何处理容错？
【问题讨论】：
-
看看 temporal.io，它是一个分布式编排平台，可以轻松支持此类用例。
-
如果您仍然使用 azure，您可以简单地使用主题来传播您的消费者订阅的事件。在生产者方面，您可以将事件作为正常实体持久化，因此在任何故障情况下，您都可以通过获取未标记为在您的消费者中处理的事件轻松恢复。显然，这并不能解决您的并发要求。但我确实看到你的方法存在问题，如果你打算异步执行并且顺序不是不重要的。确实很重要。
标签： java spring-boot events architecture scheduled-tasks