42.2. 实施 Producer 接口

实现制作者的替代方法

您可以通过以下方法之一实现制作者:

如何实施同步制作者

例 42.4 “DefaultProducer 实现” 概述了如何实施同步制作者。在这种情况下,调用 Producer.process() 块,直到收到回复为止。

例 42.4. DefaultProducer 实现

import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Producer;
import org.apache.camel.impl.DefaultProducer;

public class CustomProducer extends DefaultProducer { 1

    public CustomProducer(Endpoint endpoint) { 2
        super(endpoint);
        // Perform other initialization tasks...
    }

    public void process(Exchange exchange) throws Exception { 3
        // Process exchange synchronously.
        // ...
    }
}
1
通过扩展 org.apache.camel.impl.Default Producer 类来实施自定义 同步制作者类。
2
实施对父端点的引用。
3
process() 方法实施代表制作者代码的核心。process() 方法的实现完全取决于您实施的组件类型。

在概述中,process() 方法通常实施如下:

  • 如果交换包含 In 消息,如果这与指定的交换模式一致,请将 In 消息发送到指定的端点。
  • 如果交换模式预测出 Out 消息的收据,请等待 Out 消息收到。这通常会导致 process() 方法阻止了非常长的时间。
  • 收到回复后,调用 exchange.setOut() 将回复附加到交换对象。如果回复包含故障消息,请使用 Message.setFault(true)Out 消息上设置 fault 标志。

如何实施异步制作者

例 42.5 “CollectionProducer 实现” 概述了如何实施异步制作者。在这种情况下,您必须同时实施同步 process() 方法和异步 process() 方法(需要使用额外的 AsyncCallback 参数)。

例 42.5. CollectionProducer 实现

import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProcessor;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Producer;
import org.apache.camel.impl.DefaultProducer;

public class _CustomProducer_ extends DefaultProducer implements AsyncProcessor { 1

    public _CustomProducer_(Endpoint endpoint) { 2
        super(endpoint);
        // ...
    }

    public void process(Exchange exchange) throws Exception { 3
        // Process exchange synchronously.
        // ...
    }

    public boolean process(Exchange exchange, AsyncCallback callback) { 4
        // Process exchange asynchronously.
        CustomProducerTask task = new CustomProducerTask(exchange, callback);
        // Process 'task' in a separate thread...
        // ...
        return false; 5
    }
}

public class CustomProducerTask implements Runnable { 6
    private Exchange exchange;
    private AsyncCallback callback;

    public CustomProducerTask(Exchange exchange, AsyncCallback callback) {
        this.exchange = exchange;
        this.callback = callback;
    }

    public void run() { 7
        // Process exchange.
        // ...
        callback.done(false);
    }
}
1
通过扩展 org.apache.camel.impl.Default Producer 类以及实施 AsyncProcessor 接口来实施自定义 异步制作者类。
2
实施对父端点的引用。
3
实施同步 process() 方法。
4
实施异步 process() 方法。您可以通过几种方法实现异步方法。此处所示的方法是创建一个 java.lang.Runnable 实例 task,它代表在子线程中运行的代码。然后,您可以使用 Java 线程 API 在子线程中运行任务(例如,通过创建新线程,或通过将任务分配给现有线程池)。
5
通常,您从异步 process() 方法返回 false,以指示交换是异步处理的。
6
CustomProducerTask 类封装了在子线程中运行的处理代码。此类必须存储 Exchange 对象、交换AsyncCallback 对象的副本,作为专用成员变量。
7
run() 方法包含将 In 消息发送到 制作者端点的代码,并等待收到回复(若有)。收到回复(传出消息或 故障 消息)并将其插入到交换对象后,您必须调用 callback.done() 以通知处理完成的调用者。