此篇教程以Rabbitmq作為消息隊(duì)列服務(wù)端，使用Spring Boot產(chǎn)生和發(fā)布消息。

?

使用Spring AMQP的RabbitTemplate發(fā)布消息，使用MessageListenerAdapter訂閱消息。

?

其中對(duì)應(yīng)的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>

下面是創(chuàng)建接受者，獲取發(fā)布者的發(fā)布的消息：

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;} }

這個(gè)Receiver是POJO類，這里的POJO是指沒(méi)有帶有業(yè)務(wù)處理的類，當(dāng)注冊(cè)后，就能在任意的地方進(jìn)使用。

CountDownLatch，里面會(huì)有一個(gè)接收消息的信號(hào)。

?

注冊(cè)監(jiān)聽(tīng)者以及發(fā)送消息

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

1. 配置消息監(jiān)聽(tīng)者容器；

2.聲明隊(duì)列，交換機(jī)對(duì)其都進(jìn)行綁定；

3.配置發(fā)送者組建，使得監(jiān)聽(tīng)者能接收到。

代碼如下：

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()方法將消息監(jiān)聽(tīng)者注冊(cè)到容器中（默認(rèn)是在container()）。

queue()方法創(chuàng)建了AMQP的隊(duì)列。exchange()方法創(chuàng)建了topic交換機(jī)。

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

?

發(fā)送測(cè)試數(shù)據(jù)到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);} }

項(xiàng)目的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

程序運(yùn)行截圖如下：

?

?

源碼打包下載地址：

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

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。