创建SpringBoot和RabbitMQ的整合项目

创建SpringBoot和RabbitMQ的整合项目

首先快速创建一个maven项目

如下图：

引入SpringBoot整合rabbitMQ的依赖

如下图：

这里要SpringBoot和RabbitMQ整合的依赖对应的就是spring-boot-starter-amqp，那么amqp是什么意思呢？是Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，是面向消息的中间件设计。比如RabbitMQ消息中间件使用的就是这个协议。
因为我们要启动服务，因此需要有tomcat服务器，因此这里还要引入spring-boot-starter-web模块依赖，因为SpringBoot自带的tomcat服务器在web模块里面。

<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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>org.example</groupId>
  <artifactId>rabbitMQ</artifactId>
  <version>1.0-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>rabbitMQ</name>
  <url>http://maven.apache.org</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>

  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.6.7</version>
  </parent>

  <dependencies>
    <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>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>
  </dependencies>
</project>

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
31
32
33
34
35
36
37
38
39
40
41

在src/main目录下创建resources目录并引入配置文件

如下图：

写消息发送者MessageSender

如下图：

/**
 * @author xuan
 * @create 2024/4/25
 */
@Component
public class MessageSender {
    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void sendMessage(String exchange, String routingKey, Object message) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

写消息接收者MessageReceiver

如下图：

/**
 * @author xuan
 * @create 2024/4/25
 */
@Component
public class MessageReceiver {
    @RabbitListener(queues = "test.queue")
    public void handleMessage(Object message) {
        System.out.println("Received message: " + message);
    }
}
1
2
3
4
5
6
7
8
9
10
11

写RabbitMQConfig配置类

如下图：

/**
 * @author xuan
 * @create 2024/4/25
 */
@Configuration
public class RabbitMQConfig {
    @Bean
    public DirectExchange testExchange() {
        return new DirectExchange("test.exchange");
    }

    @Bean
    public Queue testQueue() {
        return new Queue("test.queue");
    }

    @Bean
    public Binding testBinding() {
        return BindingBuilder.bind(testQueue()).to(testExchange()).with("test.routingKey");
    }
}

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

写SpringBoot启动主类

如下图：

/**
 * @author xuan
 * @create 2024/4/25
 */
@SpringBootApplication
public class Application implements CommandLineRunner {
    @Autowired
    private MessageSender messageSender;

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        messageSender.sendMessage("test.exchange", "test.routingKey", "Hello, RabbitMQ!");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

CommandLineRunner接口的作用

从上面的SpringBoot启动类可以看到，它实现了CommandLineRunner接口，实现了这个接口中的抽象方法run，那么CommandLineRunner接口又什么作用呢？它主要是用在SpringBoot的启动主类上的，主要作用是在SpringBoot应用程序启动之后执行一些任务，比如：

• 数据初始化：加载初始数据到数据库或其他存储系统中。
• 发送通知：向用户或系统发送启动通知，例如通过邮件或消息队列。
• 执行检查：进行应用程序启动前的检查，如数据库连接检查、文件系统检查等。

比如我们上面的主类在成功启动SpringBoot的应用程序之后 就会给RabbitMQ消息队列发送一个消息。

启动程序 查看效果

启动主类之后 控制台输出如下图：

这里我们主要看消息体Body的内容，可以发现是Hello RabbitMQ。后面是消息的一些属性 类似如我们的http协议中的request请求头的信息，包括消息的内容类型了，消息的内容编码了，消息的长度了等信息。

可以发现我们这里测试成功了。

注意要点

15672和5672的区别

RabbitMQ前端管理界面使用的端口号是15672，比如我们启动rabbitmq之后，在浏览器中输入127.0.0.1:15672可以进入rabbitmq的前端页面管理器中进行相关操作，如下图：

但是后端服务器的rabbitmq在电脑上启动的端口号是5672，如下图：

这两个端口号特别容易搞错，千万要注意，如果配置文件里的端口号写15672，那么我们的idea里面的java程序就连接不上我们本地已经启动的rabbitmq服务器了。

消息发送的队列怎么找

通过RabbitMQ中的交换机Exchange，和路由RouteingKey可以唯一的定位到一个消息队列Queue，如下图：

因此我们代码里面发送者发送消息的时候，一定要指定使用的交换机和使用的路由，但不用指定使用的队列，因为有交换机和路由之后就会自动的映射到使用的队列了，看下代码如下图：

但是我们需要通过一个配置类配置一下 需要告诉java程序，如下图：

然后接收者接收消息的时候 可以直接指定test.queue队列，就可以自动的去这个队列中取出消息了。不用写交换机和路由的具体信息。下面看下接收者的处理 如下图：

但是我觉得RabbitMQ应该是会有线程安全问题的，比如说因为目前看来同一个队列 两个不同的线程都是可以访问的，那这样的话有共享资源问题 就肯定会出现线程安全问题。

我们的消息发送者和消息接收者可以只开启一个

只开启消息发送者先存一些消息到消息队列

比如我们可以先用消息发送者往RabbitMQ的消息队列test.queue里面发送三条消息，此时注意关闭消息接收者，以免接收者直接读取队列里面的消息，我们在页面上就看不到效果了，如下图：

只开启消息接收者从消息队列里面读取消息

接下来我们把消息发送者关了，把消息接收者打开，去消费队列里面的消息，如下图：


看下启动应用程序之后的效果 如下图：

接下来我们来看一下RabbitMQ中怎么控制一个线程一次消费的消息数，主要是看下concurrency关键属性

通过concurrency属性来控制线程从队列中一次性读取的消息个数。如果我们把属性concurrency的值设置为1 那么就表示一个线程一次性会从队列里面读取一条消息；如果把这个值设置成3 那么就表示一个线程一次性会从队列里面读取3条消息；concurrency的书写位置如下图：

据我的观察，RabbitMQ内部应该有线程安全机制，就是在当前线程读取消费队列消息没有完成的时候，另外的线程是不能消费队列消息的。

现在我们模拟两个线程，而我们具体的模拟方式其实是重新复制一个项目，然后改下端口号为8081，接着我们把给队列发送消息的代码注释掉 只开启接收者从消息队列里面读取消息 过程如下图：

上面是我们模拟的第一个线程，其实也就是copy一个相同的项目 改下端口号 启动的时候就是不同的线程了。

然后启动第二个线程，发送6条消息到消息队列中，如下图：

那么这个时候呢？线程一就会先读取消息队列的第一条消息，线程二此时会在线程一后面排队，当线程一处理完消息之后，线程二会去读取消息队列里面的消息，然后线程一读取…。因此正常情况的输出是 其中一个应用程序中 输出的是编号1,3，5的消息，另外一个应用程序中输出的是编号2,4，6的消息。如下图：


确实是正常的。

现在我们把concurrency线程一次性最多读取的消息个数改成4，看下控制台输出结果，如下图：


可以看到第一个线程一次性会读取消息队列的前四条消息，而第二个线程本来也会读取四条 但是队列中只剩两条消息，因此第二个线程就一次性的读取两条消息了。
但是有个线程安全问题。就是你会发现第一个线程虽然读取的确实是队列里面的前四条消息，但是呢？消费消息的时候却不是按队列里面添加消息的顺序，这可能就会出现问题了。比如假如说队列里面按照先后顺序添加了两条消息，1穿装，2脱装。但是线程读取消息之后 先消费处理的可能是2脱装，那是不是就会出现问题了？因为你玩家身上现在根本就没有装备 你要怎么脱呢？因此当concurrency设置的值超过1的时候都可能会出现消息消费顺序有误的问题。

因此我们如果想要保证消费的消息严格的按照队列里面添加的消息顺序来执行，那么必须要把concurrency设置成1。