Spring Boot 集成RabbitMQ

此处假设已经安装好了RabbitMQ，主要讲述使用Spring Boot如何集成RabbitMQ。

添加依赖

在Maven的pom.xml添加rabbitmq的starter依赖，内容如下：

<?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>
    </dependencies>

</project>

RabbitMQ配置

RabbitMQ需要配置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";

    @Bean
    Queue ordersQueue() {
        return QueueBuilder.durable(QUEUE_ORDERS).build();
    }

    @Bean
    Queue deadLetterQueue() {
        return QueueBuilder.durable(QUEUE_DEAD_ORDERS).build();
    }

    @Bean
    Exchange ordersExchange() {
        return ExchangeBuilder.topicExchange(EXCHANGE_ORDERS).build();
    }

    @Bean
    Binding binding(Queue ordersQueue, TopicExchange ordersExchange) {
        return BindingBuilder.bind(ordersQueue).to(ordersExchange).with(QUEUE_ORDERS);
    }
}

例子里定义了名为orders-queue的队列和名为orders-exchange的Exchange，以及定义了orders-queue与orders-exchange的绑定，其中设定router key为orders-queue。任何以orders-queue为router key都会发送到队列orders-queue。

在application.properties配置RabbitMQ服务器信息：

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

环境已经配置好，接着是应用。应用分为两部分：发送方OrderMessageSender 和接收方OrderMessageListener 。

消息发送OrderMessageSender

这里新建一个Bean OrderMessageSender作为发送方，它将用来发送消息到orders-exchange。

模型Order

public class Order implements Serializable {
    private String orderNumber;
    private String productId;
    private double amount;

    //setters & getters
}

用于传递数据的数据模型需要序列化。

RabbitTemplate 

OrderMessageSender主要是使用RabbitTemplate来发送消息。

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;

    @Autowired
    public OrderMessageSender(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void sendOrder(Order order) {
        this.rabbitTemplate.convertAndSend(RabbitConfig.QUEUE_ORDERS, order);
    }
}

调用rabbitTemplate.convertAndSend()发送消息，它接收两个参数：

• routerKey：第一个参数路由的键，此router key为配置里定义绑定时设置的值。
• data：第二个参数为发送消息载体

默认情况下，Spring Boot使用org.springframework.amqp.support.converter.SimpleMessageConverter来序列化数据为字节码。

监听消息OrderMessageListener

接着是在消息消费端实现消息的监听，以便接收消息。这里订单消息监听类为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，并指定监听的队列。

至此就完成了一个简单消息的发送和接收。