第 41 章 消费者接口

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

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

View larger image

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

custom:destination?consumer.myConsumerParam

custom:destination?consumer.myConsumerParam

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

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

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

getter 和 setter 方法遵循常见的 Java bean 惯例（包括属性名称的第一个字母大写）。

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

例 41.1 “FileEndpoint createConsumer （）实现” 显示从文件组件中的 FileEndpoint 类的 createConsumer （） 方法实现的示例：

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

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

在运行时，消费者参数注入可以正常工作：

遵循调度的轮询模式的消费者自动支持 表 41.1 “调度的 Poll 参数” 中显示的消费者参数（它可以显示为端点 URI 中的查询选项）。

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

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

package org.apache.camel;

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

package org.apache.camel;

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

createConsumer （） 返回事件驱动的消费者，createPollingConsumer （） 返回轮询消费者。您只实施这些方法。例如，如果您遵循消费者的事件驱动模式，您可以实施 createConsumer （） 方法，以便为 createPollingConsumer （） 提供方法实现，该方法只是引发异常。但是，在转换类的帮助下，Apache Camel 可以提供更有用的默认实现。

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

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

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);

}

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();
}

// 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 “自定义线程实施”。