Spring Boot 入門之消息中間件的使用

一、前言

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

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

二、整合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:61616spring.activemq.user=adminspring.activemq.password=adminspring.activemq.pool.enabled=falsespring.activemq.pool.max-connections=50# 使用發布/訂閱模式時，下邊配置需要設置成truespring.jms.pub-sub-domain=false

此處spring.activemq.pool.enabled=false，表示關閉連接池。

2.3 編碼

配置類：

 @Configurationpublic 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); }}

負責創建隊列和主題。

消息生產者：

 @Componentpublic 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); }}

消息消費者：

 @Componentpublic 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)@SpringBootTestpublic 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.30spring.rabbitmq.port=5672spring.rabbitmq.username=lightspring.rabbitmq.password=lightspring.rabbitmq.virtual-host=/test

3.3 編碼

配置類：

 @Configurationpublic 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 工作模式介紹》或者自行百度相關資料。

消息生產者：

 @Componentpublic 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); }}

消息消費者：

 @Componentpublic 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)@SpringBootTestpublic 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 選項中手動添加對應的隊列。

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持武林網。