-
在业务逻辑的异步处理，系统解耦，分布式通信以及控制高并发的场景下，消息队列有着广泛的应用。本项目基于Spring的AMQP模块，整合流行的开源消息队列中间件rabbitMQ,实现一个向rabbitMQ添加和读取消息的功能。并比较了两种模式：生产者-消费者模式和发布-订阅模式的区别。AMQP作为比JMS更加高级的消息协议，支持更多的消息路由和消息模式。
-
包含的特性如下：
- 如上图，生产者消费者模型：添加了一个队列，并创建了两个消费者用于监听队列消息，我们发现，当有消息到达时，两个消费者会交替收到消息。这一过程虽然不用创建交换机，但会使用默认的交换机，并用默认的直连（default-direct）策略连接队列；
下面来看具体真使用：
1.安装，关于安装之前已经介绍过了，有兴趣的话，可以参考之前的文章：
PHP安装rabbitmq扩展_php 按照rabbitmq 扩展-CSDN博客
不过使用的是php版本的，原理上是一样的。
安装好之后我们打开后台：
1.yml文件配置rabbitmq：
server.port=8081
logging.path=D:/
logging.level.boot.spring=DEBUG
spring.application.name=spring-boot-rabbitmq
spring.rabbitmq.host=192.168.33.10
spring.rabbitmq.port=5672
spring.rabbitmq.username=test
spring.rabbitmq.password=test123pom文件添加依赖：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
<version>1.5.10.RELEASE</version>
</dependency>生产者：
@Transactional
@Service("producer")
public class ProducerImpl implements Producer{
@Autowired
RabbitTemplate rabbitTemplate;
public void sendMail(String queue,Mail mail) {
rabbitTemplate.setQueue(queue);
rabbitTemplate.convertAndSend(queue,mail);
}
}消费者：
@Configuration
public class ProducerConsumerConfig {
@Bean
public Queue myQueue() {
Queue queue=new Queue("myqueue");
return queue;
}
}将队列绑定到交换机：
@Configuration
public class DirectExchangeConfig {
@Bean
public DirectExchange directExchange(){
DirectExchange directExchange=new DirectExchange("direct");
return directExchange;
}
@Bean
public Queue directQueue1() {
Queue queue=new Queue("directqueue1");
return queue;
}
//将交换机和相应队列连起来
@Bean
public Binding bindingorange(){
Binding binding=BindingBuilder.bind(directQueue1()).to(directExchange()).with("orange");
return binding;
}
}写个控制器：
@Controller
public class RabbitMQController {
@Autowired
ProducerImpl producer;
@Autowired
PublisherImpl publisher;
@RequestMapping(value="/produce",produces = {"application/json;charset=UTF-8"})
@ResponseBody
public void produce(@ModelAttribute("mail")Mail mail) throws Exception{
producer.sendMail("myqueue",mail);
}
}通过监听器来监听消费：
@Component
public class DirectListener1 {
@RabbitListener(queues = "directqueue1")
public void displayMail(Mail mail) throws Exception {
System.out.println("directqueue1队列监听器1号收到消息"+mail.toString());
}
}启动项目：我们来调用一下：
看到队列的消费情况：
上篇文章介绍 rocketmq，这2种消息队列在实际应用中都等起道会很好的作用，但在实际应中，rocketmq使用java语言开发的，跨平台比较好。rabbitmq 架构设计简单，在一般的轻量级的服务用用中，可能会比较好些。官方的对比也给了说明：
适用于易用性和灵活性要求较高的场景
1、应用场景
- 异步任务处理：RabbitMQ提供可靠的消息传递机制，适用于处理异步任务，例如将耗时的任务放入消息队列中，然后由消费者异步处理，提高系统的响应速度和可伸缩性。
- 解耦系统组件：通过使用RabbitMQ作为消息中间件，不同的系统组件可以通过消息进行解耦，实现松耦合的架构，提高系统的可维护性和灵活性。
- 事件驱动架构：RabbitMQ的发布-订阅模式可以用于构建事件驱动架构，将系统中的事件作为消息发布到相应的主题，不同的消费者可以订阅感兴趣的主题进行相应的处理。
2.RocketMQ：
适用于大规模数据处理和高吞吐量的场景
- 分布式事务：RocketMQ支持分布式事务消息，适用于涉及多个业务系统的分布式事务场景，确保消息的一致性和可靠性，同时提供高吞吐量的消息传递能力。
- 实时日志处理：由于RocketMQ具备高吞吐量和低延迟的特点，可以用于实时日志处理，例如日志收集和分析、日志聚合等场景。
2、服务架构
- RabbitMQ
- Channel（通道）：Channel是RabbitMQ与应用程序之间的虚拟连接。通过在物理连接（connection）上创建多个通道，应用程序可以并发地进行消息传递操作。通道负责发送和接收消息，并执行一些与消息相关的操作，如声明队列、定义交换机和绑定等。通道可以看作是轻量级的会话，通过一个物理连接与RabbitMQ进行交互。
- Exchange（交换机）：交换机是消息的接收和转发中心。当消息发送到RabbitMQ时，会通过交换机进行路由。交换机根据其类型和绑定规则，将消息路由到一个或多个队列中。常见的交换机类型包括直连交换机（direct）、主题交换机（topic）、扇形交换机（fanout）和头部交换机（headers）。
- Queue（队列）：队列是RabbitMQ用于存储消息的缓冲区。当消息无法立即路由到消费者时，会被存储在队列中，等待消费者来获取和处理。每个队列都有一个唯一的名称，并且按照FIFO（先进先出）的顺序进行消息的投递和消费。
- Virtual Host（虚拟主机）：虚拟主机是逻辑上的隔离环境，用于将RabbitMQ服务器划分为多个独立的部分。每个虚拟主机都有自己的交换机、队列、绑定和权限设置。虚拟主机可以帮助不同应用程序或服务之间进行隔离，并提供安全性和资源管理的控制。
- Broker（代理）：Broker是RabbitMQ消息队列服务器的实例，负责接收、存储和路由消息。它充当中间人的角色，将生产者发送的消息传递给消费者。一个RabbitMQ实例可以包含多个Broker，每个Broker可以承载多个虚拟主机和队列。
- RocketMQ
- NameServer（命名服务器）：NameServer是RocketMQ的命名服务组件，用于管理和提供Broker的路由信息。它充当元数据的中心，负责维护Broker的注册信息、Topic的路由信息等。Producer和Consumer在发送和接收消息之前，需要与NameServer进行交互以获取正确的Broker信息。
- Controller（控制器）：Controller是RocketMQ的控制器组件，负责协调和管理整个RocketMQ集群的工作。它监控Broker的状态变化，处理集群的扩容和缩容，进行负载均衡等操作。Controller是RocketMQ集群的核心组件之一，确保集群的可靠运行和自动化管理。
- Broker（代理）：Broker是RocketMQ的消息存储和处理节点。它负责接收来自Producer的消息，并将其存储在磁盘上。Broker还负责处理Consumer的消息拉取请求，并将消息推送给Consumer进行消费。一个RocketMQ集群可以包含多个Broker，每个Broker负责存储一部分Topic的消息数据。
