SpringBoot mq快速上手

1.依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2.示例代码

基础信息配置

package com.example.demo.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {public static final String EXCHANGE = "boot-EXCHANGE";public static final String QUEUE = "boot-queue";public static final String routingKey = "*.black.*";@Beanpublic Queue bootQueue(){return QueueBuilder.durable(QUEUE).build();}@Beanpublic Exchange bootExchange(){return ExchangeBuilder.topicExchange(EXCHANGE).build();}@Beanpublic Binding binding(Queue bootQueue,Exchange bootExchange){return BindingBuilder.bind(bootQueue).to(bootExchange).with(routingKey).noargs();}
}

实例配置代码如上

实操代码如下：

  rabbitTemplate.convertAndSend(EXCHANGE,"white.black.hello","nihao");System.out.println("消息发送成功");

mq确保生产者发送到交换机链路正常：

基本：

  //设置回调（保证消息发到交换机上)channel.confirmSelect();//设置回调函数channel.addConfirmListener(new ConfirmListener() {@Overridepublic void handleAck(long deliveryTag, boolean multiple) throws IOException {System.out.println("消息成功发送到交换机");}@Overridepublic void handleNack(long deliveryTag, boolean multiple) throws IOException {System.out.println("消息没有发送到交换机，请重试");}});

SpringBoot

 // publisher-confirm-type: correlated  # 新版本 spring配置rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if(ack){System.out.println("消息成功发送到交换机");}else{System.out.println("消息没有发送到交换机");}}});

该方法中的ack为true，或false代表如上，该方法会被异步调用，不会阻塞主方法，效率高，可以在这里记下来，记到数据库中，事后进行补充。。 

交换机路由到队列正常

//保证消息发到Queuechannel.addReturnListener(new ReturnCallback() {@Overridepublic void handle(Return returnMessage) {System.out.println("消息未发送到交换机，请进行相关处理");//111}});

该方法在路由队列失败时进行回调

  // publisher-returns: true # 开启Return机制rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {@Overridepublic void returnedMessage(ReturnedMessage returned) {String msg = new String(returned.getMessage().getBody());System.out.println("msg路由队列失败，请进行补救操作"+msg);}});

 该方法只会在队列路由失败时被调用，属于回调方法

设置队列持久化（不是指durable设置为true，是指mq重启后，队列依然有消息）

  //设置消息持久化AMQP.BasicProperties prop =new AMQP.BasicProperties().builder().deliveryMode(2).build();//6.发送消息channel.basicPublish(Exchange,"",prop,"nihao".getBytes(StandardCharsets.UTF_8));

设置属性，并把这个属性带上，deliveryMode设置为2即可

  rabbitTemplate.convertAndSend(EXCHANGE, "big.black.dog", "nihao", new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);return message;}});

springboot中，需要用 MessagePostProcessor里的message设置deleveryMode 并且设置为枚举类型的PERSISTENT类型，并且记得返回消息，不然无法设置队列类型为持久化

保证消息被消费者正常消费

  //4.监听消息DefaultConsumer callback=new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)throws IOException {System.out.println("消费者1好收到消息"+new String(body,"utf-8"));channel.basicAck(envelope.getDeliveryTag(),false);}};channel.basicConsume(QUEUE,false,callback);

mq在原始代码的时候需要在消费者端设置自动ack为false来进行处理，并在方法里进行手动ack。 

  listener:simple:acknowledge-mode: manual #开启手动ACKprefetch: 10 #消费者一次拿走10个消息配置如上@Component
public class Consumer {@RabbitListener(queues = RabbitConfig.QUEUE)public void Consume(String msg, Channel channel, Message message) throws IOException {System.out.println("msg"+msg);System.out.println("标示符"+message.getMessageProperties().getCorrelationId());channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}
}

springboot需要在配置文件里设置值为manual，prefetch表示一次可以取走的消息