java - 如何在 Camel 中实现事件驱动的消费者

Question

我对 Camel 非常陌生，并且一直在努力理解如何在特定场景中使用 Camel。在此场景中，有一个(基于 Java 的)代理不时生成操作。我需要一个事件驱动的消费者来获取这些事件的通知。这些事件将被路由到"file"生成器(暂时)。

在《 Camel 书》中，示例是针对轮询消费者的。我找不到针对事件驱动消费者的通用解决方案。我遇到了类似的 JMX 实现:

public class JMXConsumer extends DefaultConsumer implements NotificationListener { 

JMXEndpoint jmxEndpoint;    
public JMXConsumer(JMXEndpoint endpoint, Processor processor) { 
    super(endpoint, processor);
    this.jmxEndpoint = endpoint;
}

public void handleNotification(Notification notification, Object handback) { 
    try {
        getProcessor().process(jmxEndpoint.createExchange(notification)); 
    } catch (Throwable e) {
        handleException(e); 
    }
}

}

这里，只要 JMX 通知到达，就会调用handleNotification。

我相信我必须做类似的事情，以便在代理生成操作时通知我的消费者。然而，上面的handleNotification方法是JMX特有的。该网页显示:“在实现您自己的事件驱动消费者时，您必须确定要在自定义消费者中实现的类似事件监听器方法。”

我想知道:如何识别类似的事件监听器，以便每当我的代理有操作时我的消费者都会收到通知。

非常感谢任何建议/网页链接。

Answer 1

我知道这是一个老问题，但我一直在努力解决这个问题，只是想我会为其他寻找答案的人记录我的发现。

当您创建 Endpoint 类(扩展 DefaultEndpoint)时，您将重写以下用于创建使用者的方法:

public Consumer createConsumer(Processor processor)

然后，在您的消费者中，您可以访问处理器 - 在此处理器上调用“process”将创建一个事件并触发路由。

例如，假设您有一些监听消息的 Java API，并且有某种监听器。就我而言，监听器将传入消息放入 LinkedBlockingQueue，我的 Consumer 'doStart' 方法如下所示(添加您自己的错误处理):

@Override
protected void doStart() throws Exception {
    super.doStart();

    // Spawn a new thread that submits exchanges to the Processor
    Runnable runnable = new Runnable() {
        @Override
        public void run() {
            while(true) {
                IMessage incomingMessage = myLinkedBlockingQueue.take();
                Exchange exchange = getEndpoint().createExchange();
                exchange.getIn().setBody(incomingMessage);
                myProcessor.process(exchange);  
            }
        }
    };
    new Thread(runnable).start();
}

现在我可以将创建 Endpoint 的组件放入我的 CamelContext 中，并像这样使用它:

from("mycomponent:incoming").to("log:messages");

每次从 Java API 到达新消息时都会触发日志消息。

希望对某人有帮助!