RabbitMQ - 基于 SpringAMQP 带你实现五种消息队列模型_spring消息队列_简单队列模式amqp实现

System.out.println("消费者收到消息: " + msg);
}

• value: @Queue 注解，用于声明队列，value 为 queueName, durable 表示队列是否持久化, autoDelete 表示没有消费者之后队列是否自动删除
• exchange: @Exchange 注解，用于声明 exchange， type 就是交换机类型.
• key: 在 topic 方式下，这个就是我们熟知的 routingKey

1.2.5、为什么更建议使用 @Bean 注解创建，而不是 @RabbitListener 注解创建？

建议使用 @Bean 而不是@RabbitListener的原因如下：

1. 使用 @Bean 创建相比于 @RabbitListener 实现了配置和业务上的解耦合（队列、交换机、绑定的创建 和 业务代码 解耦合）.
2. @Bean可以更好地控制RabbitMQ监听器的创建和销毁，而@RabbitListener则需要在运行时动态创建和销毁监听器，这可能会导致一些性能问题和内存泄漏问题。
3. 使用@Bean还可以更好地管理RabbitMQ连接和通道，从而提高应用程序的可靠性和性能。

1.3、案例实现

1.3.1、SpringAMQP 实现基础消息队列（BasicQueue）

在mq上创建队列：

Configuration 这一层创建队列，否则不生效.

@Configuration
public class MqConfig {

/**

• 创建队列
• @return
  */
  @Bean
  public Queue simpleQueue() {
  return new Queue(“simple.queue”, true);
  }

}

.

服务生产者 publisher：

a）由于生产者和消费者服务都需要 amqp 依赖，因此这里直接将依赖放到父工程中

org.springframework.boot spring-boot-starter-amqp

b）在 publisher （消息发送）服务中编写application.yml，添加mq连接信息：

spring:
rabbitmq:
host: 193.168.150.185 # 主机名
port: 5672 # 端口
virtual-host: / # 虚拟主机
username: root # 用户名
password: 1234 # 密码

Ps：云服务器记得放行 5672 端口！！！

c）在publisher服务中新建一个测试类，编写测试方法：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAMQPTest {

@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testSendMessageSimpleQueue() {
String queueName = “simple.queue”;
String message = “hello, spring amqp!”;
rabbitTemplate.convertAndSend(queueName, message);
}

}

以上可得知 SpringAMQP 发送消息需要实现以下几点：

1. 引入 amqp 的 starter 依赖；
2. 配置 RabbitMQ 地址；
3. 利用 RabbitTemplate 的 convertAndSend 方法来指定队列发送消息。

服务消费者 consumer：

a）在consumer服务中编写application.yml，添加mq连接信息：

spring:
rabbitmq:
host: 193.168.150.185 # 主机名
port: 5672 # 端口
virtual-host: / # 虚拟主机
username: root # 用户名
password: 1234 # 密码

b）在consumer服务中新建一个类，编写消费逻辑：

@Component
public class SpringRabbitListener {

@RabbitListener(queues = “simple.queue”)
public void ListenQueue(String msg) {
System.out.println(“消费者接收到 simple.queue 的消息:” + msg);
}

}

以上可得知 SpringAMQP 接收消息需要实现以下几点：

1. 引入amqp的starter依赖
2. 配置RabbitMQ地址
3. 定义类，添加@Component注解
4. 类中声明方法，添加@RabbitListener注解，方法参数就时消息

Ps：消息一旦消费就会从队列删除，RabbitMQ没有消息回溯功能

最终执行效果如下：

1.3.2、SpringAMQP 实现工作消息队列（（Work Queue）

Work queue工作队列，可以提高消息处理速度，避免队列消息堆积；

**遇到的问题：**当消息进入 queue 时，消费者1 和 消费者2 都会先做一个操作叫做“消息预取”，也就是先取消息，但不执行操作，等从 queue 中拿完（如果不设置，默认是无限次拿）了以后再执行具体的消费操作，但这样会导致一个问题——当 消费者1 的消费处理速度是 消费者2 处理速度的两倍时，**但是由于“消息预取”，他们两都会取到相等的消息数量，**因此就会导致 消费者1 提前处理完，消费者2 还有很多没有处理的情况，降低了总体处理的效率。

**解决办法：**为了解决这种问题的发生，也可以通过修改消息预取的数量，设置为 1 ，就是拿一个处理完了以后再取下一个，就可以避免以上问题（设置为1，就是根据自身处理能力，拿去相应的数据）。

具体实现如下

a）在 Configuration 中创建好队列

@Configuration
public class MqConfig {

/**

• 创建队列
• @return
  */
  @Bean
  public Queue simpleQueue() {
  return new Queue(“simple.queue”, true);
  }

}

b）在 publisher 服务中添加一个测试方法，循环发送 50 条消息到 simple.queue 队列（控制时间为1秒内发送完）

@Test
public void testSendMessageWorkQueue() throws InterruptedException {
String queueName = “simple.queue”;
String message = “hello, workQueue!”;
for(int i = 0; i < 50; i++) {
rabbitTemplate.convertAndSend(queueName, message + i);
Thread.sleep(20);
}
}

c）在consumer服务中创建两个消费者，也监听simple.queue：

@Component
public class SpringRabbitListener {

@RabbitListener(queues = “simple.queue”)
public void ListenQueue(String msg) throws InterruptedException {
System.err.println(“消费者接收到 simple.queue 的消息:” + msg + LocalDate.now());
Thread.sleep(100);
}

@RabbitListener(queues = “simple.queue”)
public void ListenQueue2(String msg) throws InterruptedException {
System.out.println(“消费者接收到 simple.queue 的消息:” + msg + LocalDate.now());
Thread.sleep(25);
}

}

d）消费预取限制：修改application.yml文件，设置preFetch这个值，可以控制预取消息的上限

logging:
pattern:
dateformat: MM-dd HH:mm:ss:SSS
spring:
rabbitmq:
host: 170.193.116.135
port: 5672
username: root
password: 1234
virtual-host: /
listener:
simple:
prefetch: 1

执行结果如下：

1.3.3、SpringAMQP 实现广播交换机消息队列（Fanout Exchange）

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。实现方式是加入了exchange（交换机）。

常见exchange类型包括：

1. Fanout：广播
2. Direct：路由
3. Topic：话题

这里我们先来理解一下广播消息队列模型~

Fanout Exchange：会将接收到的消息广播到每一个跟其绑定的queue，最后交给对应的消费者，值得注意的是，exchange****负责消息路由，而不是存储，路由失败则消息丢失。

具体实现如下：

a）在consumer服务创建一个类，添加@Configuration注解，创建队列和交换机，并声明FanoutExchange、Queue和绑定关系对象 Binding ，绑定方式是 BindingBuilder.bind(队列).to(交换机)

package cn.itcast.mq.config;

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

@Configuration
public class FanoutConfig {

//itcast.fanout
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange(“itcast.fanout”);
}

//itcast.queue1
@Bean
public Queue fanoutQueue1() {
return new Queue(“fanout.queue1”);
}

//绑定队列 1 到交换机
@Bean
public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
return BindingBuilder
.bind(fanoutQueue1)
.to(fanoutExchange);
}

//itcast.queue2
@Bean
public Queue fanoutQueue2() {
return new Queue(“fanout.queue2”);
}

//绑定队列 2 到交换机
@Bean
public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {
return BindingBuilder
.bind(fanoutQueue2)
.to(fanoutExchange);
}

}

b）运行以后可以通过 RabbitMQ 客户端看到以下绑定情况：

c）添加测试方法

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAMQPTest {

@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testFanoutMessage() {
String exchangeName = “itcast.fanout”;
String message = “hello! fanout”;
rabbitTemplate.convertAndSend(exchangeName, “”, message);
}

}

d）这里一个交换机绑定了两个队列，因此运行结果如下 ：

1.3.4、Spring AMQP 实现路由交换机消息队列（Direct Exchange）

自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。

深知大多数大数据工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年大数据全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友。

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上大数据开发知识点，真正体系化！

由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新

如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）

到想自学提升又不知道该从何学起的朋友。**
[外链图片转存中…(img-fRqZiXd9-1712531510735)]
[外链图片转存中…(img-78iz5Odc-1712531510736)]
[外链图片转存中…(img-UmNjkdGO-1712531510736)]
[外链图片转存中…(img-KAxf2UPy-1712531510737)]
[外链图片转存中…(img-BOhz3l5U-1712531510737)]

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上大数据开发知识点，真正体系化！

由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新

如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）
[外链图片转存中…(img-PmYwNFUj-1712531510737)]