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表示一次可以取走的消息
