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