Apache的骆驼：春天组播+处理

Question

亲爱的社区。

我试图添加多播聚合策略在Spring配置XML以不同的方式，这是我在Apache论坛，计算器和其他很多从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> 

每个直接由具有后调用的方法，这是正常工作（并返回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] 

所以我希望身体与分裂两个String JSON逗号（就像在聚合策略类中一样），但是我只有body，它会发送到其他路由。

任何想法？

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; 
}