此篇教程以Rabbitmq作為消息隊列服務端，使用Spring Boot產生和發布消息。

?

使用Spring AMQP的RabbitTemplate發布消息，使用MessageListenerAdapter訂閱消息。

?

其中對應的Maven如下：

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.4.0</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>cn.it1995</groupId><artifactId>demo</artifactId><version>0.0.1-SNAPSHOT</version><name>demo</name><description>Demo project for Spring Boot</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-amqp</artifactId><version>2.3.1</version><scope>compile</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>2.3.1</version><scope>compile</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

下面是創建接受者，獲取發布者的發布的消息：

package cn.it1995.demo;import org.springframework.stereotype.Component;import java.util.concurrent.CountDownLatch;@Component public class Receiver {private CountDownLatch latch = new CountDownLatch(1);public void receiveMessage(String message){System.out.println("Received <" + message + ">");latch.countDown();}public CountDownLatch getLatch(){return latch;} }

這個Receiver是POJO類，這里的POJO是指沒有帶有業務處理的類，當注冊后，就能在任意的地方進使用。

CountDownLatch，里面會有一個接收消息的信號。

?

注冊監聽者以及發送消息

使用Spring AMQP的RabbitTemplate提供的函數用于接收消息，邏輯如下：

1. 配置消息監聽者容器；

2.聲明隊列，交換機對其都進行綁定；

3.配置發送者組建，使得監聽者能接收到。

代碼如下：

package cn.it1995.demo;import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.TopicExchange; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean;@SpringBootApplication public class DemoApplication {static final String topicExchangeName = "spring-boot-exchange";static final String queueName = "spring-boot";@BeanQueue queue(){return new Queue(queueName, false);}@BeanTopicExchange exchange(){return new TopicExchange(topicExchangeName);}@BeanBinding binding(Queue queue, TopicExchange exchange){return BindingBuilder.bind(queue).to(exchange).with("foo.bar.#");}@BeanSimpleMessageListenerContainer container(ConnectionFactory connectionFactory,MessageListenerAdapter listenerAdapter){SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.setQueueNames(queueName);container.setMessageListener(listenerAdapter);return container;}@BeanMessageListenerAdapter listenerAdapter(Receiver receiver){return new MessageListenerAdapter(receiver, "receiveMessage");}public static void main(String[] args) {SpringApplication.run(DemoApplication.class, args);}}

listenerAdapter()方法將消息監聽者注冊到容器中（默認是在container()）。

queue()方法創建了AMQP的隊列。exchange()方法創建了topic交換機。

binding()方法將其進行綁定。

?

發送測試數據到Rabbitmq

package cn.it1995.demo;import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.boot.CommandLineRunner; import org.springframework.stereotype.Component;import java.util.concurrent.TimeUnit;@Component public class Runner implements CommandLineRunner {private final RabbitTemplate rabbitTemplate;private final Receiver receiver;public Runner(Receiver receiver, RabbitTemplate rabbitTemplate){this.receiver = receiver;this.rabbitTemplate = rabbitTemplate;}@Overridepublic void run(String... args) throws Exception {System.out.println("Sending message ...");rabbitTemplate.convertAndSend(DemoApplication.topicExchangeName, "foo.bar.baz", "Hello from RabbitMQ!");receiver.getLatch().await(10000, TimeUnit.MILLISECONDS);} }

項目的application.properties如下：

spring.rabbitmq.host=122.xxx.xxx.xxx spring.rabbitmq.port=5672 spring.rabbitmq.username=xxxx spring.rabbitmq.password=xxxxxxx spring.rabbitmq.virtual-host=/xxxxx

程序運行截圖如下：

?

?

源碼打包下載地址：

https://github.com/fengfanchen/Java/tree/master/SpringBootRabbitmq

總結

以上是生活随笔為你收集整理的Spring Boot文档阅读笔记-对Messaging with RabbitMQ解析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。