blocks|key|1647179|text|您可以在Apache中使用缓存组件来实现这一点。我认为有EHCache组件。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1647180|实质上：|1647181|您将收到一条带有给定deviceId的消息，例如deviceId1。|ordered-list-item|1647182|您可以在缓存中查找是否收到了deviceId1的哪些消息。|1647183|只要您还没有收到所有这三个，就可以将当前的系统/消息添加到缓存中。|1647184|一旦所有消息都在那里，您将处理并清除缓存。|1647185|然后，您可以偏离路线，将每个传入的消息路由到特定的基于deviceId的队列中，以便进行临时存储。这可以是JMS、ActiveMQ或类似的东西。|1647186|entityMap^0|0|0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|Q|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|R|8|@]|9|@]|A|$]]|$1|D|3|E|5|F|7|S|8|@]|9|@]|A|$]]|$1|G|3|H|5|F|7|T|8|@]|9|@]|A|$]]|$1|I|3|J|5|F|7|U|8|@]|9|@]|A|$]]|$1|K|3|L|5|F|7|V|8|@]|9|@]|A|$]]|$1|M|3|N|5|6|7|W|8|@]|9|@]|A|$]]|$1|O|3|-4|5|6|7|X|8|@]|9|@]|A|$]]]|P|$]]

You can do this in Apache Camel using a caching component. I think there is the EHCache component.

Essentially:

<ol>
<li>You receive a message with a given deviceId say deviceId1.</li>
<li>You look up in your cache to see which messages have been received for deviceId1.</li>
<li>As long as you have not received all three you add the current system/message to the cache.</li>
<li>Once all messages are there you process and clear the cache.</li>
</ol>

You could then off course route each incoming message to a specific deviceId based queue for temporary storage. This can be JMS, ActiveMQ or something similar.

blocks|key|10647|text|为这类任务提供了组件--在收集完整个组之前不要发出。它的名字叫集料器。您的deviceId绝对是一个correlationKey。releaseStrategy实际上可能是基于系统的数量--在进入下一步之前等待了多少deviceId1消息。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|10648|entityMap|0|LINK|mutability|MUTABLE|url|https://docs.spring.io/spring-integration/docs/4.3.12.RELEASE/reference/html/messaging-routing-chapter.html#aggregator^0|11|8|1E|E|1T|F|30|9|V|3|0|0^^$0|@$1|2|3|4|5|6|7|N|8|@$9|O|A|P|B|C]|$9|Q|A|R|B|C]|$9|S|A|T|B|C]|$9|U|A|V|B|C]]|D|@$9|W|A|X|1|Y]]|E|$]]|$1|F|3|-4|5|6|7|Z|8|@]|D|@]|E|$]]]|G|$H|$5|I|J|K|E|$L|M]]]]

Spring Integration provides component for exactly this kind of tasks - do not emit until the whole group is collected. And it's name an <a href="https://docs.spring.io/spring-integration/docs/4.3.12.RELEASE/reference/html/messaging-routing-chapter.html#aggregator" rel="nofollow noreferrer">Aggregator</a>. Your <code>deviceId</code> is definitely a <code>correlationKey</code>. The <code>releaseStrategy</code> really may be based on the number of systems - how much <code>deviceId1</code> messages you are waiting before proceed to the next step.

blocks|key|10736|text|使用聚合器(+@Artem+Bilan+)的类似解决方案也可以用自定义AggregationStrategy在Camel中实现，并通过使用Exchange.AGGREGATION_COMPLETE_CURRENT_GROUP属性来控制聚合器的完成。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|10737|以下可能是一个很好的起点。(您可以在这里找到带有测试的示例项目。)|10738|路由：|10739|from("direct:start")
++++.log(LoggingLevel.INFO,+"Received+${headers.system}${headers.deviceId}")
++++.aggregate(header("deviceId"),+new+SignalAggregationStrategy(3))
++++.log(LoggingLevel.INFO,+"Signaled+body:+${body}")
++++.to("direct:result");|code-block|syntax|javascript|10740|SignalAggregationStrategy.java|10741|public+class+SignalAggregationStrategy+extends+GroupedExchangeAggregationStrategy+implements+Predicate+{

++++private+int+numberOfSystems;

++++public+SignalAggregationStrategy(int+numberOfSystems)+{
++++++++this.numberOfSystems+=+numberOfSystems;
++++}

++++@Override
++++public+Exchange+aggregate(Exchange+oldExchange,+Exchange+newExchange)+{
++++++++Exchange+exchange+=+super.aggregate(oldExchange,+newExchange);

++++++++List<Exchange>+aggregatedExchanges+=+exchange.getProperty("CamelGroupedExchange",+List.class);

++++++++//+Complete+aggregation+if+we+have+"numberOfSystems"+(currently+3)+different+messages+(where+"system"+headers+are+different)
++++++++//+https://github.com/apache/camel/blob/master/camel-core/src/main/docs/eips/aggregate-eip.adoc#completing-current-group-decided-from-the-aggregationstrategy
++++++++if+(numberOfSystems+==+aggregatedExchanges.stream().map(e+->+e.getIn().getHeader("system",+String.class)).distinct().count())+{
++++++++++++exchange.setProperty(Exchange.AGGREGATION_COMPLETE_CURRENT_GROUP,+true);
++++++++}

++++++++return+exchange;
++++}

++++@Override
++++public+boolean+matches(Exchange+exchange)+{
++++++++//+make+it+infinite+(4th+bullet+point+@+https://github.com/apache/camel/blob/master/camel-core/src/main/docs/eips/aggregate-eip.adoc#about-completion)
++++++++return+false;
++++}
}|10742|希望能帮上忙！|10743|entityMap|0|LINK|mutability|MUTABLE|url|https://github.com/mgyongyosi/camel-cookbook/tree/master/custom-aggregator^0|Z|J|1X|17|0|E|I|0|0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|12|8|@$9|13|A|14|B|C]|$9|15|A|16|B|C]]|D|@]|E|$]]|$1|F|3|G|5|6|7|17|8|@]|D|@$9|18|A|19|1|1A]]|E|$]]|$1|H|3|I|5|6|7|1B|8|@]|D|@]|E|$]]|$1|J|3|K|5|L|7|1C|8|@]|D|@]|E|$M|N]]|$1|O|3|P|5|6|7|1D|8|@]|D|@]|E|$]]|$1|Q|3|R|5|L|7|1E|8|@]|D|@]|E|$M|N]]|$1|S|3|T|5|6|7|1F|8|@]|D|@]|E|$]]|$1|U|3|-4|5|6|7|1G|8|@]|D|@]|E|$]]]|V|$W|$5|X|Y|Z|E|$10|11]]]]

A similar solution with the Aggregator (what @Artem Bilan mentioned) can also be implemented in Camel with a custom <code>AggregationStrategy</code> and with controlling the Aggregator completion by using the <code>Exchange.AGGREGATION_COMPLETE_CURRENT_GROUP</code> property.

The following might be a good starting point. (<a href="https://github.com/mgyongyosi/camel-cookbook/tree/master/custom-aggregator" rel="nofollow noreferrer">You can find the sample project with tests here</a>)

Route:

<pre><code>from("direct:start")
 .log(LoggingLevel.INFO, "Received ${headers.system}${headers.deviceId}")
 .aggregate(header("deviceId"), new SignalAggregationStrategy(3))
 .log(LoggingLevel.INFO, "Signaled body: ${body}")
 .to("direct:result");
</code></pre>

SignalAggregationStrategy.java

<pre><code>public class SignalAggregationStrategy extends GroupedExchangeAggregationStrategy implements Predicate {

 private int numberOfSystems;

 public SignalAggregationStrategy(int numberOfSystems) {
 this.numberOfSystems = numberOfSystems;
 }

 @Override
 public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
 Exchange exchange = super.aggregate(oldExchange, newExchange);

 List&lt;Exchange&gt; aggregatedExchanges = exchange.getProperty("CamelGroupedExchange", List.class);

 // Complete aggregation if we have "numberOfSystems" (currently 3) different messages (where "system" headers are different)
 // https://github.com/apache/camel/blob/master/camel-core/src/main/docs/eips/aggregate-eip.adoc#completing-current-group-decided-from-the-aggregationstrategy
 if (numberOfSystems == aggregatedExchanges.stream().map(e -&gt; e.getIn().getHeader("system", String.class)).distinct().count()) {
 exchange.setProperty(Exchange.AGGREGATION_COMPLETE_CURRENT_GROUP, true);
 }

 return exchange;
 }

 @Override
 public boolean matches(Exchange exchange) {
 // make it infinite (4th bullet point @ https://github.com/apache/camel/blob/master/camel-core/src/main/docs/eips/aggregate-eip.adoc#about-completion)
 return false;
 }
}
</code></pre>

Hope it helps!

I am building a system that will receive messages via a Message broker (Currently, JMS) from different systems. All the messages from all the senders systems have a deviceId and there is no order in the reception of the message.
For instance, system A can send a message with deviceId=1 and system b be can send a message with deviceId=2. 

My goal is not to start processing of the messages concerning the same deviceId unless I got all the message from all the senders with the same deviceId.

For example, if I have 3 systems A, B and C sending messages to my system : 

<pre><code>System A sends messageA1 with deviceId=1
System B sends messageB1 with deviceId=1
System C sends messageC1 with deviceId=3
System C sends messageC2 with deviceId=1 &lt;--- here I should start processing of messageA1, messageB1 and messageC2 because they are having the same deviceID 1.
</code></pre>

Should this problem be resolved by using some sync mechanism in my system , by the message broker or an integration framework like spring-integration/apache camel ?

Integration pattern : how to sync processing message received from multiple systems

翻译质量差，导致语言生硬或混乱。

没有提供实际的解决方法或示例。

解答不清晰，无法理解或解决问题。

页面排版不美观，阅读体验差。

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

我正在构建一个系统，它将通过Message (目前是JMS)从不同的系统接收消息。来自所有发件人系统的所有消息都有一个deviceId，并且在接收消息时没有顺序。例如，系统A可以用deviceId=1发送消息，系统b可以用deviceId=2发送消息。我的目标是不开始处理有关同一个deviceId的消息，除非我从所有具...

问集成模式:如何同步从多个系统接收的处理消息
EN

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问集成模式:如何同步从多个系统接收的处理消息EN