第 41 章 消费者接口

org.apache.camel.Consumer 类型的实例代表路由中的源端点。实施消费者的方式有多种不同方法（请参阅 第 38.1.3 节 “消费者模式和线程”），这种灵活性在继承分级（查看 图 41.1 “消费者继承层次结构”）中有所反映，其中包括多个不同的基本类来实施消费者。

图 41.1. 消费者继承层次结构

对于遵循调度轮询模式的用户（请参阅 “调度的轮询模式”一节），Apache Camel 为将参数注入消费者实例提供支持。例如，考虑由 自定义 前缀标识的组件的以下端点 URI：

custom:destination?consumer.myConsumerParam

Apache Camel 支持自动注入 使用者形式的查询选项。\*。对于 consumer.myConsumerParam 参数，您需要在 Consumer 实施类上定义对应的 setter 和 getter 方法，如下所示：

public class CustomConsumer extends ScheduledPollConsumer {
    ...
    String getMyConsumerParam() { ... }
    void setMyConsumerParam(String s) { ... }
    ...
}

如果 getter 和 setter 方法遵循常见的 Java Bean 规范（包括属性名的第一个字母的大写）。

除了在消费者实施中定义 bean 方法外，还必须记住在 Endpoint.createConsumer（） 实施中调用 configureConsumer（） 方法（请参阅 “调度轮询端点实现”一节。

例 41.1 “FileEndpoint createConsumer（）实施” 显示了一个 createConsumer（） 方法实现示例，它取自文件组件中的 FileEndpoint 类：

...
public class FileEndpoint extends ScheduledPollEndpoint {
    ...
    public Consumer createConsumer(Processor processor) throws Exception {
        Consumer result = new FileConsumer(this, processor);
        configureConsumer(result);
        return result;
    }
    ...
    }

在运行时，使用者参数注入功能如下：

按照调度的轮询模式的用户自动支持 表 41.1 “调度的轮询参数” 中显示的消费者（它们可能显示为端点 URI 中的查询选项）。

Apache Camel 提供了两种特殊的消费者实施，可用于在事件驱动的消费者和轮询消费者之间转换和发送。提供了以下转换类：

在实践中，这些类用于简化实施端点类型的任务。Endpoint 接口定义了以下两种方法来创建消费者实例：

package org.apache.camel;

public interface Endpoint {
    ...
    Consumer createConsumer(Processor processor) throws Exception;
    PollingConsumer createPollingConsumer() throws Exception;
}

createConsumer（） 返回事件驱动的消费者，创建PollingConsumer（） 返回轮询消费者。您只能实施这些方法。例如，如果您遵循消费者的事件驱动的模式，则实施 createConsumer（） 方法，为创建 PollingConsumer（） 提供方法实施，以简单地引发异常。但是，通过转换类，Apache Camel 可以提供更有用的默认实施。

例如，如果要根据事件驱动的模式实施消费者，您可以通过扩展 DefaultEndpoint 并实施 createConsumer（） 方法来实施端点。createPollingConsumer（） 的实现从 DefaultEndpoint 继承，其定义如下：

public PollingConsumer<E> createPollingConsumer() throws Exception {
    return new EventDrivenPollingConsumer<E>(this);
}

EventDrivenPollingConsumer 构造器引用事件驱动的消费者，这 实际上将其嵌套，并将其转换为轮询消费者。要实现转换，EventDrivenPollingConsumer 实例缓冲传入的事件，并通过 receive （长超时）和 receive NoWait（） 方法根据需要提供它们。

类似地，如果您按照轮询模式实施消费者，通过扩展 DefaultPollingEndpoint 和实施 createPollingConsumer（） 方法来实施端点。在本例中，createConsumer（） 方法的实现从 DefaultPollingEndpoint 继承，默认的实现会返回一个 DefaultScheduledPollConsumer 实例（将轮询消费者转换为事件驱动的消费者）。

消费者类可以选择实施 org.apache.camel.spi.ShutdownPrepared 接口，它使您的自定义消费者端点能够接收关闭通知。

例 41.2 “ShutdownPrepared Interface” 显示 ShutdownPrepared 接口的定义。

package org.apache.camel.spi;

public interface ShutdownPrepared {

    void prepareShutdown(boolean forced);

}

ShutdownPrepared 接口定义了以下方法：

消费者类可以选择实施 org.apache.camel.spi.ShutdownAware 接口，该界面与安全关闭机制交互，让消费者要求额外的时间关机。这通常需要这个组件，比如 SEDA，它们可能会使存储在内部队列中的待处理交换。通常，您希望在关闭 SEDA 消费者之前处理队列中的所有交换。

例 41.3 “ShutdownAware Interface” 显示 ShutdownAware 接口的定义。

// Java
package org.apache.camel.spi;

import org.apache.camel.ShutdownRunningTask;

public interface ShutdownAware extends ShutdownPrepared {

    boolean deferShutdown(ShutdownRunningTask shutdownRunningTask);

    int getPendingExchangesSize();
}

ShutdownAware 接口定义以下方法：

有关如何定义 ShutdownAware 方法的示例，请参考 例 41.7 “自定义线程实施”。