逻辑

Producer 发送数据, 并获取异步 Future 对象, 并设置回调对象

        AsyncRabbitTemplate.RabbitConverterFuture<Object> future = asyncRabbitTemplate.convertSendAndReceive("hello", "world " + in.incrementAndGet());
        future.addCallback(new ListenableFutureCallback<Object>() {
            @Override
            public void onFailure(Throwable ex) {
                ex.printStackTrace();
            }

            @Override
            public void onSuccess(Object result) {
                System.out.println("回调收到结果=> " + result);
            }
        });

• 第一个参数是队列名
• 第二个参数是请求数据

Consumer 处理数据, 返回结果

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

        return "处理完毕 hello " + data + ", request id " + correlationId;
    }

代码

import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

@EnableScheduling
@SpringBootApplication
public class DemoApplication implements CommandLineRunner {

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


    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Bean
    public AsyncRabbitTemplate asyncRabbitTemplate() {
        SimpleMessageListenerContainer smc = new SimpleMessageListenerContainer(rabbitTemplate.getConnectionFactory());
        smc.setQueueNames("replay-queue");
        AsyncRabbitTemplate async = new AsyncRabbitTemplate(rabbitTemplate, smc);
        return async;
    }

    @Autowired
    private AsyncRabbitTemplate asyncRabbitTemplate;

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

        return "处理完毕 hello " + data + ", request id " + correlationId;
    }

    private AtomicInteger in = new AtomicInteger();

    @Scheduled(cron = "*/2 * * * * * ")
    private void async() throws ExecutionException, InterruptedException {

        AsyncRabbitTemplate.RabbitConverterFuture<Object> future = asyncRabbitTemplate.convertSendAndReceive("hello", "world " + in.incrementAndGet());
        future.addCallback(new ListenableFutureCallback<Object>() {
            @Override
            public void onFailure(Throwable ex) {
                ex.printStackTrace();
            }

            @Override
            public void onSuccess(Object result) {
                System.out.println("回调收到结果=> " + result);
            }
        });
    }

    @Override
    public void run(String... args) throws Exception {

    }
}