使用RabbitMQ的SpringBoot消息传递

RabbitMQ是流行的消息代理解决方案之一，并提供可用于各种编程语言的客户端库，包括Java，Scala，.NET，Go，Python，Ruby，PHP等。在本教程中，我们将学习如何使用RabbitMQ消息代理从SpringBoot应用程序发送和接收消息。我们还将研究如何将消息作为JSON负载发送，以及如何使用Dead Letter Queue（DLQ）处理错误。

首先，按照此处https://www.rabbitmq.com/download.html所述在本地计算机上安装RabbitMQ服务器，或者使用以下docker-compose.yml作为Docker映像运行。

version: '3'
services:rabbitmq:container_name: rabbitmqimage: 'rabbitmq:management'ports:- "5672:5672"- "15672:15672"

现在，您可以使用docker-compose启动RabbitMQ ，并在http：// localhost：15672 /启动管理UI。

如果您熟悉ActiveMQ等其他消息传递代理，则通常使用队列和主题发送一对一和发布-订阅通信模型。在RabbitMQ中，我们将邮件发送到Exchange，并根据路由密钥将邮件转发到队列。您可以在https://www.rabbitmq.com/tutorials/amqp-concepts.html上阅读有关RabbitMQ概念的更多信息。

您可以在https://github.com/sivaprasadreddy/sivalabs-blog-samples-code/tree/master/springboot-rabbitmq-demo中找到本文的源代码

RabbitMQ的SpringBoot应用程序

现在，让我们从http://start.spring.io/选择Web ， Thymeleaf和RabbitMQ启动器创建一个SpringBoot应用程序。

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.sivalabs</groupId><artifactId>springboot-rabbitmq-demo</artifactId><version>1.0-SNAPSHOT</version><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.0.0.RC1</version><relativePath/></parent><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-thymeleaf</artifactId></dependency></dependencies></project>

让我们从RabbitMQ配置开始。创建RabbitConfig配置类，并定义Queue ， Exchange和Binding Bean，如下所示：

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 QUEUE_ORDERS = "orders-queue";public static final String EXCHANGE_ORDERS = "orders-exchange";@BeanQueue ordersQueue() {return QueueBuilder.durable(QUEUE_ORDERS).build();}@BeanQueue deadLetterQueue() {return QueueBuilder.durable(QUEUE_DEAD_ORDERS).build();}@BeanExchange ordersExchange() {return ExchangeBuilder.topicExchange(EXCHANGE_ORDERS).build();}@BeanBinding binding(Queue ordersQueue, TopicExchange ordersExchange) {return BindingBuilder.bind(ordersQueue).to(ordersExchange).with(QUEUE_ORDERS);}
}

在这里，我们声明一个名称为orders-queue的队列和一个名称为orders-exchange的Exchange。
我们还定义了orders-queue和orders-exchange之间的绑定，以便将任何以routing-key作为“ orders-queue”发送到orders-exchange的消息都发送到orders-queue。

我们可以在application.properties中配置RabbitMQ服务器的详细信息，如下所示：

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

让我们创建一个Spring bean OrderMessageSender来发送消息到orders-exchange。

Spring Boot自动配置向RabbitMQ代理发送消息或从RabbitMQ代理接收消息所需的基础结构bean。我们可以简单地通过调用RabbitTemplate.convertAndSend（“ routingKey”，Object）方法自动连接RabbitTemplate并发送消息。

public class Order implements Serializable {private String orderNumber;private String productId;private double amount;//setters & getters
}

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;@Service
public class OrderMessageSender {private final RabbitTemplate rabbitTemplate;@Autowiredpublic OrderMessageSender(RabbitTemplate rabbitTemplate) {this.rabbitTemplate = rabbitTemplate;}public void sendOrder(Order order) {this.rabbitTemplate.convertAndSend(RabbitConfig.QUEUE_ORDERS, order);}
}

默认情况下，Spring Boot使用org.springframework.amqp.support.converter.SimpleMessageConverter并将对象串行化为byte [] 。

现在有了此配置，我们可以通过调用OrderMessageSender.sendOrder（Order）方法将消息发送到RabbitMQ订单队列。

发送消息后，您可以通过使用来宾/来宾凭证登录从Administration UI应用程序中查看消息。您可以单击“ 交易所 / 队列”选项卡以查看已创建的订单交换和订单队列 。您还可以检查订单交换的绑定，如下所示：

现在转到“队列”选项卡，然后单击“订单队列”。向下滚动到“ 获取消息”部分，然后单击“ 获取消息”按钮，可以查看消息的内容。

现在，使用@RabbitListener创建订单队列的侦听器。

创建一个Spring bean OrderMessageListener ，如下所示：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class OrderMessageListener {static final Logger logger = LoggerFactory.getLogger(OrderMessageListener.class);@RabbitListener(queues = RabbitConfig.QUEUE_ORDERS)public void processOrder(Order order) {logger.info("Order Received: "+order);}
}

而已！！通过简单地添加@RabbitListener并定义要监听的队列，我们可以创建一个Listener。

现在，如果您向Order-queue发送一条消息，该消息应该由OrderMessageListener.processOrder（）方法使用，并且应该看到日志语句“ Order Received：”。

以JSON有效载荷的形式发送和接收消息

如我们所见，默认的序列化机制使用SimpleMessageConverter将消息对象转换为byte []，并在接收端将使用GenericMessageConverter将byte []反序列化为Object类型（在我们的示例中为Order）。

为了更改此行为，我们需要定制Spring Boot RabbitMQ自动配置的bean。

以JSON格式发送消息

一种将消息作为JSON有效负载发送的快速方法是使用ObjectMapper，我们可以将Order对象转换为JSON并发送。

@Autowired
private ObjectMapper objectMapper;public void sendOrder(Order order) {try {String orderJson = objectMapper.writeValueAsString(order);Message message = MessageBuilder.withBody(orderJson.getBytes()).setContentType(MessageProperties.CONTENT_TYPE_JSON).build();this.rabbitTemplate.convertAndSend(RabbitConfig.QUEUE_ORDERS, message);} catch (JsonProcessingException e) {e.printStackTrace();}
}

但是像这样将对象转换为JSON是一种样板。相反，我们可以采用以下方法。

我们可以配置让RabbitTemplate使用org.springframework.amqp.support.converter.Jackson2JsonMessageConverter bean，以便将消息序列化为JSON而不是byte []。

@Configuration
public class RabbitConfig 
{......@Beanpublic RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);rabbitTemplate.setMessageConverter(producerJackson2MessageConverter());return rabbitTemplate;}@Beanpublic Jackson2JsonMessageConverter producerJackson2MessageConverter() {return new Jackson2JsonMessageConverter();}
}

现在，当您发送一条消息时，它将转换为JSON并将其发送到Queue。

以JSON格式接收消息

为了将消息有效负载视为JSON，我们应该通过实现RabbitListenerConfigurer来定制RabbitMQ配置。

@Configuration
public class RabbitConfig implements RabbitListenerConfigurer {......@Overridepublic void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {registrar.setMessageHandlerMethodFactory(messageHandlerMethodFactory());}@BeanMessageHandlerMethodFactory messageHandlerMethodFactory() {DefaultMessageHandlerMethodFactory messageHandlerMethodFactory = new DefaultMessageHandlerMethodFactory();messageHandlerMethodFactory.setMessageConverter(consumerJackson2MessageConverter());return messageHandlerMethodFactory;}@Beanpublic MappingJackson2MessageConverter consumerJackson2MessageConverter() {return new MappingJackson2MessageConverter();}
}

使用DeadLetterQueues（DLQ）处理错误和无效消息

我们可能希望将无效消息发送到单独的队列，以便以后可以检查和重新处理它们。我们可以使用DLQ概念自动执行此操作，而无需手动编写代码来处理这种情况。

我们可以在定义Queue Bean的同时声明Queue的dead-letter-exchange ， dead-letter-routing-key ：

@Configuration
public class RabbitConfig implements RabbitListenerConfigurer {public static final String QUEUE_ORDERS = "orders-queue";public static final String EXCHANGE_ORDERS = "orders-exchange";public static final String QUEUE_DEAD_ORDERS = "dead-orders-queue";@BeanQueue ordersQueue() {return QueueBuilder.durable(QUEUE_ORDERS).withArgument("x-dead-letter-exchange", "").withArgument("x-dead-letter-routing-key", QUEUE_DEAD_ORDERS).withArgument("x-message-ttl", 15000) //if message is not consumed in 15 seconds send to DLQ.build();}@BeanQueue deadLetterQueue() {return QueueBuilder.durable(QUEUE_DEAD_ORDERS).build();}......
}

现在尝试将无效的JSON消息发送到orders-queue，它将被发送到dead-orders-queue。