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

41.2. 实施 Consumer 接口

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

您可以使用以下方法之一实现消费者：

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

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

例 41.4 “JMXConsumer 实现” 显示 JMXConsumer 类的实现，该类从 Apache Camel JMX 组件实施中获取。JMXConsumer 类是一个事件驱动的消费者示例，它由从 org.apache.camel.impl.DefaultConsumer 类继承来实现。对于 JMXConsumer 示例，事件由 NotificationListener.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 通知到达时，由 JMX 自动调用 handleNotification （） 方法（在 NotificationListener中定义）。此方法的正文应包含执行使用者事件处理的代码。由于 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 Implementation” 演示了如何实施遵循调度的轮询模式的消费者，该模式通过扩展 ScheduledPollConsumer 类来实现。

例 41.5. ScheduledPollConsumer Implementation

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 类，实施调度的轮询消费者类 CustomConsumer。
2
您必须至少实施一个构造器，它将对父 端点、端点 和链中下一 处理器 的引用作为参数。
3
覆盖 poll （） 方法，以接收调度的轮询事件。在这里，您应该放置检索和处理传入事件（由 Exchange 对象代表）的代码。
4
在本例中，事件是同步处理的。如果要异步处理事件，您应该通过调用 getAsyncProcessor （） 来使用对异步处理器的引用。有关如何异步处理事件的详情，请参考 第 38.1.4 节 “异步处理”。
5
（可选） 如果您希望将某些代码执行为消费者启动，请按所示覆盖 doStart （） 方法。
6
（可选） 如果您希望将某些代码执行为消费者停止，请按所示覆盖 doStop （） 方法。

轮询消费者实施
复制链接

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

例 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 类，实施轮询消费者类 CustomConsumer。
2
您必须至少实施一个构造器，它将引用父端点 端点端点 作为参数。轮询消费者不需要引用处理器实例。
3
receiveNoWait （） 方法应该实施用于检索事件（交换对象）的非阻塞算法。如果没有可用的事件，它应该返回 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 类来实现，并实施 Consumer、Runnable 和 ShutdownAware 接口。
2
实施 Runnable.run （） 方法，以定义消费者在线程中运行时执行的操作。在这种情况下，消费者在循环中运行，轮询新交换的队列，然后在队列的后部分处理交换。
3
doStart （） 方法继承自 ServiceSupport。您可以覆盖此方法，以定义消费者启动时的作用。
4
您应该使用 CamelContext 注册的 ExecutorServiceStrategy 对象创建线程池，而不是直接创建线程。这很重要，因为它使 Apache Camel 能够实施对线程的集中管理，并支持如安全关闭等功能。详情请查看 第 2.8 节 “线程模型”。
5
通过调用 ExecutorService.execute （） 方法 poolSize 时间来启动线程。
6
doStop （） 方法继承自 ServiceSupport。您可以覆盖此方法，以定义消费者关闭时的作用。
7
关闭线程池，由 executor 实例表示。
返回顶部
Red Hat logoGithubredditYoutubeTwitter

学习

尝试、购买和销售

社区

关于红帽文档

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

让开源更具包容性

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

關於紅帽

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

Theme

© 2025 Red Hat