spring - 在 Spring 中以编程方式将 QueueChannel 桥接到 MessageChannel

Question

我正在尝试将队列连接到 MessageChannel 的前面，并且我需要以编程方式执行此操作，以便可以在运行时完成以响应 osgi:listener 被触发。到目前为止我已经得到:

public void addService(MessageChannel mc, Map<String,Object> properties)
{
    //Create the queue and the QueueChannel
    BlockingQueue<Message<?>> q = new LinkedBlockingQueue<Message<?>>();
    QueueChannel qc = new QueueChannel(q);

    //Create the Bridge and set the output to the input parameter channel
    BridgeHandler b = new BridgeHandler();
    b.setOutputChannel(mc);

    //Presumably, I need something here to poll the QueueChannel
    //and drop it onto the bridge.  This is where I get lost

}

查看各种相关类(class)，我想出了:

    PollerMetadata pm = new PollerMetadata();
    pm.setTrigger(new IntervalTrigger(10));

    PollingConsumer pc = new PollingConsumer(qc, b);

但我无法将它们放在一起。我错过了什么？

Answer 1

所以，最终对我有用的解决方案是:

public void addEngineService(MessageChannel mc, Map<String,Object> properties)
{
    //Create the queue and the QueueChannel
    BlockingQueue<Message<?>> q = new LinkedBlockingQueue<Message<?>>();
    QueueChannel qc = new QueueChannel(q);

    //Create the Bridge and set the output to the input parameter channel 
    BridgeHandler b = new BridgeHandler();
    b.setOutputChannel(mc);

    //Setup a Polling Consumer to poll the queue channel and 
    //retrieve 1 thing at a time
    PollingConsumer pc = new PollingConsumer(qc, b);
    pc.setMaxMessagesPerPoll(1);

    //Now use an interval trigger to poll every 10 ms and attach it
    IntervalTrigger trig = new IntervalTrigger(10, TimeUnit.MILLISECONDS);
    trig.setInitialDelay(0);
    trig.setFixedRate(true);
    pc.setTrigger(trig);

    //Now set a task scheduler and start it
    pc.setTaskScheduler(taskSched);
    pc.setAutoStartup(true);
    pc.start();
}

我不太清楚上述所有内容是否都是明确需要的，但触发器或任务调度程序单独工作都不起作用，我似乎确实需要两者。我还应该注意，使用的 taskSched 是通过

从 spring 注入(inject)的默认 taskScheduler 依赖项

<property name="taskSched" ref="taskScheduler"/>