Red Hat Summit 红帽全球峰会支持控制台（Console）开发人员开始试用联系人

41.2. 实现使用者接口

实施消费者的替代方法
复制链接

您可以通过以下任一方法实现消费者：

事件驱动的消费者实施
复制链接

在事件驱动的消费者中，外部事件明确驱动处理。事件通过 event-listener 接口接收，其中监听程序接口特定于特定事件源。

例 41.4 “JMXConsumer 实施” 演示了 JMXConsumer 类的实施，该类取自 Apache Camel JMX 组件实施。JMXConsumer 类是事件驱动的消费者的示例，通过从 org.apache.camel.impl.DefaultConsumer 类继承来实施该类。在 JMXConsumer 示例示例中，事件由 notification Listener.handleNotification（） 方法上的调用表示，这是接收 JMX 事件的标准方法。要接收这些 JMX 事件，需要实施 NotificationListener 接口并覆盖 handleNotification（） 方法，如 例 41.4 “JMXConsumer 实施” 所示。

例 41.4. JMXConsumer 实施

package org.apache.camel.component.jmx;

import javax.management.Notification;
import javax.management.NotificationListener;
import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultConsumer;

public class JMXConsumer extends DefaultConsumer implements NotificationListener {
1 


    JMXEndpoint jmxEndpoint;

    public JMXConsumer(JMXEndpoint endpoint, Processor processor) {
2 

        super(endpoint, processor);
        this.jmxEndpoint = endpoint;
    }

    public void handleNotification(Notification notification, Object handback) {
3 

        try {
            getProcessor().process(jmxEndpoint.createExchange(notification));
4 

        } catch (Throwable e) {
            handleException(e);
5 

        }
    }
}
Copy to Clipboard Toggle word wrap
1
JMXConsumer 模式通过扩展 DefaultConsumer 类，遵循事件驱动的用户的常见模式。此外，由于此使用者旨在接收来自 JMX 通知的事件（由 JMX 通知表示），因此实施 NotificationListener 接口是必须的。
2
您必须实施一个构造器，它引用了父端点、端点，以及对链的下一个处理器（ 处理器 ）的引用。
3
当 JMX 通知到达时，handleNotification（） 方法（在 NotificationListener中定义）会自动调用 JMX。这个方法的正文应该包含执行消费者的事件处理的代码。由于 handleNotification（） 调用源自 JMX 层，因此消费者的线程模型由 JMX 层隐式控制，而不是由 JMXConsumer 类控制。
4
这种代码行组合了两个步骤。首先，JMX 通知对象转换为交换对象，这是 Apache Camel 中事件的通用表示。然后，新创建的交换对象会被传递给路由中的下一个处理器（同步）。
5
handleException（） 方法由 DefaultConsumer 基本类实施。默认情况下，它利用 org.apache.camel.impl.LoggingExceptionHandler 类来处理异常。
注意

handleNotification（） 方法特定于 JMX 示例。在实施您自己的事件驱动的消费者时，您必须确定要在自定义消费者中实施的类似事件监听程序方法。

计划轮询消费者实施
复制链接

在计划的轮询消费者中，轮询事件由计时器类自动生成，java.util.concurrent.ScheduledExecutorService。要接收生成的轮询事件，您必须实施 ScheduledPollConsumer.poll（） 方法（请参阅 第 38.1.3 节 “消费者模式和线程”。

例 41.5 “ScheduledPollConsumer 实现” 演示了如何实施遵循调度轮询模式的使用者，具体是通过扩展 ScheduledPollConsumer 类来实现的。

例 41.5. ScheduledPollConsumer 实现

import java.util.concurrent.ScheduledExecutorService;

import org.apache.camel.Consumer;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.PollingConsumer;
import org.apache.camel.Processor;

import org.apache.camel.impl.ScheduledPollConsumer;

public class pass:quotes[CustomConsumer] extends ScheduledPollConsumer {
1 

    private final pass:quotes[CustomEndpoint] endpoint;

    public pass:quotes[CustomConsumer](pass:quotes[CustomEndpoint] endpoint, Processor processor) {
2 

        super(endpoint, processor);
        this.endpoint = endpoint;
    }

    protected void poll() throws Exception {
3 

        Exchange exchange = /* Receive exchange object ... */;

        // Example of a synchronous processor.
        getProcessor().process(exchange);
4 

    }

    @Override
    protected void doStart() throws Exception {
5 

        // Pre-Start:
        // Place code here to execute just before start of processing.
        super.doStart();
        // Post-Start:
        // Place code here to execute just after start of processing.
    }

    @Override
    protected void doStop() throws Exception {
6 

        // Pre-Stop:
        // Place code here to execute just before processing stops.
        super.doStop();
        // Post-Stop:
        // Place code here to execute just after processing stops.
    }
}
Copy to Clipboard Toggle word wrap
1
通过扩展 org.apache.camel.impl.ScheduledPollConsumer 类来实施计划的轮询类。
2
您必须实施一个构造器，它引用了父端点、端点，以及对链的下一个处理器（ 处理器 ）的引用。
3
覆盖 poll（） 方法，以接收调度的轮询事件。这是您应该将检索和处理传入事件的代码位置（由交换对象代表）。
4
在本例中，事件被同步处理。如果要异步处理事件，您应该通过调用 getAsyncProcessor（） 来使用对异步处理器的引用。有关如何异步处理事件的详情，请参考 第 38.1.4 节 “异步处理”。
5
（可选） 如果要在消费者启动时执行某些代码行，请按如下所示覆盖 doStart（） 方法。
6
（可选） 如果您希望以消费者的身份执行某些代码行是停止的，请按如下所示覆盖 doStop（） 方法。

轮询消费者实施
复制链接

例 41.6 “PollingConsumerSupport 实现” 概述了如何实施遵循轮询模式的消费者，该模式通过扩展Polling ConsumerSupport 类来实现。

例 41.6. PollingConsumerSupport 实现

import org.apache.camel.Exchange;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.impl.PollingConsumerSupport;

public class pass:quotes[CustomConsumer] extends PollingConsumerSupport {
1 

    private final pass:quotes[CustomEndpoint] endpoint;

    public pass:quotes[CustomConsumer](pass:quotes[CustomEndpoint] endpoint) {
2 

        super(endpoint);
        this.endpoint = endpoint;
    }

    public Exchange receiveNoWait() {
3 

        Exchange exchange = /* Obtain an exchange object. */;
        // Further processing ...
        return exchange;
    }

    public Exchange receive() {
4 

        // Blocking poll ...
    }

    public Exchange receive(long timeout) {
5 

        // Poll with timeout ...
    }

    protected void doStart() throws Exception {
6 

        // Code to execute whilst starting up.
    }

    protected void doStop() throws Exception {
        // Code to execute whilst shutting down.
    }
}
Copy to Clipboard Toggle word wrap
1
通过扩展 org.apache.camel.impl.PollingConsumerSupport 类来实施轮询消费者类。
2
您必须实施一个构造器，它将引用父端点 端点端点端点 作为参数。轮询消费者不需要对处理器实例的引用。
3
receiveNoWait（） 方法应该实施用于检索事件（exchange 对象）的非阻塞算法。如果没有可用的事件，它应该返回 null。
4
receive（） 方法应该实施用于检索事件的阻塞算法。如果事件不可用，此方法可以无限期地阻止。
5
receive（长超时） 方法实施一个算法，只要指定的超时（通常在毫秒单位指定）。
6
如果要在启动或关闭消费者时插入执行的代码，请分别实施 doStart（） 方法和 doStop（） 方法。

自定义线程实现
复制链接

如果标准消费者模式不适用于您的消费者实施，您可以直接实现 Consumer 接口，并自行编写线程代码。但是，在编写线程代码时，务必要遵守标准 Apache Camel 线程模型，如 第 2.8 节 “线程模型” 所述。

例如，来自 camel-core 的 SEDA 组件实施自己的消费者线程处理，与 Apache Camel 线程模型一致。例 41.7 “自定义线程实施” 展示了 SedaConsumer 类如何实施其线程的概要。

例 41.7. 自定义线程实施

package org.apache.camel.component.seda;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.camel.Consumer;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.ShutdownRunningTask;
import org.apache.camel.impl.LoggingExceptionHandler;
import org.apache.camel.impl.ServiceSupport;
import org.apache.camel.util.ServiceHelper;
...
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * A Consumer for the SEDA component.
 *
 * @version $Revision: 922485 $
 */
public class SedaConsumer extends ServiceSupport implements Consumer, Runnable, ShutdownAware {
1 

    private static final transient Log LOG = LogFactory.getLog(SedaConsumer.class);

    private SedaEndpoint endpoint;
    private Processor processor;
    private ExecutorService executor;
    ...
    public SedaConsumer(SedaEndpoint endpoint, Processor processor) {
        this.endpoint = endpoint;
        this.processor = processor;
    }
    ...

    public void run() {
2 

        BlockingQueue<Exchange> queue = endpoint.getQueue();
        // Poll the queue and process exchanges
        ...
    }

    ...
    protected void doStart() throws Exception {
3 

        int poolSize = endpoint.getConcurrentConsumers();
        executor = endpoint.getCamelContext().getExecutorServiceStrategy()
            .newFixedThreadPool(this, endpoint.getEndpointUri(), poolSize);
4 

        for (int i = 0; i < poolSize; i++) {
5 

            executor.execute(this);
        }
        endpoint.onStarted(this);
    }

    protected void doStop() throws Exception {
6 

        endpoint.onStopped(this);
        // must shutdown executor on stop to avoid overhead of having them running
        endpoint.getCamelContext().getExecutorServiceStrategy().shutdownNow(executor);
7 


        if (multicast != null) {
            ServiceHelper.stopServices(multicast);
        }
    }
    ...
    //----------
    // Implementation of ShutdownAware interface

    public boolean deferShutdown(ShutdownRunningTask shutdownRunningTask) {
        // deny stopping on shutdown as we want seda consumers to run in case some other queues
        // depend on this consumer to run, so it can complete its exchanges
        return true;
    }

    public int getPendingExchangesSize() {
        // number of pending messages on the queue
        return endpoint.getQueue().size();
    }

}
Copy to Clipboard Toggle word wrap
1
SedaConsumer 类通过扩展 org.apache.camel.impl.ServiceSupport 类来实施，并实施 使用者、Runnable 和 ShutdownAware 接口。
2
实施 Runnable.run（） 方法，以定义使用者在线程中运行时所执行的操作。在这种情况下，使用者在循环中运行，轮询新交换的队列，然后在队列的后一部分处理交换。
3
doStart（） 方法继承自 ServiceSupport。您覆盖这种方法，以定义消费者启动时的作用。
4
您不必直接创建线程，而是使用在 CamelContext 中注册的 ExecutorServiceStrategy 对象创建一个线程池。这很重要，因为它可让 Apache Camel 实现对线程的集中管理并支持如安全关闭。详情请查看 第 2.8 节 “线程模型”。
5
通过调用 ExecutorService.execute（） 方法 池Size 时间来启动线程。
6
doStop（） 方法从 ServiceSupport 继承。您可以覆盖这个方法，以定义消费者在关闭时做什么。
7
关闭线程池，该池由 executor 实例表示。
返回顶部
Red Hat logoGithubredditYoutubeTwitter

学习

尝试、购买和销售

社区

关于红帽文档

通过我们的产品和服务，以及可以信赖的内容，帮助红帽用户创新并实现他们的目标。 了解我们当前的更新.

让开源更具包容性

红帽致力于替换我们的代码、文档和 Web 属性中存在问题的语言。欲了解更多详情，请参阅红帽博客.

關於紅帽

我们提供强化的解决方案，使企业能够更轻松地跨平台和环境（从核心数据中心到网络边缘）工作。

Theme

© 2025 Red Hat