Spring cloud stream RabbitMQ动态路由消息

Question

我已经实现了如下所示的示例Spring Dynamic Destination

在rabbitmq中，它动态地创建一个交换，但没有提供绑定或路由密钥的选项。我的要求是用路由密钥向这个动态创建的exchange发送消息。我需要如何实现这一点来设置路由密钥？

@Component
public class DDProducerBean {

    @Autowired
    private BinderAwareChannelResolver poChannelResolver = null;

    public void publish(DDSocketVO ddSocketVO) throws Exception {
        this.poChannelResolver.resolveDestination(ddSocketVO.getDestination()).send(MessageBuilder.withPayload(new ObjectMapper().
                setVisibility(PropertyAccessor.FIELD, Visibility.ANY).
                writeValueAsString(ddSocketVO)).build());
    }

}

Answer 1

以下是建议的解决办法

基本上，使用BinderAwareChannelResolver创建一个带有动态目的地的MessageChannel，然后使用RabbitAdmin API连接到RabbitMQ，并在发送消息之前使用路由密钥将新创建的exchange绑定到另一个队列或exchange。

@Autowired
private BinderAwareChannelResolver poChannelResolver;

public void publish(WebSocketVO webSocketVO) throws Exception {

    MessageChannel channel = this.poChannelResolver.resolveDestination(webSocketVO.getDestination());

    CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
    connectionFactory.setUsername(System.getProperty("spring.cloud.stream.binders.corerabbit.environment.spring.rabbitmq.username"));
    connectionFactory.setPassword(System.getProperty("spring.cloud.stream.binders.corerabbit.environment.spring.rabbitmq.password"));
    connectionFactory.setAddresses(System.getProperty("spring.cloud.stream.binders.corerabbit.environment.spring.rabbitmq.addresses"));
    connectionFactory.setVirtualHost(System.getProperty("spring.cloud.stream.binders.corerabbit.environment.spring.rabbitmq.virtual-host"));

    AmqpAdmin amqpAdmin = new RabbitAdmin(connectionFactory);

    TopicExchange sourceExchange = new TopicExchange(webSocketVO.getDestination(), false, true);
    TopicExchange destExchange = new TopicExchange("amq.topic");

    amqpAdmin.declareBinding(BindingBuilder.bind(destExchange).to(sourceExchange).with(webSocketVO.getRoutingKeyExpression()));




    channel.send(MessageBuilder.withPayload(new ObjectMapper().
            setVisibility(PropertyAccessor.FIELD, Visibility.ANY).
            writeValueAsString(webSocketVO)).build());


    amqpAdmin.deleteExchange(webSocketVO.getDestination());

    connectionFactory.destroy();

}