使用带有 Request/Reply 模式的 AsyncRabbitTemplate 时如何构建非阻塞消费者

Question

我是 rabbitmq 的新手，目前正在尝试使用非阻塞消费者实现非阻塞生产者。我建立了一些测试制作人，在那里我使用了类型引用：

@Service
public class Producer {
    @Autowired
    private AsyncRabbitTemplate asyncRabbitTemplate;

    public <T extends RequestEvent<S>, S> RabbitConverterFuture<S> asyncSendEventAndReceive(final T event) {
        return asyncRabbitTemplate.convertSendAndReceiveAsType(QueueConfig.EXCHANGE_NAME, event.getRoutingKey(), event, event.getResponseTypeReference());
    }
}

在其他地方，在 RestController 中调用的测试函数

@Autowired
Producer producer;

public void test() throws InterruptedException, ExecutionException {
    TestEvent requestEvent = new TestEvent("SOMEDATA");
    RabbitConverterFuture<TestResponse> reply = producer.asyncSendEventAndReceive(requestEvent);
    log.info("Hello! The Reply is: {}", reply.get());
}

到目前为止，这非常简单，我现在遇到的问题是如何创建一个非阻塞的消费者。我目前的听众：

@RabbitListener(queues = QueueConfig.QUEUENAME)
public TestResponse onReceive(TestEvent event) {
    Future<TestResponse> replyLater = proccessDataLater(event.getSomeData())
    return replyLater.get();
}

据我所知，在使用@RabbitListener 时，此侦听器在其自己的线程中运行。我可以配置 MessageListener 为活动侦听器使用多个线程。因此，使用 future.get() 阻塞侦听器线程并不会阻塞应用程序本身。仍然可能存在所有线程现在都阻塞并且新事件卡在队列中的情况，而它们可能不需要。我想做的是只接收事件而不需要立即 return 结果。 @RabbitListener 可能无法做到这一点。类似于：

@RabbitListener(queues = QueueConfig.QUEUENAME)
public void onReceive(TestEvent event) {
   /* 
    * Some fictional RabbitMQ API call where i get a ReplyContainer which contains
    * the CorrelationID for the event. I can call replyContainer.reply(testResponse) later 
    * in the code without blocking the listener thread
    */        
    ReplyContainer replyContainer = AsyncRabbitTemplate.getReplyContainer()

    // ProcessDataLater calls reply on the container when done with its action
    proccessDataLater(event.getSomeData(), replyContainer);
}

在 spring 中使用 rabbitmq 实现此类行为的最佳方法是什么？

编辑配置 Class:

@Configuration
@EnableRabbit
public class RabbitMQConfig implements RabbitListenerConfigurer {

    public static final String topicExchangeName = "exchange";

    @Bean
    TopicExchange exchange() {
        return new TopicExchange(topicExchangeName);
    }

    @Bean
    public ConnectionFactory rabbitConnectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost("localhost");
        return connectionFactory;
    }

    @Bean
    public MappingJackson2MessageConverter consumerJackson2MessageConverter() {
        return new MappingJackson2MessageConverter();
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(rabbitConnectionFactory());
        rabbitTemplate.setMessageConverter(producerJackson2MessageConverter());
        return rabbitTemplate;
    }

    @Bean
    public AsyncRabbitTemplate asyncRabbitTemplate() {
        return new AsyncRabbitTemplate(rabbitTemplate());
    }

    @Bean
    public Jackson2JsonMessageConverter producerJackson2MessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    Queue queue() {
        return new Queue("test", false);
    }

    @Bean
    Binding binding() {
        return BindingBuilder.bind(queue()).to(exchange()).with("foo.#");
    }

    @Bean
    public SimpleRabbitListenerContainerFactory myRabbitListenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(rabbitConnectionFactory());
        factory.setMaxConcurrentConsumers(5);
        factory.setMessageConverter(producerJackson2MessageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return factory;
    }

    @Override
    public void configureRabbitListeners(final RabbitListenerEndpointRegistrar registrar) {
        registrar.setContainerFactory(myRabbitListenerContainerFactory());
    }

}

Answer 1

我现在没有时间测试它，但像这样的东西应该有用；大概你不想丢失消息，所以你需要将 ackMode 设置为 MANUAL 并自己进行确认（如图所示）。

更新

@SpringBootApplication
public class So52173111Application {

    private final ExecutorService exec = Executors.newCachedThreadPool();

    @Autowired
    private RabbitTemplate template;

    @Bean
    public ApplicationRunner runner(AsyncRabbitTemplate asyncTemplate) {
        return args -> {
            RabbitConverterFuture<Object> future = asyncTemplate.convertSendAndReceive("foo", "test");
            future.addCallback(r -> {
                System.out.println("Reply: " + r);
            }, t -> {
                t.printStackTrace();
            });
        };
    }

    @Bean
    public AsyncRabbitTemplate asyncTemplate(RabbitTemplate template) {
        return new AsyncRabbitTemplate(template);
    }

    @RabbitListener(queues = "foo")
    public void listen(String in, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag,
            @Header(AmqpHeaders.CORRELATION_ID) String correlationId,
            @Header(AmqpHeaders.REPLY_TO) String replyTo) {

        ListenableFuture<String> future = handleInput(in);
        future.addCallback(result -> {
            Address address = new Address(replyTo);
            this.template.convertAndSend(address.getExchangeName(), address.getRoutingKey(), result, m -> {
                m.getMessageProperties().setCorrelationId(correlationId);
                return m;
            });
            try {
                channel.basicAck(tag, false);
            }
            catch (IOException e) {
                e.printStackTrace();
            }
        }, t -> {
            t.printStackTrace();
        });
    }

    private ListenableFuture<String> handleInput(String in) {
        SettableListenableFuture<String> future = new SettableListenableFuture<String>();
        exec.execute(() -> {
            try {
                Thread.sleep(2000);
            }
            catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            future.set(in.toUpperCase());
        });
        return future;
    }

    public static void main(String[] args) {
        SpringApplication.run(So52173111Application.class, args);
    }

}

使用带有 Request/Reply 模式的 AsyncRabbitTemplate 时如何构建非阻塞消费者

How to build a nonblocking Consumer when using AsyncRabbitTemplate with Request/Reply Pattern

spring

rabbitmq

spring-rabbit