Spring 对 JMS 队列的集成负载平衡

Question

我想从单个输入队列中获取 JMS 消息并将它们扇出到 N 个输出队列中。

我有一个简单的流程，可以将消息转发到单个目的地，但不知道如何应用 LoadBalancer 以循环方式允许多个目的地。

知道怎么做吗？

@Configuration
public class TestLoadBalance {

    public static final String INPUT_QUEUE = "_dev.lb.input";
    public static final String OUTPUT_QUEUE_PREFIX = "_dev.lb.output-";


    @Bean
    public IntegrationFlow testLoadBalanceFlow(
            ConnectionFactory jmsConnectionFactory) {

        IntegrationFlow flow =  IntegrationFlows.from(
                Jms.messageDrivenChannelAdapter(jmsConnectionFactory)
                        .destination(INPUT_QUEUE)
        )
                .handle(buildOutput(jmsConnectionFactory, 1))
                // cant have 2nd handle. gets warn & flow end:
                // The 'currentComponent' (org.springframework.integration.jms.JmsSendingMessageHandler@516462cc) 
                // is a one-way 'MessageHandler' and it isn't appropriate to configure 'outputChannel'
                //.handle(buildOutput(jmsConnectionFactory, 2))
                .get();
        return flow;
    }


    private JmsSendingMessageHandler buildOutput(ConnectionFactory jmsConnectionFactory, int i){
        return Jms.outboundAdapter(jmsConnectionFactory)
                .destination(OUTPUT_QUEUE_PREFIX + i).get();
    }
}

Answer 1

有几种方法可以做到；您可以在频道上拥有多个订阅者...

@Bean
public IntegrationFlow inbound(ConnectionFactory cf) {
    return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(cf)
                .destination("foo"))
            .channel(roundRobin())
            .get();
}

@Bean
public DirectChannel roundRobin() {
    return new DirectChannel();
}

@Bean
public IntegrationFlow outbound1(ConnectionFactory cf) {
    return IntegrationFlows.from(roundRobin())
            .bridge() // otherwise log() will wire tap the roundRobin channel
            .log()
            .log(new LiteralExpression("Sending to bar"))
            .handle(Jms.outboundAdapter(cf)
                    .destination("bar"))
            .get();
}

@Bean
public IntegrationFlow outbound2(ConnectionFactory cf) {
    return IntegrationFlows.from(roundRobin())
            .bridge() // otherwise log() will wire tap the roundRobin channel
            .log()
            .log(new LiteralExpression("Sending to baz"))
            .handle(Jms.outboundAdapter(cf)
                    .destination("baz"))
            .get();
}

或者，您可以使用目标表达式：

@Bean
public AtomicInteger toggle() {
    return new AtomicInteger();
}

@Bean
public IntegrationFlow inbound(ConnectionFactory cf) {
    return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(cf)
                    .destination("foo"))
            .handle(Jms.outboundAdapter(cf)
                    .destinationExpression("@toggle.getAndIncrement() % 2 == 0 ? 'bar' : 'baz'"))
            .get();
}

@JmsListener(destination = "bar")
public void bar(String in) {
    System.out.println("received " + in + " from bar");
}

@JmsListener(destination = "baz")
public void baz(String in) {
    System.out.println("received " + in + " from baz");
}

结果：

received test1 from bar
received test2 from baz

Answer 2

根据 Gary 的示例，我采用了 destinationExpression 方法：

@Configuration
public class TestLoadBalance {

    public static final String INPUT_QUEUE = "_dev.lb.input";
    public static final String OUTPUT_QUEUE_PREFIX = "_dev.lb.output-";

    @Bean
    public JmsDestinationPartitioner partitioner() {
        return new JmsDestinationPartitioner(OUTPUT_QUEUE_PREFIX,1,3);
    }

    @Bean
    public IntegrationFlow testLoadBalanceFlow(
            ConnectionFactory jmsConnectionFactory) {

        IntegrationFlow flow =  IntegrationFlows.from(
                Jms.messageDrivenChannelAdapter(jmsConnectionFactory)
                        .destination(INPUT_QUEUE)
        )
                .handle(Jms.outboundAdapter((jmsConnectionFactory))
                        .destinationExpression("@partitioner.nextDestination()"))
                .get();
        return flow;
    }

}

使用 AtomicInt 的包装器来处理带有前缀的命名：

public class JmsDestinationPartitioner {

    private int min;
    private int max;
    private String prefix;

    private AtomicInteger current;

    public JmsDestinationPartitioner(String prefix, int min, int max){
        this.prefix = prefix;
        this.min = min;
        this.max = max;
        current = new AtomicInteger(min);
    }


    public int getAndIncrement(){
        int  i = current.get();
        current.getAndIncrement();
        if (current.get() > max){
            current.set(min);
        }
        return i;
    }

    public String nextDestination(){
        return prefix + getAndIncrement();
    }
}

Spring 对 JMS 队列的集成负载平衡

Spring Integration Load Balance to JMS Queues

spring-integration

spring-integration-dsl