Spring Boot文档阅读笔记-对Messaging with RabbitMQ解析

野性酷女 2022-12-24 04:56 159阅读 0赞

此篇教程以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/> 
        </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";
    
        @Bean
        Queue queue(){
    
            return new Queue(queueName, false);
        }
    
        @Bean
        TopicExchange exchange(){
    
            return new TopicExchange(topicExchangeName);
        }
    
        @Bean
        Binding binding(Queue queue, TopicExchange exchange){
    
            return BindingBuilder.bind(queue).to(exchange).with("foo.bar.#");
        }
    
        @Bean
        SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,
                                                 MessageListenerAdapter listenerAdapter){
    
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
            container.setConnectionFactory(connectionFactory);
            container.setQueueNames(queueName);
            container.setMessageListener(listenerAdapter);
            return container;
        }
    
        @Bean
        MessageListenerAdapter 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;
        }
    
        @Override
        public 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

程序运行截图如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxNzg0NDI3NjE_size_16_color_FFFFFF_t_70][]

源码打包下载地址：

[https://github.com/fengfanchen/Java/tree/master/SpringBootRabbitmq][https_github.com_fengfanchen_Java_tree_master_SpringBootRabbitmq]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxNzg0NDI3NjE_size_16_color_FFFFFF_t_70]: /images/20221120/0a285ff0b6d144158570baf514b55ff8.png
[https_github.com_fengfanchen_Java_tree_master_SpringBootRabbitmq]: https://github.com/fengfanchen/Java/tree/master/SpringBootRabbitmq