注册Apache Camel处理器的侦听器

Question

我尝试将在Apache Camel处理器中处理的数据推送到侦听器类。在处理器类实例中，我尝试在Camel上下文的实例化期间注册侦听器，但不知何故失败了。也许我在这里根本错误，这是不可能的。如果是这种情况，最好你告诉我。

我有一个Apache Camel路由，从ActiveMQ服务器获取JSON消息，并将这些JSON推送到一个自定义处理器类，该类由Camel-Spring XML定义：

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans.xsd
    http://camel.apache.org/schema/spring
    http://camel.apache.org/schema/spring/camel-spring.xsd">

    <bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="url to ActiveMQ" />
        <property name="clientID" value="clientID" />
        <property name="userName" value="theUser" />
        <property name="password" value="thePassword" />
    </bean>
    <bean id="pooledConnectionFactory" class="org.apache.activemq.jms.pool.PooledConnectionFactory"
        init-method="start" destroy-method="stop">
        <property name="maxConnections" value="8" />
        <property name="connectionFactory" ref="jmsConnectionFactory" />
    </bean>
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="pooledConnectionFactory" />
    </bean>
    <bean id="customProcessor" class="...CustomProcessorClass" />
    <camelContext id="matrixProfileContext" xmlns="http://camel.apache.org/schema/spring">  
        <route id="matrixProfileRoute" autoStartup="false">
            <from uri="activemq:queue:queuename" />
            <log message="${body}" />
            <to uri="customProcessor" />
        </route>
    </camelContext>
</beans>

我的想法是，类CustomProcess解组通过路由传递的JSON内容，并将POJO推送到实现侦听器接口的侦听器类：

public interface ProcessorListenerIF {

    public void doOnDataProcessed(POJO processedData);
}

我通过单元测试测试整个设置：

public class TestProcessor extends TestCase {

    @Test
    public void testRoute() throws Exception {
        MyActiveMQConnector camelContext = new MyActiveMQConnector(new TestListener());
        try {
            camelContext.startConnections();
            Thread.sleep(1000);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            camelContext.stopConnection();
        }
    }

    private class TestListener implements ProcessorListenerIF {

        @Override
        public void doOnDataProcessed(POJO data) {
            System.out.println(data);
        }
    }
}

骆驼处理器有两种方法：

public void addListener(MatrixProfileProcessorListenerIF listener) {
    _processorListeners.add(listener);
}

@Override
public void process(Exchange exchange) throws Exception {

    Pseudocode: POJO data = unmarshal_by_JSON-JAVA(exchange)

    _processorListeners.parallelStream().forEach(listener -> {
        listener.doOnDataProcessed(data);
    });
}

其中我在ActiveMQConnector的构造函数中注册侦听器：

public class ActiveMQConnector {

    private SpringCamelContext _camelContext = null;

    public ActiveMQConnector(ProcessorListenerIF listener) {
        ApplicationContext appContext = new ClassPathXmlApplicationContext("camelContext.xml");
        _camelContext = new SpringCamelContext(appContext);
-------------------------------------
        ((CustomProcessor) _camelContext.getProcessor("customProcessor")).addListener(listener);
-------------------------------------
    }

    public void startConnections() throws Exception {
        try {
            _camelContext.start();
        } catch (Exception e) {
            exception handling
        }
    }
... more methods

上面突出显示的行（（CustomProcencer）...失败：语句_camelContext.get处理器没有找到任何内容，实例_camelContext中的路由为空。

如何实现将处理后的数据从处理器推送到某个观察者？

Answer 1

我找到了另一个解决方案，它完全基于Java，没有Spring XML。

单元测试仍然如上图所示。我没有通过Spring XML定义ActiveMQEndpoint，而是创建了一个新类：

public class MyActiveMQConnection {

    public static ActiveMQConnectionFactory createActiveMQConnectionFactory() {

        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        connectionFactory.setBrokerURL("tcp://<activemq-url>:<port>");
        connectionFactory.setUserName("myUsername");
        // connection factory configuration:
        connectionFactory.setUseAsyncSend(false);
        connectionFactory.setClientID(UUID.randomUUID().toString());
        connectionFactory.setConnectResponseTimeout(300);
        ... whatever ...

        return connectionFactory;
    }
}

此外，我更改了类ActiveMQConnector的构造函数：

public ActiveMQConnector(ProcessorListenerIF listener) throws Exception {

    _camelContext = new DefaultCamelContext();
    _camelContext.addComponent("activemqEndpoint",
            JmsComponent.jmsComponent(MyActiveMQConnection.createActiveMQConnectionFactory()));
    _camelContext.addRoutes(new RouteBuilder() {

        @Override
        public void configure() throws Exception {

            MyCustomProcessor processor = MyCustomProcessor.getInstance();
            processor.addListener(listener);

            from("activemqEndpoint:queue:matrixprofile") //
                    .process(processor) //
                    .to("stream:out");
        }
    });
}

我将处理器实现为单例，如下所示（为了完整性）：

public class MyCustomProcessor implements Processor {

    private final Set<ProcessorListenerIF> _processorListeners = new HashSet<>();
    private static volatile MyCustomProcessor      _myInstance         = null;
    private static Object                          _token              = new Object();

    private MyCustomProcessor() {

    }

    public static MyCustomProcessor getInstance() {
        MyCustomProcessor result = _myInstance;
        if (result == null) {
            synchronized (_token) {
                result = _myInstance;
                if (result == null)
                    _myInstance = result = new MyCustomProcessor();
            }
        }
        return result;
    }

    public void addListener(ProcessorListenerIF listener) {
        _processorListeners.add(listener);
    }

    /**
     * I assume the JSON has the following structure:
     * {timestamp: long, data: double[]}
    **/
    @Override
    public void process(Exchange exchange) throws Exception {

        _processorListeners.parallelStream().forEach(listener -> {
            // convert incoming message body to json object assuming data structure above
            JSONObject    jsonObject    = new JSONObject(exchange.getMessage().getBody().toString());
            MyPOJO myPojo = new MyPOJO();

            try {
                myPojo.setTimestamp(jsonObject.getLong("timestamp"));
            } catch (Exception e) {
                ...
            }
            try {
                JSONArray dataArray = jsonObject.getJSONArray("data");
                double[]  data      = new double[dataArray.length()];
                for (int i = 0; i < dataArray.length(); i++) {
                    data[i] = Double.valueOf(dataArray.get(i).toString());
                }
                myPojo.setData(data);
            } catch (Exception e) {
                ...
            }

            listener.doOnDataProcessed(myPojo);
        });
    }
}