RabbitMQ整合Spring AMQP实战

常用组件介绍

• RabbitAdmin

• Spring AMQP声明 通过@Bean注解进行声明

• RabbitTemplate

• SimpleMessageListenerContainer 对消息消费进行详细配置和优化

• MessageListenerAdapter 消息监听适配器，建立在监听器基础之上

• MessageConverter

RabbitAdmin

• RabbitAdmin类可以很好的操作RabbitMQ，在Spring中直接进行注入即可

• 注意：autoSatrtup必须设置为true,否则spring容器不会加载RabbitAdmin类

• RabbitAdmin底层实现就是从Spring容器中获取Exchange、Bingding、RoutingKey以及Queue的@Bean声明；

• 底层使用RabbitTemplate的execute方法执行对应的声明、修改、删除等一系列RabbitMQ基础功能操作；

RabbitMQ简单使用

pom.xml

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

配置

@Configuration
public class RabbitMqConfig1 {/*** 设置连接* @return ConnectionFactory*/@Beanpublic ConnectionFactory connectionFactory(){CachingConnectionFactory connectionFactory = new CachingConnectionFactory();connectionFactory.setHost("localhost");connectionFactory.setPort(5672);connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");connectionFactory.setVirtualHost("/");return connectionFactory;}/*** 创建RabbitAdmin* @return RabbitAdmin*/@Beanpublic RabbitAdmin rabbitAdmin() {RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory());//默认就是truerabbitAdmin.setAutoStartup(true);return rabbitAdmin;}
}

测试

 @Autowiredprivate RabbitAdmin rabbitAdmin;/*** RabbitAdmin api应用*/@Testpublic void testAdmin() {rabbitAdmin.declareExchange(new DirectExchange("test.direct", false, false));rabbitAdmin.declareExchange(new TopicExchange("test.topic", false, false));rabbitAdmin.declareExchange(new FanoutExchange("test.fanout", false, false));rabbitAdmin.declareQueue(new Queue("test.direct.queue", false));rabbitAdmin.declareQueue(new Queue("test.topic.queue", false));rabbitAdmin.declareQueue(new Queue("test.fanout.queue", false));//绑定rabbitAdmin.declareBinding(new Binding("test.direct.queue",Binding.DestinationType.QUEUE,"test.direct", "direct", new HashMap<>()));//使用 BindingBuilder 创建绑定// https://docs.spring.io/spring-amqp/docs/2.1.16.BUILD-SNAPSHOT/reference/html/#builder-apirabbitAdmin.declareBinding(BindingBuilder.bind(new Queue("test.topic.queue", false))        //直接创建队列.to(new TopicExchange("test.topic", false, false))    //直接创建交换机 建立关联关系.with("user.#"));    //指定路由Key//FanoutExchange 类型exchange不走路由键rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue("test.fanout.queue", false)).to(new FanoutExchange("test.fanout", false, false)));//清空队列数据// rabbitAdmin.purgeQueue("test.topic.queue", false);}

SpringAMQP声明(Exchange、Queue、Binding)

在RabbitMQ基础AP里面声明一个Exchange、声明一个绑定、一个队列

//基础API声明一个exchange
channel.exchangeDeclare(String exchange,String type,boolean durable,boolean autoDelete,boolean internal,Map<String, Object> arguments)
//基础API 声明一个队列
channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,Map<String, Object> arguments) 
//基础API 声明binding
channel.queueBind(String queue, String exchange, String routingKey)

使用SpringAMQP去声明，就需要使用SpringAMQP的如下模式，即声明@Bean方式

   //声明Topic 类型的exchange @Beanpublic TopicExchange topicExchange() {//exchange 持久化// Exchange springEchange = ExchangeBuilder.topicExchange("spring_amqp_test_echange").durable(true).build();return new TopicExchange("spring_amqp_test_echange", true, false);}//声明队列@Beanpublic Queue queue() {//   Queue spring_amqp_test_echange = QueueBuilder.durable("spring_amqp_test_echange").build();return new Queue("spring_amqp_test_queue");}//建立绑定@Beanpublic Binding binding(TopicExchange topicExchange, Queue queue) {return BindingBuilder.bind(queue).to(topicExchange).with("spring.*");}

消息模板 RabbitTemplate

• RabbitTemplate,即消息模板。

• 在与SpringAMQP整合的时候进行发送消息的关键类

• 该类提供了丰富的发送消息的方法，包括可靠性投递消息方法、回调监听消息接口ConfirmCallback、返回值确认接口ReturnCallback等等。同样我们需要进入注入到Spring容器中，然后直接使用；

• 在与Spring整合时需要实例化，但是在与SpringBoot整合时，在配置文件里添加配置即可；

RabbitTemplate简单使用

配置

@Configuration
public class RabbitMqConfig3 {/*** 设置连接** @return ConnectionFactory*/@Beanpublic ConnectionFactory connectionFactory() {CachingConnectionFactory connectionFactory = new CachingConnectionFactory();connectionFactory.setHost("localhost");connectionFactory.setPort(5672);connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");connectionFactory.setVirtualHost("/");return connectionFactory;}/*** 创建RabbitAdmin** @return RabbitAdmin*/@Beanpublic RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);//默认就是truerabbitAdmin.setAutoStartup(true);return rabbitAdmin;}/*** 消息模板** @param connectionFactory connectionFactory* @return RabbitTemplate*/@Beanpublic RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {return new RabbitTemplate(connectionFactory);}/*** 针对消费者配置* 1. 设置交换机类型* 2. 将队列绑定到交换机* FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念* HeadersExchange ：通过添加属性key-value匹配* DirectExchange:按照routingkey分发到指定队列* TopicExchange:多关键字匹配*/@Beanpublic TopicExchange exchange001() {return new TopicExchange("topic001", true, false);}@Beanpublic Queue queue001() {return new Queue("queue001", true); //队列持久}@Beanpublic Binding binding001(TopicExchange exchange001, Queue queue001) {return BindingBuilder.bind(queue001).to(exchange001).with("spring.*");}
}

测试

   @Testpublic void testSendMessage() {//1 创建消息//AMQP消息的消息属性//MessageBuilder(也可以构建Message) 使用流利的API从byte[]主体或其他消息构建Spring AMQP消息。MessageProperties messageProperties = new MessageProperties();messageProperties.getHeaders().put("desc", "信息描述..");messageProperties.getHeaders().put("type", "自定义消息类型..");Message message = new Message("Hello RabbitMQ".getBytes(), messageProperties);rabbitTemplate.convertAndSend("topic001", "spring.amqp", message, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {System.err.println("------添加额外的设置---------");message.getMessageProperties().getHeaders().put("desc", "额外修改的信息描述");message.getMessageProperties().getHeaders().put("attr", "额外新加的属性");return message;}});}

队列queue001

 @Test
public void testSendMessage2() throws Exception {//1 创建消息MessageProperties messageProperties = new MessageProperties();messageProperties.setContentType("text/plain");Message message = new Message("mq 消息1234".getBytes(), messageProperties);rabbitTemplate.send("topic001", "spring.abc", message);rabbitTemplate.convertAndSend("topic001", "spring.amqp", "hello object message send!");rabbitTemplate.convertAndSend("topic002", "rabbit.abc", "hello object message send!");}

队列queue001

队列queue002

简单消息监听容器：SimpleMessageListenerContainer

• 这个类非常的强大，我们可以对他进行很多设置，对于消费者的配置项，这个类都可以满足
• 监听队列（多个队列）、自动启动、自动声明功能
• 设置事务特性、事务管理器、事务属性、事务容量(并发)、是否开启事务、回滚消息等
• 设置消费者数量、最小最大数量、批量消费
• 设置消息确认和自动确认模式、是否重回队列、异常捕获handler函数
• 设置消费者标签生成策略、是否独占模式、消费者属性等
• 设置具体的监听器、消息转换器等等

注意：SimpleMessageListenerContainer可以进行动态设置，比如在运行中的应用可以动态的修改其消费者数量的大小、接收消息的模式等。很多基于RabbitMQ的自制定化后端管控台在进行动态设置的时候，也是根据这一特性去实现的。所以可以看出SpringAMQP非常的强大；
思考一下：SimpleMessageListenerContainer为什么可以动态感知配置变更？

配置

 @Beanpublic SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);// container.setQueueNames(); 接收字符串的队列名//container.setQueues(queue001(), queue002(), queue003());//当前消费者数量container.setConcurrentConsumers(1);//最大消费者数量container.setMaxConcurrentConsumers(5);//是否使用重队列container.setDefaultRequeueRejected(false);//自动签收container.setAcknowledgeMode(AcknowledgeMode.AUTO);container.setExposeListenerChannel(true);//消费端的标签策略container.setConsumerTagStrategy(new ConsumerTagStrategy() {@Overridepublic String createConsumerTag(String queue) {return queue + "_" + UUID.randomUUID().toString();}});//设置消息监听//必须设置消息监听 否则 报  No message listener specified - see property 'messageListener'container.setMessageListener(new ChannelAwareMessageListener() {@Overridepublic void onMessage(Message message, Channel channel) throws Exception {String msg = new String(message.getBody());System.err.println("----------消费者: " + msg);//做消息处理....}});return container;}

消息监听适配器：MessageListenerAdapter

通过MessageListenerAdapter的代码我们可以看出如下核心属性：

• defaultListenerMethod默认监听方法名称：用于设置监听方法名称
• Delegate委托对象：实际真实的委托对象，用于处理消息、
• queueOrTagToMethodName: 队列标识与方法名称组成的集合
• 可以一一进行队列与方法名称的匹配；
• 队列和方法名称绑定，即指定队列里的消息会被绑定的方法所接收处理；

配置

public class MessageDelegate1 {public void handleMessage(byte[] messageBody) {System.err.println("默认方法, 消息内容:" + new String(messageBody));}public void consumeMessage(byte[] messageBody) {System.err.println("字节数组方法, 消息内容:" + new String(messageBody));}public void consumeMessage(String messageBody) {System.err.println("字符串方法, 消息内容:" + messageBody);}public void method1(String messageBody) {System.err.println("method1 收到消息内容:" + new String(messageBody));}public void method2(String messageBody) {System.err.println("method2 收到消息内容:" + new String(messageBody));}public void consumeMessage(Map messageBody) {System.err.println("map方法, 消息内容:" + messageBody);}
}

  @Beanpublic SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);// container.setQueueNames(); 接收字符串的队列名//container.setQueues(queue001(), queue002(), queue003());//当前消费者数量container.setConcurrentConsumers(1);//最大消费者数量container.setMaxConcurrentConsumers(5);//是否使用重队列container.setDefaultRequeueRejected(false);//自动签收container.setAcknowledgeMode(AcknowledgeMode.AUTO);container.setExposeListenerChannel(true);//消费端的标签策略container.setConsumerTagStrategy(new ConsumerTagStrategy() {@Overridepublic String createConsumerTag(String queue) {return queue + "_" + UUID.randomUUID().toString();}});//1 适配器方式. 默认是有自己的方法名字的：handleMessage// 可以自己指定一个方法的名字: consumeMessage// 也可以添加一个转换器: 从字节数组转换为String//MessageDelegate1如何写 MessageListenerAdapter 源码里面也给出了一些建议MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate1());//默认的方法是 	public static final String ORIGINAL_DEFAULT_LISTENER_METHOD = "handleMessage";adapter.setDefaultListenerMethod("consumeMessage");//TextMessageConverter 自定义的消息转换器//new TextMessageConverter()-->consumeMessage(byte[] messageBody))->MessageProperties.setContentType("text/plian")//new Jackson2JsonMessageConverter()--->consumeMessage(Map messageBody))->MessageProperties.setContentType("application/json")// adapter.setMessageConverter(new Jackson2JsonMessageConverter());container.setMessageListener(adapter);return container;}

MessageConverter消息转换器

• 我们在进行发送消息的时候，正常情况下消息体为二进制的数据方式进行传输，如果希望内部帮我们进行转换，或者指定自定义的转换器，就需要用到MessageConverter;

• 自定义常用转换器：MessageConverter,一般来讲都需要实现这个接口

• 重写下面两个方法：

  • toMessage:java对象转换为Message
  • fromMessage:Message对象转换为java对象

• MessageConverter消息转换器：

• Json转换器：Jackson2JsonMessageConverter:可以进行java对象的转换功能；

• DefaultJackson2JavaTypeMapper映射器：可以进行java对象的映射关系；

• 自定义二进制转换器：比如图片类型、PDF、PPT、流媒体等

使用转换器的目的是当传入不同的类型的数据(如json，类，PDF，图片等)时，在消息的接收方接收到时也总是以传入的类型接收结果对象；我们通过写入不同的转换器以达到此种效果。具体可百度。

JSON格式转换

默认监听方法的参数为Map

public class Order {private String id;private String name;private String content;public Order() {}public Order(String id, String name, String content) {this.id = id;this.name = name;this.content = content;}public String getId() {return id;}public void setId(String id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}public String getContent() {return content;}public void setContent(String content) {this.content = content;}}

配置

     // 1.1 支持json格式的转换器MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());//  public void consumeMessage(Map messageBody) {//        System.err.println("map方法, 消息内容:" + messageBody);//    }//对应map参数方法adapter.setDefaultListenerMethod("consumeMessage");Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();adapter.setMessageConverter(jackson2JsonMessageConverter);container.setMessageListener(adapter);

测试

  @Testpublic void testSendJsonMessage() throws Exception {Order order = new Order();order.setId("001");order.setName("消息订单");order.setContent("描述信息");ObjectMapper mapper = new ObjectMapper();String json = mapper.writeValueAsString(order);System.err.println("order 4 json: " + json);MessageProperties messageProperties = new MessageProperties();//这里注意一定要修改contentType为 application/jsonmessageProperties.setContentType("application/json");Message message = new Message(json.getBytes(), messageProperties);rabbitTemplate.send("topic001", "spring.order", message);}

JSON格式转换支持Java对象

默认监听方法的参数为Java对象

委托对象方法

public void consumeMessage(Order order) {System.err.println("order对象, 消息内容, id: " + order.getId() + ", name: " + order.getName() + ", content: "+ order.getContent());}

配置

     // 1.2 DefaultJackson2JavaTypeMapper & Jackson2JsonMessageConverter 支持java对象转换MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());adapter.setDefaultListenerMethod("consumeMessage");Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();DefaultJackson2JavaTypeMapper javaTypeMapper = new DefaultJackson2JavaTypeMapper();//信任所有的包，否则会报 报不信任javaTypeMapper.setTrustedPackages("*");jackson2JsonMessageConverter.setJavaTypeMapper(javaTypeMapper);adapter.setMessageConverter(jackson2JsonMessageConverter);container.setMessageListener(adapter);

测试

 @Testpublic void testSendJavaMessage() throws Exception {Order order = new Order();order.setId("001");order.setName("订单消息");order.setContent("订单描述信息");ObjectMapper mapper = new ObjectMapper();String json = mapper.writeValueAsString(order);System.err.println("order 4 json: " + json);MessageProperties messageProperties = new MessageProperties();//这里注意一定要修改contentType为 application/jsonmessageProperties.setContentType("application/json");//__TypeId__ 这个是固定写法messageProperties.getHeaders().put("__TypeId__", "com.niugang.spring.entity.Order");Message message = new Message(json.getBytes(), messageProperties);rabbitTemplate.send("topic001", "spring.order", message);}

输出

order对象, 消息内容, id: 001, name: 订单消息, content: 订单描述信息

JSON格式转换支持Java对象(二)

委托对象方法

	public void consumeMessage(Order order) {System.err.println("order对象, 消息内容, id: " + order.getId() + ", name: " + order.getName() + ", content: "+ order.getContent());}public void consumeMessage(Packaged pack) {System.err.println("package对象, 消息内容, id: " + pack.getId() + ", name: " + pack.getName() + ", content: "+ pack.getDescription());}

配置

 //1.3 DefaultJackson2JavaTypeMapper & Jackson2JsonMessageConverter 支持java对象多映射转换MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());adapter.setDefaultListenerMethod("consumeMessage");Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();DefaultJackson2JavaTypeMapper javaTypeMapper = new DefaultJackson2JavaTypeMapper();Map<String, Class<?>> idClassMapping = new HashMap<String, Class<?>>();idClassMapping.put("order", com.niugang.spring.entity.Order.class);idClassMapping.put("packaged", com.niugang.spring.entity.Packaged.class);javaTypeMapper.setIdClassMapping(idClassMapping);jackson2JsonMessageConverter.setJavaTypeMapper(javaTypeMapper);adapter.setMessageConverter(jackson2JsonMessageConverter);container.setMessageListener(adapter);

测试

   @Testpublic void testSendMappingMessage() throws Exception {ObjectMapper mapper = new ObjectMapper();Order order = new Order();order.setId("001");order.setName("订单消息");order.setContent("订单描述信息");String json1 = mapper.writeValueAsString(order);System.err.println("order 4 json: " + json1);MessageProperties messageProperties1 = new MessageProperties();//这里注意一定要修改contentType为 application/jsonmessageProperties1.setContentType("application/json");messageProperties1.getHeaders().put("__TypeId__", "order");Message message1 = new Message(json1.getBytes(), messageProperties1);rabbitTemplate.send("topic001", "spring.order", message1);Packaged pack = new Packaged();pack.setId("002");pack.setName("包裹消息");pack.setDescription("包裹描述信息");String json2 = mapper.writeValueAsString(pack);System.err.println("pack 4 json: " + json2);MessageProperties messageProperties2 = new MessageProperties();//这里注意一定要修改contentType为 application/jsonmessageProperties2.setContentType("application/json");messageProperties2.getHeaders().put("__TypeId__", "packaged");Message message2 = new Message(json2.getBytes(), messageProperties2);rabbitTemplate.send("topic001", "spring.pack", message2);}

全局消息转化器与自定义转化器

自定义文本转化器

public class TextMessageConverter implements MessageConverter {@Overridepublic Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {return new Message(object.toString().getBytes(), messageProperties);}@Overridepublic Object fromMessage(Message message) throws MessageConversionException {String contentType = message.getMessageProperties().getContentType();if(null != contentType && contentType.contains("text")) {return new String(message.getBody());}return message.getBody();}}

自定义图片转化器

/*** 图片转化器*/
public class ImageMessageConverter implements MessageConverter {@Overridepublic Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {throw new MessageConversionException(" convert error ! ");}@Overridepublic Object fromMessage(Message message) throws MessageConversionException {System.err.println("-----------Image MessageConverter----------");Object _extName = message.getMessageProperties().getHeaders().get("extName");String extName = _extName == null ? "png" : _extName.toString();byte[] body = message.getBody();String fileName = UUID.randomUUID().toString();//目录必须存在String path = "d:/springbootlog/" + fileName + "." + extName;File f = new File(path);try {//拷贝到指定路径Files.copy(new ByteArrayInputStream(body), f.toPath());} catch (IOException e) {e.printStackTrace();}return f;}}

自定义pdf转化器

public class PDFMessageConverter implements MessageConverter {@Overridepublic Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {throw new MessageConversionException(" convert error ! ");}@Overridepublic Object fromMessage(Message message) throws MessageConversionException {System.err.println("-----------PDF MessageConverter----------");byte[] body = message.getBody();String fileName = UUID.randomUUID().toString();String path = "d:/springbootlog/" + fileName + ".pdf";File f = new File(path);try {Files.copy(new ByteArrayInputStream(body), f.toPath());} catch (IOException e) {e.printStackTrace();}return f;}}

委托对象

    public void consumeMessage(File file) {System.err.println("文件对象 方法, 消息内容:" + file.getName());}

配置

     //1.4 ext convertMessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());adapter.setDefaultListenerMethod("consumeMessage");//全局的转换器:ContentTypeDelegatingMessageConverter convert = new ContentTypeDelegatingMessageConverter();TextMessageConverter textConvert = new TextMessageConverter();convert.addDelegate("text", textConvert);convert.addDelegate("html/text", textConvert);convert.addDelegate("xml/text", textConvert);convert.addDelegate("text/plain", textConvert);Jackson2JsonMessageConverter jsonConvert = new Jackson2JsonMessageConverter();convert.addDelegate("json", jsonConvert);convert.addDelegate("application/json", jsonConvert);ImageMessageConverter imageConverter = new ImageMessageConverter();convert.addDelegate("image/png", imageConverter);convert.addDelegate("image", imageConverter);PDFMessageConverter pdfConverter = new PDFMessageConverter();convert.addDelegate("application/pdf", pdfConverter);adapter.setMessageConverter(convert);container.setMessageListener(adapter);

测试

@Testpublic void testSendExtConverterMessage() throws Exception {byte[] body = Files.readAllBytes(Paths.get("C:\\Users\\Administrator\\Desktop\\公众号", "spring.png"));MessageProperties messageProperties = new MessageProperties();messageProperties.setContentType("image/png");messageProperties.getHeaders().put("extName", "png");Message message = new Message(body, messageProperties);rabbitTemplate.send("", "image_queue", message);byte[] body1 = Files.readAllBytes(Paths.get("D:\\Documents\\技术书籍", "Java huashan-2019-06-20.pdf"));MessageProperties messageProperties1 = new MessageProperties();messageProperties.setContentType("application/pdf");Message message1 = new Message(body1, messageProperties);rabbitTemplate.send("", "pdf_queue", message1);}

SpringBoot整合配置详解(生产端)

• publisher-confirms,实现一个监听器用于监听Broker端给我们返回的确认请求：RabbitTemplate.ConfirmCallback
• publisher-returns,保证消息对Broker端是可达的，如果出现路由键不可达的情况，则使用监听器对不可达的消息进行后续的处理，保证消息的路由成功： RabbitTemplate.ReturnCallback

注意一点，在发送消息的时候对template进行配置mandatory=true保证监听有效；生产端还可以配置其他属性，比如发送重试，超时时间、次数、间隔等。

生产端代码示例

application.properties

spring.rabbitmq.addresses=localhost:5672
#spring.rabbitmq.host=localhost
#spring.rabbitmq.port=5762
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/
spring.rabbitmq.connection-timeout=15000# 消息确认模式
spring.rabbitmq.publisher-confirms=true
# 消息返回模式
spring.rabbitmq.publisher-returns=true
# 为true 消息返回模式才生效
spring.rabbitmq.template.mandatory=true

配置

/*** springboot  消息生产者** @author niugang*/
@Configuration
public class RabbitMqConfig {/*** 自动注入RabbitTemplate模板类*/@Autowiredprivate RabbitTemplate rabbitTemplate;/*** 回调函数: confirm确认*/final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {System.err.println("correlationData: " + correlationData);System.err.println("ack: " + ack);if (!ack) {System.err.println("异常处理....");}}};/*** 回调函数: return返回*/final ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {@Overridepublic void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,String exchange, String routingKey) {System.err.println("return exchange: " + exchange + ", routingKey: "+ routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);}};/*队列监听在消费者端配置，没有将会自动创建@Beanpublic TopicExchange topicExchange() {return new TopicExchange("exchange-1");}@Beanpublic Queue queue() {return new Queue("queue-1");}@Beanpublic Binding binding(Queue queue, TopicExchange topicExchange) {return BindingBuilder.bind(queue).to(topicExchange).with("springboot.#");}*//*** 发送消息方法调用: 构建Message消息** @param message    消息体* @param properties 消息属性*/public void send(Object message, Map<String, Object> properties) {MessageProperties messageProperties = new MessageProperties();if (properties != null && properties.size() > 0) {Set<Map.Entry<String, Object>> entries = properties.entrySet();for (Map.Entry<String, Object> entry : entries) {String key = entry.getKey();Object value = entry.getValue();messageProperties.setHeader(key, value);}}//org.springframework.amqp.coreMessage msg = MessageBuilder.withBody(message.toString().getBytes()).andProperties(messageProperties).build();//id + 时间戳 全局唯一CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());rabbitTemplate.setConfirmCallback(confirmCallback);rabbitTemplate.setReturnCallback(returnCallback);//routingKey修改 为 spring.abc 消息将走 returnCallbackrabbitTemplate.convertAndSend("exchange-1", "springboot.abc", msg, correlationData);}}

测试

在rabbitmq控制台新建，Exchange名为exchange-1，新建队列queue-1,并建立两者之间的绑定，routingKey为springboot.#

@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTests {@Testpublic void contextLoads() {}@Autowiredprivate RabbitMqConfig rabbitMqConfig ;private static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");@Testpublic void testSender1() throws Exception {Map<String, Object> properties = new HashMap<>();properties.put("number", "12345");properties.put("send_time", simpleDateFormat.format(new Date()));rabbitMqConfig.send("Hello RabbitMQ For Spring Boot!"+System.currentTimeMillis(), properties);}}

注意：进行单元测试，ack一直是false;改为url请求，ack就正常了

SpringBoot整合配置详解(消费端)

消费端核心配置

#    NONE, MANUAL, AUTO;  手工消息消息确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual
#监听器调用程序线程的最小数量。
spring.rabbitmq.listener.simple.concurrency=5
#监听器调用程序线程的最大数量。
spring.rabbitmq.listener.simple.max-concurrency=10
# spring.rabbitmq.listener.type=simple 默认为 SimpleContainer  模式对应  spring.rabbitmq.listener.simple 前缀相关的

注意点

• 首先配置手工确认模式，用于ACK的手工处理，这样我们可以保证消息的可靠性送达，或者在消费端消费失败的时候可以做到重回队列、根据业务记录日志等处理
• 可以设置消费端的监听个数和最大个数，用于控制消费端的并发情况。

@RabbitListener注解的使用

• 消费端监听@RabbitMQListener注解，这个对于在实际工作中非常的好用。
• @RabbitListener是一个组合注解，里面可以注解配置
• @QueueBinding、@Queue、@Exchange直接通过这个组合注解一次性搞定消费端交换机、队列、绑定、路由、并且配置监听功能等。

消费者端代码示例

类配置写在代码里非常不友好，所以强烈建议大家使用配置文件配置。

properties

#spring.rabbitmq.addresses=localhost:5672
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/
spring.rabbitmq.connection-timeout=15000
#    NONE, MANUAL, AUTO;  手工消息消息确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual
#监听器调用程序线程的最小数量。
spring.rabbitmq.listener.simple.concurrency=5
#监听器调用程序线程的最大数量。
spring.rabbitmq.listener.simple.max-concurrency=10
# spring.rabbitmq.listener.type=simple 默认为 SimpleContainer  模式对应  spring.rabbitmq.listener.simple 前缀相关的
spring.rabbitmq.listener.order.queue.name=queue-2
spring.rabbitmq.listener.order.queue.durable=true
spring.rabbitmq.listener.order.exchange.name=exchange-2
spring.rabbitmq.listener.order.exchange.durable=true
spring.rabbitmq.listener.order.exchange.type=topic
spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true
spring.rabbitmq.listener.order.key=springboot.*

配置

public class Order implements Serializable {private String id;private String name;public Order() {}public Order(String id, String name) {super();this.id = id;this.name = name;}public String getId() {return id;}public void setId(String id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}}

/*** 消费者类** @author niugang*/
@Configuration
public class RabbitMQReceiver {/*** 从1.5.0版开始，您可以在类级别指定@RabbitListener注释。* 与新的@RabbitHandler批注一起，这使单个侦听器可以根据传入消息的有效负载类型调用不同的方法。** @RabbitListener(id="multi", queues = "someQueue")* @SendTo("my.reply.queue") public class MultiListenerBean {* @RabbitHandler public String thing2(Thing2 thing2) {* ...* }* @RabbitHandler public String cat(Cat cat) {* ...* }* @RabbitHandler public String hat(@Header("amqp_receivedRoutingKey") String rk, @Payload Hat hat) {* ...* }* @RabbitHandler(isDefault = true)* public String defaultMethod(Object object) {* ...* }* }* 在这种情况下，如果转换后的有效负载是Thing2，Cat或Hat，则会调用各个@RabbitHandler方法。* 您应该了解，系统必须能够根据有效负载类型识别唯一方法。* 检查该类型是否可分配给没有注释或带有@Payload注释的单个参数。* 请注意，如方法级别@RabbitListener（前面所述）中所述，应用了相同的方法签名。*///队列 exchange  绑定 没有 自动创建@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "queue-1",durable = "true"),exchange = @Exchange(value = "exchange-1",durable = "true",type = ExchangeTypes.TOPIC,ignoreDeclarationExceptions = "true"),key = "springboot.*" //routing key))@RabbitHandler//@RabbitListener 提供了很多灵活的签名 如Message Channel  @Payload  @Header 等 具体可查看源码// org.springframework.amqp.core.Message// org.springframework.messaging.Messagepublic void onMessage(Message message, Channel channel) throws Exception {System.err.println("--------------------------------------");System.err.println("消费端Payload: " + new String(message.getBody()));System.err.println("消费端MessageProperties.: " + message.getMessageProperties());//AmqpHeaders header属性封装//手工ACKchannel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}/*** spring.rabbitmq.listener.order.queue.name=queue-2* spring.rabbitmq.listener.order.queue.durable=true* spring.rabbitmq.listener.order.exchange.name=exchange-1* spring.rabbitmq.listener.order.exchange.durable=true* spring.rabbitmq.listener.order.exchange.type=topic* spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true* spring.rabbitmq.listener.order.key=springboot.*** @param order   order* @param channel channel* @param headers headers* @throws Exception Exception*/@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "${spring.rabbitmq.listener.order.queue.name}",durable = "${spring.rabbitmq.listener.order.queue.durable}"),exchange = @Exchange(value = "${spring.rabbitmq.listener.order.exchange.name}",durable = "${spring.rabbitmq.listener.order.exchange.durable}",type = "${spring.rabbitmq.listener.order.exchange.type}",ignoreDeclarationExceptions = "${spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions}"),key = "${spring.rabbitmq.listener.order.key}"))@RabbitHandler//@Headers 必须通过Map接收//@Header("amqp_receivedRoutingKey") String rk 直接获取header中某一个key//默认前缀为amqp_/*** {amqp_receivedDeliveryMode=PERSISTENT,* amqp_receivedExchange=exchange-2,* amqp_deliveryTag=1,* amqp_consumerQueue=queue-2,* amqp_redelivered=false,amqp_receivedRoutingKey=springboot.def,spring_listener_return_correlation=175a21c4-ffd5-4a3e-ac3a-2f63d60c18a5,spring_returned_message_correlation=0987654321,id=53443ced-0b23-3079-71c2-09997897a553,amqp_consumerTag=amq.ctag-V0hqyVObrHXJeC60MwPSVQ,contentType=application/x-java-serialized-object,timestamp=1591240122842}*/public void onOrderMessage(@Payload Order order,Channel channel,@Headers Map<String, Object> headers) throws Exception {System.err.println("--------------------------------------");System.err.println("消费端order: " + order.getId());System.err.println("消费端headers: " + headers);Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);//手工ACKchannel.basicAck(deliveryTag, false);}}