香港私服网站大全郑州网络营销推广

一.核心概念

1.Producer

生产者，也是RabbitMQ Server的客户端，向RabbitMQ发送消息。

⽣产者(Producer)创建消息,然后发布到RabbitMQ中.在实际应⽤中,消息通常是⼀个带有⼀定业务逻辑结构的数据,⽐如JSON字符串.消息可以带有⼀定的标签,RabbitMQ会根据标签进⾏路由,把消息发送给感兴趣的消费者(Consumer)

2.Consumer

消费者,也是RabbitMQ Server的客⼾端,从RabbitMQ接收消息。

消费者连接到RabbitMQ服务器,就可以消费消息了,消费的过程中,标签会被丢掉.消费者只会收到消息,并不知道消息的⽣产者是谁,当然消费者也不需要知道.

3.Broker

其实就是RabbitMQ Server,主要是接收和收发消息

对于RabbitMQ来说,⼀个RabbitMQ Broker可以简单地看作⼀个RabbitMQ服务节点,或者RabbitMQ服务实例.⼤多数情况下也可以将⼀个RabbitMQBroker看作⼀台RabbitMQ服务器

4.三者关系图：

5.Connection

Connection：连接.是客⼾端和RabbitMQ服务器之间的⼀个TCP连接.这个连接是建⽴消息传递的基础,它负责传输客⼾端和服务器之间的所有数据和控制信息.

6.Channel

Channel：通道,信道.Channel是在Connection之上的⼀个抽象层.在RabbitMQ中,⼀个TCP连接可以有多个Channel,每个Channel都是独⽴的虚拟连接.消息的发送和接收都是基于Channel的.

通道的主要作⽤是将消息的读写操作复⽤到同⼀个TCP连接上，这样可以减少建⽴和关闭连接的开销,提⾼性能.

7. Virtual host

Virtual host:虚拟主机.这是⼀个虚拟概念.它为消息队列提供了⼀种逻辑上的隔离机制.对于RabbitMQ⽽⾔,⼀个BrokerServer上可以存在多个VirtualHost.当多个不同的⽤⼾使⽤同⼀个RabbitMQ Server提供的服务时，可以虚拟划分出多个vhost，每个⽤⼾在⾃⼰的vhost创建exchange/queue等

8.Queue

Queue:队列,是RabbitMQ的内部对象,⽤于存储消息。

9.Exchange

Exchange：交换机.message到达broker的第⼀站,它负责接收⽣产者发送的消息,并根据特定的规则把这些消息路由到⼀个或多个Queue列中.Exchange起到了消息路由的作⽤，它根据类型和规则来确定如何转发接收到的消息.

二.AMQP

1.概述

全称：Advanced Message Queuing Protocol（高级消息队列协议）

定位：开放标准的应用层协议，专为可靠、异步、跨平台的消息通信设计，支持企业级消息中间件功能。

起源：2003年由金融行业（JPMorgan Chase等）发起，旨在解决异构系统间消息传递的兼容性问题。后由OASIS组织标准化，最新版本为AMQP 1.0（ISO/IEC 19464）。

2. 核心设计目标

互操作性：打破厂商锁定，实现不同语言/平台的消息互通（如Java、.NET、Python等）。

可靠性：支持事务、持久化、消息确认，确保数据不丢失。

灵活性：动态路由、多消息模式（点对点、发布订阅）、复杂路由规则。

扩展性：适应高并发场景，支持集群和负载均衡。

3. 核心模型与组件

AMQP基于生产者-消费者模型，关键组件包括：

组件 功能描述
生产者 发送消息的应用，指定消息的routing key和属性（如优先级、持久化）。
消费者 订阅队列并处理消息的应用，支持手动确认（ACK/NACK）确保可靠性。
交换器（Exchange） 消息路由中枢，根据类型和绑定规则将消息分发到队列。类型包括：

• 直连（Direct） 精确匹配routing key（如日志级别过滤）。
• 扇出（Fanout） 广播到所有绑定队列（适合通知类场景）。
• 主题（Topic） 通配符匹配（*匹配单词，#匹配多级，如news.#）。
• 头（Headers） 根据消息头键值对路由，忽略routing key。
  队列（Queue） 存储消息的缓冲区，可配置持久化、TTL（存活时间）、死信队列等属性。
  绑定（Binding） 定义交换器与队列的连接规则（如routing key匹配或头匹配）。
  虚拟主机（vHost） 逻辑隔离单位，类似命名空间，包含独立权限、交换器、队列等资源。

4. 协议工作流程

生产者发送消息至交换器，附带routing key和属性。

交换器根据类型和绑定规则将消息路由到匹配的队列。

队列存储消息，等待消费者拉取（或主动推送）。

消费者处理消息后，发送ACK确认（若失败则NACK触发重试或进入死信队列）。

三.代码演示

1.引入依赖

在Maven仓库下选择：

<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.20.0</version>
</dependency>

2.生产者代码

public class ProducerDemo {public static void main(String[] args) throws IOException, TimeoutException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("1.95.194.117");//IPconnectionFactory.setPort(5672);//端口号connectionFactory.setUsername("账号");//用户名connectionFactory.setPassword("密码");//密码connectionFactory.setVirtualHost("study");//虚拟机//2.开启通信Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();//3.声明交换机 (此处使用内置默认的交换机)//4.声明队列/***queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,*                                  Map<String, Object> arguments)** queue:队列的名称* durable:可持久化* exclusive:是否独占* autoDelete: 是否自动删除* arguments: 参数*/channel.queueDeclare("hello",true,false,false,null);//5.发送消息/***basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)** exchange: 交换机* routingKey: 内置交换机，routingkey和队列名称保持一致* props: 属性配置* body: 消息*/String message = "hello RabbitMQ";channel.basicPublish("","hello",null,message.getBytes());//6.资源释放channel.close();connection.close();}
}

3.消费之代码

public class ConsumerDemo {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("1.95.194.117");connectionFactory.setPort(5672);connectionFactory.setUsername("zhanghao");connectionFactory.setPassword("mima");connectionFactory.setVirtualHost("study");Connection connection = connectionFactory.newConnection();//2.通信Channel channel = connection.createChannel();//3.声明队列(可以省略)channel.queueDeclare("hello",true,false,false,null);//4.消费消息/***String basicConsume(String queue, boolean autoAck, Consumer callback)* queue: 队列名称* autoAck: 是否自动确认* callback: 接受消息后，执行的逻辑*/DefaultConsumer consumer = new DefaultConsumer(channel) {//从队列中收到消息, 就会执行的方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {/*** consumerTag: 消费者标签, 通常是消费者在订阅队列时指定的。* envelope: 包含消息的封包信息, 如队列名称，交换机等。* properties: 一些配置信息* body: 消息的具体内容*/System.out.println("接收到消息: " + new String(body));}};channel.basicConsume("hello",true,consumer);Thread.sleep(2000);//5.资源关闭channel.close();connection.close();}
}