RabbitMQ（三）消费者生产者_rabbitmq消费者

前置：RabbitMQ的工作机制和架构图

消息中间件最大的特点就是解耦：生产者只要把消息发送不去就可以不用管了，消费者不管在不在线都行，消费者上线的时候会来处理带消费的信息

• Publisher - 生产者：发布消息到RabbitMQ中的Exchange
• Consumer - 消费者：监听RabbitMQ中的Queue中的消息
• Exchange - 交换机：和生产者建立连接并接收生产者的消息
• Queue - 队列：Exchange会将消息分发到指定的Queue，Queue和消费者进行交互
• Routes - 路由：交换机以什么样的策略将消息发布到Queue

简单架构图

原理：消息发送的时候，
1.先由生产者发送到消息交换机（有默认，可配置）
2.消息交换机根据既定的策略帮你把消息路由到不同的队列上面去(意味着消息收发的形式更加的多样)
3.消息消费者就去监听不同的队列

完整架构图

一. 搭建一个空的工程

目的:方便日后多个项目关联及扩充

第一步

1.1 建立consumer01工程—创建消费者

1.1.1 依赖

 <dependency>                                            
     <groupId>org.springframework.boot</groupId>         
     <artifactId>spring-boot-starter-amqp</artifactId>   
 </dependency>                                           
 <dependency>                                            
     <groupId>org.springframework.boot</groupId>         
     <artifactId>spring-boot-starter-web</artifactId>    
 </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>                                           
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

1.1.2 application配置

spring.rabbitmq.host=1.12.235.192
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

#虚拟主机，这个配置非常重要
spring.rabbitmq.virtual-host=/
1
2
3
4
5
6

1.1.3 RabbitConfig

package com.huang.config;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;



@Configuration
public class RabbitConfig {
    public static  final   String MY_QUEUE_NAME_01= "hello-queue";

    /**
     * 在Spring中，只需要将一个Queue对象注册到Spring容器中，RabbitMQ就会自动创建该队列
     * @return
     */
    @Bean
    Queue myQueue(){
        /**
         * 1.队列的名字   我们这里为"hello-queue"
         * 2.队列是否持久化：RabbitMQ重启之后，队列是否还存在   我们这里设置为true，即队列存在
         * 3.队列的排他性：这个为true，只有创建这个队列的连接才能给这个队列发消息    我们这里设置为false，即谁都可以给这个队列发消息
         * 4.如果没有人消费这个队列，是否直接删除该队列
         */

        return  new Queue(MY_QUEUE_NAME_01,true,false,false);
    }
}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

1.1.4 RabbitConsumer

package com.huang.consumer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;



@Component
public class RabbitConsumer {
       private static  final Logger logger = LoggerFactory.getLogger(RabbitConsumer.class);

    /**
     * @RabbitListener(queues = "hello-queue")
     * 通过这个注解，来指定该方法需要监听的消息队列，注解中的参数就是要去监听的消息队列的名称。
     * @param msg
     */
    @RabbitListener(queues = "hello-queue")
       public  void  handleMsg(String msg){
           logger.info("msg:{}",msg);
       }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

1.2 建立producer01工程—创建生产者

1.2.1 依赖

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

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14

1.2.2 application

spring.rabbitmq.host=1.12.235.192
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

//5672通信管理，15672界面管理
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/
1
2
3
4
5
6
7

1.2.3 创建一个和消费者一样的RabbitConfig

1.2.4 生产者的测试类创建消息模板

@SpringBootTest
class Producer01ApplicationTests {
    //这个就是一个消息发送模板
    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() {
        /**
         * 1.消息队列的名字
         * 2.要发送的消息内容
         * 3.
         */
     rabbitTemplate.convertAndSend(RabbitConfig.MY_QUEUE_NAME_01,"hello rabbitmq!");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

运行这个测试类去发送消息，消费者收到消息

1.2.5 测试结果即RabbitMQ网页显示

java的生产者启动后，消费者控制台收到消息

RabbitMQ界面显示

1.Ready等待消费的消息数量
2.Unacked消息已经消费确认数量
3.totl是1和2的总和