Spring Integration 中的并行处理答案

【问题标题】：Parallel processing in Spring IntegrationSpring Integration 中的并行处理
【发布时间】：2020-10-14 10:09:09
【问题描述】：

我需要按顺序处理具有相同参考 id 但不同参考 id 的交易，因为它们是独立的。交易可以有各种事件，因此无法并行处理同一交易，因为它们会修改同一交易实例，从而导致意外更新。

我正在使用 Spring 集成框架来处理交易，但任务执行器正在并行处理所有交易，即使其中一些交易共享相同的参考 ID 但不同的事件。所以请分享代码 sn-p 来处理这种情况，因为我找不到任何解决方案。

谢谢

【问题讨论】：

标签： java spring spring-integration

【解决方案1】：

我不确定这是否会奏效，但我认为会奏效。在您处理交易的方法中，同步 id 的实习版本。（我假设 id 是一个字符串。）

{
   String a = "some string";  // this is already intern'ed.
   String b = new String("some string");  // this is not.  You might have gotten the value from user, or whatever

   boolean b1 = a == b;  // false
   boolean b2 = a.equals(b);  // true
   boolean b3 = a.intern() == b.intern();  // Always true if b2 was true
}

所以关键是，如果你实习了几个字符串，如果其中两个是相同的值，你实际上会得到相同的字符串。因此，在您交给执行者的可运行文件中，

public void run() {
    synchronize(trade.getId().intern()) {
        // process trade
    }
}

我不确定它是否会起作用的唯一原因是我不确定您是否可以在 String 的内部版本上进行同步，因为这就像一个静态元素。我真的不明白为什么不这样做，但你应该建立一个实验来确定。

【讨论】：

不过，这对订单没有帮助。我认为您需要使用 Map<String, BlockingQueue<Request>> 之类的东西并从单个线程插入到地图中并在下游异步处理。该用例的框架中没有内置任何内容。您可以使用 Kafka 之类的东西作为中间媒介，其中相同键的所有记录都可以发送到同一个分区。
我同意。我的解决方案可能会以与到达顺序不同的顺序处理单个 ID 的交易。如果 ID 的数量相对较少，比如少于几千，那么 BlockingQueues 的 Map 就可以工作。但是，如果这些是股票交易，那么您将在必须创建的数万或数十万个 BlockingQueue 上浪费太多内存。

【解决方案2】：

看看LockRegistry能不能帮到你！所以，你会得到一个 Lock 作为“参考 ID”，并且不要让其他线程用你的特定 Trade 做这些事情，直到你在一个线程中完成。

您可以使用WhileLockedProcessor 以获得更简洁的代码。

【讨论】：