Spring Boot 入門之消息中間件篇(轉發）

一、前言

在消息中間件中有 2 個重要的概念：消息代理和目的地。當消息發送者發送消息后，消息就被消息代理接管，消息代理保證消息傳遞到指定目的地。

我們常用的消息代理有 JMS 和 AMQP 規范。對應地，它們常見的實現分別是 ActiveMQ 和 RabbitMQ。

上篇文章《Spring Boot 入門之緩存和 NoSQL 篇（四）》。

二、整合 ActiveMQ

2.1 添加依賴

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

<!-- 如果需要配置連接池，添加如下依賴 -->
<dependency>  
    <groupId>org.apache.activemq</groupId>  
    <artifactId>activemq-pool</artifactId>  
</dependency>

2.2 添加配置

# activemq 配置
spring.activemq.broker-url=tcp://192.168.2.12:61616
spring.activemq.user=admin
spring.activemq.password=admin
spring.activemq.pool.enabled=false
spring.activemq.pool.max-connections=50
# 使用發布/訂閱模式時，下邊配置需要設置成 true
spring.jms.pub-sub-domain=false
此處 spring.activemq.pool.enabled=false，表示關閉連接池。

2.3 編碼

配置類：

@Configuration
public class JmsConfirguration {
    public static final String QUEUE_NAME = "activemq_queue";
    
    public static final String TOPIC_NAME = "activemq_topic";
    
    @Bean
    public Queue queue() {
        return new ActiveMQQueue(QUEUE_NAME);
    }
    
    @Bean
    public Topic topic() {
        return new ActiveMQTopic(TOPIC_NAME);
    }
}

負責創建隊列和主題。

消息生產者：

@Component
public class JmsSender {
    @Autowired
    private Queue queue;
    
    @Autowired
    private Topic topic;
    
    @Autowired
    private JmsMessagingTemplate jmsTemplate;
    
    public void sendByQueue(String message) {
        this.jmsTemplate.convertAndSend(queue, message);
    }
    
    public void sendByTopic(String message) {
        this.jmsTemplate.convertAndSend(topic, message);
    }
}

消息消費者：

@Component
public class JmsReceiver {
    
    @JmsListener(destination = JmsConfirguration.QUEUE_NAME)
    public void receiveByQueue(String message) {
        System.out.println("接收隊列消息:" + message);
    }
    
    @JmsListener(destination = JmsConfirguration.TOPIC_NAME)
    public void receiveByTopic(String message) {
        System.out.println("接收主題消息:" + message);
    }
}

消息消費者使用 @JmsListener 注解監聽消息。

2.4 測試

@RunWith(SpringRunner.class)
@SpringBootTest
public class JmsTest {
    @Autowired
    private JmsSender sender;
    @Test
    public void testSendByQueue() {
        for (int i = 1; i < 6; i++) {
            this.sender.sendByQueue("hello activemq queue " + i);
        }
    }
    
    @Test
    public void testSendByTopic() {
        for (int i = 1; i < 6; i++) {
            this.sender.sendByTopic("hello activemq topic " + i);
        }
    }
}

打印結果：

接收隊列消息:hello activemq queue 1
接收隊列消息:hello activemq queue 2
接收隊列消息:hello activemq queue 3
接收隊列消息:hello activemq queue 4
接收隊列消息:hello activemq queue 5
測試發布/訂閱模式時，設置 spring.jms.pub-sub-domain=true

接收主題消息:hello activemq topic 1
接收主題消息:hello activemq topic 2
接收主題消息:hello activemq topic 3
接收主題消息:hello activemq topic 4
接收主題消息:hello activemq topic 5

三、整合 RabbitMQ

3.1 添加依賴

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

3.2 添加配置

spring.rabbitmq.host=192.168.2.30
spring.rabbitmq.port=5672
spring.rabbitmq.username=light
spring.rabbitmq.password=light
spring.rabbitmq.virtual-host=/test

3.3 編碼

配置類：

@Configuration
public class AmqpConfirguration {
    //=============簡單、工作隊列模式===============
    
    public static final String SIMPLE_QUEUE = "simple_queue";
    @Bean
    public Queue queue() {
        return new Queue(SIMPLE_QUEUE, true);
    }
    
    //===============發布/訂閱模式============
    
    public static final String PS_QUEUE_1 = "ps_queue_1";
    public static final String PS_QUEUE_2 = "ps_queue_2";
    public static final String FANOUT_EXCHANGE = "fanout_exchange";
    
    @Bean
    public Queue psQueue1() {
        return new Queue(PS_QUEUE_1, true);
    }
    
    @Bean
    public Queue psQueue2() {
        return new Queue(PS_QUEUE_2, true);
    }
    
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(FANOUT_EXCHANGE);
    }
    
    @Bean
    public Binding fanoutBinding1() {
        return BindingBuilder.bind(psQueue1()).to(fanoutExchange());
    }
    
    @Bean
    public Binding fanoutBinding2() {
        return BindingBuilder.bind(psQueue2()).to(fanoutExchange());
    }
    //===============路由模式============
    
    public static final String ROUTING_QUEUE_1 = "routing_queue_1";
    public static final String ROUTING_QUEUE_2 = "routing_queue_2";
    public static final String DIRECT_EXCHANGE = "direct_exchange";
    
    @Bean
    public Queue routingQueue1() {
        return new Queue(ROUTING_QUEUE_1, true);
    }
    
    @Bean
    public Queue routingQueue2() {
        return new Queue(ROUTING_QUEUE_2, true);
    }
    
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange(DIRECT_EXCHANGE);
    }
    
    @Bean
    public Binding directBinding1() {
        return BindingBuilder.bind(routingQueue1()).to(directExchange()).with("user");
    }
    
    @Bean
    public Binding directBinding2() {
        return BindingBuilder.bind(routingQueue2()).to(directExchange()).with("order");
    }
    
    //===============主題模式============
    
    public static final String TOPIC_QUEUE_1 = "topic_queue_1";
    public static final String TOPIC_QUEUE_2 = "topic_queue_2";
    public static final String TOPIC_EXCHANGE = "topic_exchange";
    
    @Bean
    public Queue topicQueue1() {
        return new Queue(TOPIC_QUEUE_1, true);
    }
    
    @Bean
    public Queue topicQueue2() {
        return new Queue(TOPIC_QUEUE_2, true);
    }
    
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }
    
    @Bean
    public Binding topicBinding1() {
        return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("user.add");
    }
    
    @Bean
    public Binding topicBinding2() {
        return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("user.#");
    }
    
}

RabbitMQ 有多種工作模式，因此配置比較多。想了解相關內容的讀者可以查看本站的《RabbitMQ 工作模式介紹》或者自行百度相關資料。

消息生產者：

@Component
public class AmqpSender {
    @Autowired
    private AmqpTemplate amqpTemplate;
    /**
     * 簡單模式發送
     * 
     * @param message
     */
    public void simpleSend(String message) {
        this.amqpTemplate.convertAndSend(AmqpConfirguration.SIMPLE_QUEUE, message);
    }
    /**
     * 發布/訂閱模式發送
     * 
     * @param message
     */
    public void psSend(String message) {
        this.amqpTemplate.convertAndSend(AmqpConfirguration.FANOUT_EXCHANGE, "", message);
    }
    /**
     * 路由模式發送
     * 
     * @param message
     */
    public void routingSend(String routingKey, String message) {
        this.amqpTemplate.convertAndSend(AmqpConfirguration.DIRECT_EXCHANGE, routingKey, message);
    }
    /**
     * 主題模式發送
     * 
     * @param routingKey
     * @param message
     */
    public void topicSend(String routingKey, String message) {
        this.amqpTemplate.convertAndSend(AmqpConfirguration.TOPIC_EXCHANGE, routingKey, message);
    }
}

消息消費者：

@Component
public class AmqpReceiver {
    /**
     * 簡單模式接收
     * 
     * @param message
     */
    @RabbitListener(queues = AmqpConfirguration.SIMPLE_QUEUE)
    public void simpleReceive(String message) {
        System.out.println("接收消息:" + message);
    }
    /**
     * 發布/訂閱模式接收
     * 
     * @param message
     */
    @RabbitListener(queues = AmqpConfirguration.PS_QUEUE_1)
    public void psReceive1(String message) {
        System.out.println(AmqpConfirguration.PS_QUEUE_1 + "接收消息:" + message);
    }
    @RabbitListener(queues = AmqpConfirguration.PS_QUEUE_2)
    public void psReceive2(String message) {
        System.out.println(AmqpConfirguration.PS_QUEUE_2 + "接收消息:" + message);
    }
    /**
     * 路由模式接收
     * 
     * @param message
     */
    @RabbitListener(queues = AmqpConfirguration.ROUTING_QUEUE_1)
    public void routingReceive1(String message) {
        System.out.println(AmqpConfirguration.ROUTING_QUEUE_1 + "接收消息:" + message);
    }
    @RabbitListener(queues = AmqpConfirguration.ROUTING_QUEUE_2)
    public void routingReceive2(String message) {
        System.out.println(AmqpConfirguration.ROUTING_QUEUE_2 + "接收消息:" + message);
    }
    /**
     * 主題模式接收
     * 
     * @param message
     */
    @RabbitListener(queues = AmqpConfirguration.TOPIC_QUEUE_1)
    public void topicReceive1(String message) {
        System.out.println(AmqpConfirguration.TOPIC_QUEUE_1 + "接收消息:" + message);
    }
    
    @RabbitListener(queues = AmqpConfirguration.TOPIC_QUEUE_2)
    public void topicReceive2(String message) {
        System.out.println(AmqpConfirguration.TOPIC_QUEUE_2 + "接收消息:" + message);
    }
}

消息消費者使用 @RabbitListener 注解監聽消息。

3.4 測試

@RunWith(SpringRunner.class)
@SpringBootTest
public class AmqpTest {
    @Autowired
    private AmqpSender sender;
    @Test
    public void testSimpleSend() {
        for (int i = 1; i < 6; i++) {
            this.sender.simpleSend("test simpleSend " + i);
        }
    }
    @Test
    public void testPsSend() {
        for (int i = 1; i < 6; i++) {
            this.sender.psSend("test psSend " + i);
        }
    }
    
    @Test
    public void testRoutingSend() {
        for (int i = 1; i < 6; i++) {
            this.sender.routingSend("order", "test routingSend " + i);
        }
    }
    
    @Test
    public void testTopicSend() {
        for (int i = 1; i < 6; i++) {
            this.sender.topicSend("user.add", "test topicSend " + i);
        }
    }
}

測試結果略過。。。

踩坑提醒1：ACCESS_REFUSED - Login was refused using authentication mechanism PLAIN

解決方案：

1. 請確保用戶名和密碼是否正確，需要注意的是用戶名和密碼的值是否包含空格或制表符（筆者測試時就是因為密碼多了一個制表符導致認證失敗）。

2. 如果測試賬戶使用的是 guest，需要修改 rabbitmq.conf 文件。在該文件中添加 “loopback_users = none” 配置。

踩坑提醒2：Cannot prepare queue for listener. Either the queue doesn’t exist or the broker will not allow us to use it

解決方案：

我們可以登陸 RabbitMQ 的管理界面，在 Queue 選項中手動添加對應的隊列。

四、參考資料

消息中間件簡單介紹

Spring Boot 官方文檔

Rabbit MQ 訪問控制相關

本文作者： moonlightL
本文鏈接： https://www.extlight.com/2018/01/26/Spring-Boot-入門之消息中間件篇（五）/