• 在业务逻辑的异步处理，系统解耦，分布式通信以及控制高并发的场景下，消息队列有着广泛的应用。本项目基于Spring的AMQP模块，整合流行的开源消息队列中间件rabbitMQ,实现一个向rabbitMQ添加和读取消息的功能。并比较了两种模式：生产者-消费者模式和发布-订阅模式的区别。AMQP作为比JMS更加高级的消息协议，支持更多的消息路由和消息模式。

• 包含的特性如下：

1. 如上图，生产者消费者模型：添加了一个队列，并创建了两个消费者用于监听队列消息，我们发现，当有消息到达时，两个消费者会交替收到消息。这一过程虽然不用创建交换机，但会使用默认的交换机，并用默认的直连（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=test123

pom文件添加依赖：

<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

1. Channel（通道）：Channel是RabbitMQ与应用程序之间的虚拟连接。通过在物理连接（connection）上创建多个通道，应用程序可以并发地进行消息传递操作。通道负责发送和接收消息，并执行一些与消息相关的操作，如声明队列、定义交换机和绑定等。通道可以看作是轻量级的会话，通过一个物理连接与RabbitMQ进行交互。
2. Exchange（交换机）：交换机是消息的接收和转发中心。当消息发送到RabbitMQ时，会通过交换机进行路由。交换机根据其类型和绑定规则，将消息路由到一个或多个队列中。常见的交换机类型包括直连交换机（direct）、主题交换机（topic）、扇形交换机（fanout）和头部交换机（headers）。
3. Queue（队列）：队列是RabbitMQ用于存储消息的缓冲区。当消息无法立即路由到消费者时，会被存储在队列中，等待消费者来获取和处理。每个队列都有一个唯一的名称，并且按照FIFO（先进先出）的顺序进行消息的投递和消费。
4. Virtual Host（虚拟主机）：虚拟主机是逻辑上的隔离环境，用于将RabbitMQ服务器划分为多个独立的部分。每个虚拟主机都有自己的交换机、队列、绑定和权限设置。虚拟主机可以帮助不同应用程序或服务之间进行隔离，并提供安全性和资源管理的控制。
5. Broker（代理）：Broker是RabbitMQ消息队列服务器的实例，负责接收、存储和路由消息。它充当中间人的角色，将生产者发送的消息传递给消费者。一个RabbitMQ实例可以包含多个Broker，每个Broker可以承载多个虚拟主机和队列。

• RocketMQ

1. NameServer（命名服务器）：NameServer是RocketMQ的命名服务组件，用于管理和提供Broker的路由信息。它充当元数据的中心，负责维护Broker的注册信息、Topic的路由信息等。Producer和Consumer在发送和接收消息之前，需要与NameServer进行交互以获取正确的Broker信息。
2. Controller（控制器）：Controller是RocketMQ的控制器组件，负责协调和管理整个RocketMQ集群的工作。它监控Broker的状态变化，处理集群的扩容和缩容，进行负载均衡等操作。Controller是RocketMQ集群的核心组件之一，确保集群的可靠运行和自动化管理。
3. Broker（代理）：Broker是RocketMQ的消息存储和处理节点。它负责接收来自Producer的消息，并将其存储在磁盘上。Broker还负责处理Consumer的消息拉取请求，并将消息推送给Consumer进行消费。一个RocketMQ集群可以包含多个Broker，每个Broker负责存储一部分Topic的消息数据。