SpringBoot整合RabbitMQ,常用操作

本文介绍三种常用操作,基于spring-boot-starter-amqp依赖

手动ack
work模式(能者多劳)
消息格式转换

手动ack

消息确认模式

在amqp协议中消息确认有两种模式

自动确认模式（automatic acknowledgement model）当消息代理将消息发送给应用后立即删除
显式确认模式（explicit acknowledgement model）待应用发送一个确认回执后再删除消息

而在spring-boot-starter-amqp,spring定义了三种

NONE 没有ack的意思,对应rabbitMQ的自动确认模式
MANUAL 手动模式,对应rabbitMQ的显式确认模式
AUTO 自动模式,对应rabbitMQ的显式确认模式

首先注意的是spring-amqp中的自动模式与rabbit中的自动模式是不一样的,其次,在spring-amqp中MANUAL 与 AUTO的关系有点类似于在spring中手动提交事务与自动提交事务的区别,一个是手动发送ack一个是在方法执行完,没有异常的情况下自动发送ack

代码实现

三个步骤

设置消费者的消息确认模式
手动确认/拒绝消息
设置消息拒绝策略

设置消费者的消息确认模式:

@Configuration
public class ListenerConfig {

   @Bean("myListenerFactory")
    public RabbitListenerContainerFactory myFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory containerFactory= 
                new SimpleRabbitListenerContainerFactory();
        containerFactory.setConnectionFactory(connectionFactory);
        //设置消费者的消息确认模式
        containerFactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return containerFactory;
    }
}

手动确认/拒绝消息:

@Component
@RabbitListener(
        containerFactory = "myListenerFactory",
        bindings = @QueueBinding(
            value = @Queue("myManualAckQueue"),
            exchange = @Exchange(value = "myManualAckExchange", type = ExchangeTypes.DIRECT),
            key = "mine.manual"))
public class MyAckListener {

    @RabbitHandler
    public void onMessage(@Payload String msg, 
                          @Headers Map<String, Object> headers, 
                          Channel channel) throws Exception{
        try {
            System.out.println(msg);
            //消息确认,(deliveryTag,multiple是否确认所有消息)
            channel.basicAck((Long) headers.get(AmqpHeaders.DELIVERY_TAG), false);
        } catch (Exception e) {
            //消息拒绝(deliveryTag,multiple,requeue拒绝后是否重新回到队列)
            channel.basicNack((Long) headers.get(AmqpHeaders.DELIVERY_TAG), false, false);
            // 拒绝一条
            // channel.basicReject();
        }
    }
}

设置消息拒绝策略:

拒绝策略是指,当消息被消费者拒绝时该如何处理,丢弃或者是重新回到队列.

在MANUAL 模式下,在拒绝消息的方法中设置

//消息拒绝(deliveryTag,multiple,requeue拒绝后是否重新回到队列)
channel.basicNack((Long) headers.get(AmqpHeaders.DELIVERY_TAG), false, false);

在AUTO 模式下可通过RabbitListenerContainerFactory或是ListenerContainer设置,如

@Bean("myListenerFactory")
    public RabbitListenerContainerFactory myFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory containerFactory=
                new SimpleRabbitListenerContainerFactory();
        containerFactory.setConnectionFactory(connectionFactory);
        //自动ack
        containerFactory.setAcknowledgeMode(AcknowledgeMode.AUTO);
        //拒绝策略,true回到队列 false丢弃
        containerFactory.setDefaultRequeueRejected(false);
        return containerFactory;
    }

需要注意的是,默认的拒绝策略是回到队列,所以,如果队列只有一个消费者的话就会产生死循环

work模式-能者多劳

默认情况下,如果有多个消费者在一个队列上,消息是公平的分发给消费者的,一人一个轮着来,不考虑每个消费者之间的处理能力的差异,这可以通过设置预处理消息数(prefetchCount)缓解,或是使用work-能者多劳模式

work-能者多劳模式: 每个消费者的预处理消息数(prefetchCount)都设置为1,每个消费者消息确认都为显式确认模式,即MANUAL,或是AUTO

如下,两个消费者消费同一个queue上的消息,理论上consumer-one处理能力是consumer-two的两倍

@Component
public class WorkListener {
    private int one = 1;
    private int two = 1;

    @RabbitListener(containerFactory = "workListenerFactory",
            queuesToDeclare = @Queue("workQueue"))
    public void onMessageOne(String msg) throws InterruptedException {
        Thread.sleep(100);
        System.out.println("consumer-one 第 " + one + " 个消息 :" + msg);
        one++;
    }

    @RabbitListener(containerFactory = "workListenerFactory",
            queuesToDeclare = @Queue("workQueue"))
    public void onMessageTwo(String msg) throws InterruptedException {
        Thread.sleep(200);
        System.out.println("consumer-two 第 " + two + " 个消息 :" + msg);
        two++;
    }
}

生产者,使用了上一篇中介绍的默认交换机

    @Autowired
    private RabbitTemplate rabbitTemplate;

    private void send() {
        for (int i = 0; i < 100; i++) {
            rabbitTemplate.convertAndSend("workQueue", "this is a message");
        }
    }

执行结果如下,符合预期,两个消费者几乎同时消费完毕,且one消费的消息数是two的两倍

......
consumer-two 第 31 个消息 :this is a message
consumer-one 第 62 个消息 :this is a message
consumer-one 第 63 个消息 :this is a message
consumer-two 第 32 个消息 :this is a message
consumer-one 第 64 个消息 :this is a message
consumer-one 第 65 个消息 :this is a message
consumer-two 第 33 个消息 :this is a message
consumer-one 第 66 个消息 :this is a message
consumer-two 第 34 个消息 :this is a message

消息格式转换

rabbirMQ中的消息对应到java中对应的实体类是 org.springframework.amqp.core.Message,所以消息转换接口MessageConverter 有两个主要方法 toMessage 和 fromMessage 顾名思义,即将发送的内容与Message的互转

SimpleMessageConverter

spring中默认使用的是 SimpleMessageConverter 它的两个转化方法如下

toMessage,根据 object instanceof xxx 转化

toMessage.png

fromMessage,根据MessageProperties的ContentType转换

fromMessage.png

所以你大可以自己实现MessageConverter 接口自己转换,当然spring也提供了常用的转化,如转json,xml

Jackson2JsonMessageConverter

常用的将object与json互转

生产者

@Autowired
private RabbitTemplate rabbitTemplate;

private void send() {
    //实际项目不建议这么干,spring单例模式,
    // 所以最好自己构建一个"jasonRabbitTemplate",用的使用注入
    rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
    rabbitTemplate.convertAndSend("jsonQueue", new Student("zhangSan",15,"男"));
}

消费者

@Bean("jasonTemplate")
public RabbitTemplate jasonRabbitTemplate(ConnectionFactory connectionFactory) {
    Jackson2JsonMessageConverter messageConverter = 
        new Jackson2JsonMessageConverter();
    RabbitTemplate rabbitTemplate = new RabbitTemplate();
    rabbitTemplate.setConnectionFactory(connectionFactory);
    //设置转化类
    rabbitTemplate.setMessageConverter(messageConverter);
    return rabbitTemplate;
}

...

@Component
@RabbitListener(containerFactory = "jsonListenerFactory",
                queuesToDeclare = @Queue("jsonQueue"))
public class JasonListener {

    @RabbitHandler
    public void onMessage(Student student) {
        System.out.println(student);
    }
}

消息内容:

json.png

转载请注明出处
系列文章