java - Apache Camel : Spring multicast + processing

Question

亲爱的社区。

我尝试以不同的方式在 spring config xml 中添加多播聚合策略，这是我在 apache 论坛、stackoverflow 和许多其他从 java dsl 转换而来的地方找到的。

我错过了一些事情，你能想象一下，如何配置它才能针对下面描述的情况获得正确的行为。

我有一些名称列表(所有 bean 都已正确定义)，

<to id="_to85" uri="bean:otrsCIApi?method=ConfigItemNameList()"/>

然后我需要在两个不同的系统中进行相互拆分的后调用:

<split streaming="true">
    <simple>${body}</simple>
    <!-- iterate in JsonArray -->
    <choice>
        <when>
            <simple>${body} != ''</simple>
            <multicast stopOnException="true" parallelProcessing="true" strategyRef="OtrsZabbixCiAggregationStrategy" >
                <to uri="direct:zabbix_multicast"/>
                <to uri="direct:otrs_multicast"/>
                <to uri="log:jms_apachemq3"/>
            </multicast>
            <to id="_to92" uri="log:jms_apachemq4"/>
        </when>
        <when>
            <simple>${body} == ''</simple>

        </when>
    </choice>
</split>

每个直接都包含 post 调用方法，这些方法可以正确工作(并返回 JSONObject.toString()):

<route>
    <from uri="direct:zabbix_multicast"/>
    <to uri="bean:zabbixApi?method=getHostBody(body)"/>
    <to uri="log:multicast_zab1"/>
</route>

<route>
    <from uri="direct:otrs_multicast"/>
    <to uri="bean:otrsCIApi?method=searchCI(body)"/>
    <to uri="log:multicast_otrs1"/>
    <to uri="bean:otrsCIApi?method=getCIBodyByID(body)"/>
    <to uri="log:multicast_otrs2"/>
</route>

聚合策略非常简单，并且应该有效:

import org.apache.camel.Exchange;
import org.apache.camel.processor.aggregate.AggregationStrategy;
public class OtrsZabbixCiAggregationStrategy implements AggregationStrategy{
    public Exchange aggregate(Exchange exchange1, Exchange exchange2) {
            String body1 = exchange1.getIn().getBody().toString();
            String body2 = exchange2.getIn().getBody().toString();
            String merged = body1 + "," + body2;
            exchange1.getIn().setBody(merged);
            return exchange1;
    }
}

然后我得到了非常奇怪的输出，就像没有开始聚合一样:

[fc8135) thread #39 - Multicast] jms_apachemq3                  INFO  Exchange[ExchangePattern: InOnly, BodyType: String, Body: Zabbix server]
[fc8135) thread #37 - Multicast] multicast_zab1                 INFO  Exchange[ExchangePattern: InOnly, BodyType: String, Body: [{"available":"1","description":"","disable_until":"0","error":"","errors_from":"0","flags":"0","host":"Zabbix server","hostid":"10084","ipmi_authtype":"-1","ipmi_available":"0","ipmi_disable_until":"0","ipmi_error":"","ipmi_errors_from":"0","ipmi_password":"","ipmi_privilege":"2","ipmi_username":"","jmx_available":"0","jmx_disable_until":"0","jmx_error":"","jmx_errors_from":"0","lastaccess":"0","maintenance_from":"0","maintenance_status":"0","maintenance_type":"0","maintenanceid":"0","name":"Zabbix server","proxy_hostid":"0","snmp_available":"0","snmp_disable_until":"0","snmp_error":"","snmp_errors_from":"0","status":"0","templateid":"0","tls_accept":"1","tls_connect":"1","tls_issuer":"","tls_psk":"","tls_psk_identity":"","tls_subject":""}]]
[fc8135) thread #38 - Multicast] multicast_otrs1                INFO  Exchange[ExchangePattern: InOnly, BodyType: com.alibaba.fastjson.JSONArray, Body: 2]
[fc8135) thread #38 - Multicast] multicast_otrs2                INFO  Exchange[ExchangePattern: InOnly, BodyType: String, Body: {"ConfigItem":[{"CurInciState":"Operational","ConfigItemID":"2","InciStateType":"operational","CurInciStateType":"operational","Number":"1022000002","CreateBy":"2","LastVersionID":"2","DeplState":"Production","CurDeplState":"Production","CreateTime":"2016-11-01 13:47:44","DefinitionID":"1","VersionID":"2","DeplStateType":"productive","CIXMLData":{"SerialNumber":"","Ram":"","WarrantyExpirationDate":"2016-11-01","Vendor":"","CPU":"","Model":"","Owner":"","Type":"","HardDisk":{"HardDisk":"","Capacity":""},"GraphicAdapter":"","FQDN":"","OperatingSystem":"","Description":""},"Class":"Computer","InciState":"Operational","CurDeplStateType":"productive","Name":"Zabbix server"}]}]
[hread #32 - JmsConsumer[queue]] jms_apachemq4                  INFO  Exchange[ExchangePattern: InOnly, BodyType: String, Body: Zabbix server]

所以我期望主体有两个用逗号分隔的字符串 JSON(就像在聚合策略类中一样)，但我只有主体，它发送到其他路由。

有什么想法吗？

Answer 1

我相信您的 AggregationStrategy 中有一个 NullPointerException。第一次调用时，exchange1 为 NULL，exchange2 是第一次运行时的交换对象。由于参数 stopOnException="true"，异常被忽略并停止多播。因此，您之后就有了原始的交换对象。

您只需要进行需求检查，它就应该可以工作:

public Exchange aggregate(Exchange exchange1, Exchange exchange2) {
        if (exchange1==null){
           return exchange2;
        }
        String body1 = exchange1.getIn().getBody().toString();
        String body2 = exchange2.getIn().getBody().toString();
        String merged = body1 + "," + body2;
        exchange1.getIn().setBody(merged);
        return exchange1;
}