消息队列06SpringBoot搭建Rabbitmq投递确认和消费应答

gralde

compile "org.springframework.cloud:spring-cloud-starter-bus-amqp"

application.properties

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.prod.port=5674
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
spring.rabbitmq.virtual-host=/

#投递确认
spring.rabbitmq.publisher-cOnfirms=true
# 发送回调
spring.rabbitmq.publisher-returns=true
# 消费手动确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.direct.acknowledge-mode=manual
#消费者最小数量
spring.rabbitmq.listener.simple.cOncurrency=1
#消费之最大数量
spring.rabbitmq.listener.simple.max-cOncurrency=10
#在单个请求中处理的消息个数，他应该大于等于事务数量(unack的最大数量)
spring.rabbitmq.listener.simple.prefetch=1

 

代码

消费者

@Component
public class DirectReceiver {


    @RabbitListener(bindings = @QueueBinding(//绑定队列和交换机
            value = @Queue(value = "Direct-A"),//创建queuqe
            exchange = @Exchange(value = "DirectExchange",type = "direct"),//创建交换机
            key = "Direct-A"//路由规则，routingkey如果是Direct-A就发到这个监听

    ))
    @RabbitHandler
    public  void directReceiverA(Message message, Channel channel) throws Exception{
        System.out.println("directReceiverA:"+message.toString() );
        System.out.println("directReceiverA:"+message.getMessageProperties().getAppId() );
        System.out.println("directReceiverA:"+message.getMessageProperties().getMessageId() );
        System.out.println("directReceiverA:"+message.getMessageProperties().getReceivedExchange() );
        System.out.println("directReceiverA:"+message.getMessageProperties().getReceivedRoutingKey() );
        System.out.println("directReceiverA:"+message.getMessageProperties().getDeliveryTag() );
        System.out.println("directReceiverA:"+message.getMessageProperties().getHeaders() );
        //睡眠10秒，10秒后再应答消息队列
        Thread.sleep(10000);
        //应答消息队列
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

    }
}

生产者

@Component
public class DirectSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void send() throws Exception  {
        CorrelationData correlatiOnData= new CorrelationData(UUID.randomUUID().toString());
        String cOntext= "hi, i am DirectSender  A message  ";
        this.rabbitTemplate.convertAndSend("DirectExchange", "Direct-A", context,correlationData);
        //设置超时时间
        boolean isAck =   correlationData.getFuture().get(1, TimeUnit.MINUTES).isAck();
        if(isAck){
            System.out.println("消息已经投递:"+correlationData.getId());
        }else{
            System.out.println("消息没有被投递:"+correlationData.getId());
        }

    }

}

 

消息发送出去之后显示已经投递

消费者已经消费了

但是线程休眠，从管理界面可以看到消息一致 处于unacked的状态

10秒后Unacked的状态已经不见