如何制作h5做网站,上海网站建设的软件,尖叫直播,网站建设面临的困难目 录 一. RabbitMQ 安装二. RabbitMQ 核心概念2.1 Producer 和 Consumer2.2 Connection 和 Channel2.3 Virtual host2.4 Queue2.5 Exchange2.6 RabbitMQ 工作流程 三. AMQP四. web界面操作4.1 用户相关操作4.2 虚拟主机相关操作 五. RabbitMQ 快速入门5.1 引入依赖5.2 编写生产… 目 录 一. RabbitMQ 安装二. RabbitMQ 核心概念2.1 Producer 和 Consumer2.2 Connection 和 Channel2.3 Virtual host2.4 Queue2.5 Exchange2.6 RabbitMQ 工作流程 三. AMQP四. web界面操作4.1 用户相关操作4.2 虚拟主机相关操作 五. RabbitMQ 快速入门5.1 引入依赖5.2 编写生产者代码5.2.1 创建连接5.2.2 创建 Channel5.2.3 声明⼀个队列 Queue5.2.4 发送消息5.2.5 释放资源5.2.6 运行代码, 观察结果 5.3 编写消费者代码5.3.1 消费当前队列5.3.2 释放资源5.3.3 运行代码, 观察结果 5.4 附源码 一. RabbitMQ 安装
我们对于 RabbitMQ 已经有了简单的了解, 接下来进行 RabbitMQ 的安装
RabbitMQ 是⼀套开源的消息队列服务软件, 基于 Erlang 语言编写, 所以安装RabbitMQ 之前, 需要先安装部署 Erlang 环境, 再安装 RabbitMQ 环境. RabbitMQ 大多部署在 Linux 操作系统 此处不做详细安装教程~
二. RabbitMQ 核心概念
在安装完 RabbitMQ 之后, 接下来学习如何去使用 RabbitMQ
来到 RabbitMQ 管理界面 界面上的导航栏共分6部分, 这6部分分别是什么意思呢, 我们先看看 RabbitMQ 的工作流程 RabbitMQ 是⼀个消息中间件, 也是⼀个生产者消费者模型. 它负责接收, 存储并转发消息 消息传递的过程类似邮局. 当你要发送⼀个邮件时你把你的邮件放到邮局邮局接收邮件, 并通过邮递员送到收件人的手上 按照这个逻辑, Producer 就类似邮件发件人. Consumer 就是收件人, RabbitMQ 就类似于邮局 2.1 Producer 和 Consumer
Producer: 生产者, 是 RabbitMQ Server 的客户端, 向 RabbitMQ 发送消息Consumer: 消费者, 也是 RabbitMQ Server 的客户端, 从 RabbitMQ 接收消息Broker其实就是 RabbitMQ Server, 主要是接收和收发消息
生产者 (Producer) 创建消息, 然后发布到 RabbitMQ 中. 在实际应用中, 消息通常是⼀个带有⼀定业务逻辑结构的数据, 比如 JSON 字符串. 消息可以带有⼀定的标签, RabbitMQ 会根据标签进行路由, 把消息发送给感兴趣的消费者(Consumer).消费者连接到 RabbitMQ 服务器, 就可以消费消息了, 消费的过程中, 标签会被丢掉. 消费者只会收到消息, 并不知道消息的生产者是谁, 当然消费者也不需要知道.对于 RabbitMQ 来说,⼀个 RabbitMQ Broker 可以简单地看作⼀个 RabbitMQ 服务节点, 或者 RabbitMQ 服务实例. 大多数情况下也可以将⼀个 RabbitMQ Broker 看作⼀台 RabbitMQ 服务器 2.2 Connection 和 Channel
Connection: 连接. 是客户端和 RabbitMQ 服务器之间的⼀个 TCP 连接. 这个连接是建立消息传递的基础, 它负责传输客户端和服务器之间的所有数据和控制信息.
Channel: 通道, 信道. Channel 是在 Connection 之上的⼀个抽象层. 在 RabbitMQ 中, ⼀个 TCP 连接可以有多个 Channel, 每个 Channel 都是独立的虚拟连接. 消息的发送和接收都是基于 Channel 的
通道的主要作用是将消息的读写操作复用到同⼀个 TCP 连接上这样可以减少建立和关闭连接的开销,提高性能 2.3 Virtual host
Virtual host: 虚拟主机. 这是⼀个虚拟概念. 它为消息队列提供了⼀种逻辑上的隔离机制. 对于 RabbitMQ 而言, ⼀个 BrokerServer 上可以存在多个 Virtual Host. 当多个不同的用户使用同⼀个RabbitMQ Server 提供的服务时可以虚拟划分出多个 vhost每个用户在自己的 vhost 创建 exchange/queue 等 类似 MySQL 的 “database”, 是⼀个逻辑上的集合. ⼀个 MySQL 服务器可以有多个 database 2.4 Queue
Queue: 队列, 是 RabbitMQ 的内部对象, 用于存储消息 多个消费者, 可以订阅同⼀个队列 2.5 Exchange
Exchange: 交换机. message 到达 broker 的第⼀站, 它负责接收生产者发送的消息, 并根据特定的规则把这些消息路由到⼀个或多个 Queue 列中.
Exchange 起到了消息路由的作用它根据类型和规则来确定如何转发接收到的消息 类似于发快递之后, 物流公司怎么处理呢, 根据咱们的地址来分派这个快递到不同的站点, 然后再送到收件人手里. 这个分配的工作,就是交换机来做的 2.6 RabbitMQ 工作流程
理解了上面的概念之后, 再来回顾⼀下这个图, 来看 RabbitMQ 的工作流程 Producer 生产了⼀条消息Producer 连接到 RabbitMQBroker, 建立⼀个连接 (Connection),开启⼀个信道(Channel)Producer 声明⼀个交换机 (Exchange), 路由消息Producer 声明⼀个队列 (Queue), 存放信息Producer 发送消息至 RabbitMQ BrokerRabbitMQ Broker 接收消息, 并存入相应的队列 (Queue) 中, 如果未找到相应的队列, 则根据生产者的配置, 选择丢弃或者退回给生产者 如果我们把 RabbitMQ 比作⼀个物流公司那么它的⼀些核心概念可以这样理解 Broker就类似整个物流公司的总部, 它负责协调和管理所有的物流站点, 确保包裹安全、⾼效地送达.Virtual Host 可以看作是物流公司为不同的客户或业务部门划分的独立运营中⼼. 每个运营中心都有自己的仓库(Queue), 分拣规则(Exchange)和运输路线(Connection和Channel), 这样可以确保不同客户的包裹处理不会相互干扰, 同时提供定制化的服务Exchange 就像是站点里的分拣中心. 当包裹到达时, 分拣中心会根据包裹上的标签来决定这个包裹应该送往哪个目的地(队列). 快递站点可能有不同类型的分拣中心, 有的按照具体地址分拣, 有的将包裹复制给多个收件人等.Queue 就是快递站点里的⼀个个仓库, 用来临时存放等待派送的包裹. 每个仓库都有⼀个或多个快递员(消费者)负责从仓库中取出包裹并派送给最终的收件人.Connection 就像是快递员与快递站点之间的通信线路. 快递员需要通过这个线路来接收派送任务(消息).Channel 就像是快递员在执行任务时使用的多个并行的通信线路. 这样快递员可以同时处理多个包裹, 比如⼀边派送包裹, ⼀边接收新的包裹 三. AMQP
AMQPAdvanced Message Queuing Protocol是⼀种高级消息队列协议, AMQP 定义了⼀套确定的消息交换功能, 包括交换器(Exchange), 队列(Queue) 等. 这些组件共同工作, 使得生产者能够将消息发送到交换器. 然后由队列接收并等待消费者接收. AMQP 还定义了⼀个网络协议, 允许客户端应用通过该协议与消息代理和 AMQP 模型进行交互通信
RabbitMQ 是遵从 AMQP 协议的,换句话说,RabbitMQ 就是 AMQP 协议的 Erlang 的实现(当然 RabbitMQ 还支持 STOMP2, MQTT2 等协议). AMQP 的模型结构和 RabbitMQ 的模型结构是⼀样的 四. web界面操作
RabbitMQ 管理界面上的 Connections,Channels, Exchange, Queues 就是和上面流程图的概念是⼀样的, Overview 就是视图的意思, Admin 是用户管理.
我们在操作 RabbitMQ 前, 需要先创建 Virtual host
接下来看具体操作:
4.1 用户相关操作
添加用户
a) 点击 Admin - Add user b) 设置账号密码及权限 ①: 设置账号 ②: 设置密码 ③: 确认密码 ④: 设置权限 添加完成后, 点击[Add user]
c) 观察用户是否添加成功 用户相关操作
a) 点击要删除的用户, 查看用户详情 b) 在用户详情页面进行更新或删除操作
设置对虚拟机的操作权限 更新/删除用户 退出当前用户 4.2 虚拟主机相关操作
创建虚拟主机
在 Admin 标签页下, 点击右侧 Virtual Hosts - Add a new virtual host
设置虚拟主机名称 观察设置结果 此操作会为当前登录用户设置虚拟主机
五. RabbitMQ 快速入门
步骤:
引入依赖编写生产者代码编写消费者代码
5.1 引入依赖
dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.7.3/version
/dependency5.2 编写生产者代码
5.2.1 创建连接 RabbitMQ 默认的用于客户端连接的 TCP 端口号是 5672, 需要提前进行开放 此处修改了端口号为15673 // 1. 创建连接⼯⼚
ConnectionFactory factory new ConnectionFactory();
//2. 设置参数
factory.setHost(110.41.51.65);//ip 默认值localhost
factory.setPort(15673); //默认值5672
factory.setVirtualHost(bite);//虚拟机名称, 默认 /factory.setUsername(study);//⽤⼾名,默认guest
factory.setPassword(study);//密码, 默认guest
//3. 创建连接Connection
Connection connection factory.newConnection();5.2.2 创建 Channel 生产者和消费者创建的 channel 并不是同⼀个 //4. 创建channel通道
Channel channel connection.createChannel();5.2.3 声明⼀个队列 Queue
/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, MapString, Object arguments)1.queue: 队列名称2.durable: 是否持久化.true-设置队列为持久化, 待久化的队列会存盘,服务器重启之后, 消息不丢失。3.exclusive:* 是否独占, 只能有⼀个消费者监听队列* 当 Connection 关闭时, 是否删除队列4.autoDelete: 是否⾃动删除, 当没有Consumer时, ⾃动删除掉5.arguments: ⼀些参数
*/
//如果没有⼀个hello_world 这样的⼀个队列, 会⾃动创建, 如果有, 则不创建
channel.queueDeclare(hello,true,false,false,null);5.2.4 发送消息
当⼀个新的 RabbitMQ 节点启动时, 它会预声明declare几个内置的交换机内置交换机名称是空字符串(“”). 生产者发送的消息会根据队列名称直接路由到对应的队列. 例如: 如果有⼀个名为 “hello” 的队列, 生产者可以直接发送消息到 “hello” 队列, 而消费者可以从hello 队列中接收消息, 而不需要关心交换机的存在. 这种模式非常适合简单的应用场景其中生产者和消费者之间的通信是⼀对⼀的 //6. 通过channel发送消息到队列中
/*basicPublish(String exchange, String routingKey,AMQP.BasicProperties props, byte[] body)1.exchange: 交换机名称, 简单模式下, 交换机会使⽤默认的2.routingKey: 路由名称, routingKey 队列名称3.props: 配置信息4.body: 发送消息的数据
*/
String msg Hello World;
//使⽤的是内置交换机. 使⽤内置交换机时, routingKey要和队列名称⼀样, 才可以路由到对应的队列上去
channel.basicPublish(,hello,null,msg.getBytes());
System.out.println(msg 消息发送成功);5.2.5 释放资源
//显式地关闭Channel是个好习惯, 但这不是必须的, Connection关闭的时候,Channel也会⾃动关闭.
channel.close();
connection.close();5.2.6 运行代码, 观察结果
运行之前 运行之后, 队列中就已经有了hello这个队列的信息 右上角需要选择虚拟机 如果在代码中注掉资源释放的代码, 在 Connections 和 Channels 也可以看到相关信息 Queue 也可以配置显示 Consumer 相关信息 5.3 编写消费者代码
消费者代码和生产者前3步都是⼀样的, 第4步改为消费当前队列
创建连接创建 Channel声明⼀个队列 Queue消费消息释放资源
5.3.1 消费当前队列
basicConsume
/*basicConsume(String queue, boolean autoAck, Consumer callback)参数:1. queue: 队列名称2. autoAck: 是否⾃动确认, 消费者收到消息之后,⾃动和MQ确认3. callback: 回调对象
*/
String basicConsume(String queue, boolean autoAck, Consumer callback) throwsIOException;Consumer
Consumer 用于定义消息消费者的行为. 当我们需要从 RabbitMQ 接收消息时, 需要提供⼀个实现了 Consumer 接口的对象.
DefaultConsumer 是 RabbitMQ 提供的⼀个默认消费者, 实现了 Consumer 接口
核心方法:
handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) : 从队列接收到消息时, 会自动调用该方法.
在这个方法中, 我们可以定义如何处理接收到的消息, 例如打印消息内容, 处理业务逻辑或者将消息存储到数据库等.
参数说明如下
consumerTag : 消费者标签, 通常是消费者在订阅队列时指定的.envelope : 包含消息的封包信息如队列名称, 交换机等.properties : ⼀些配置信息body : 消息的具体内容
//6. 接收消息, 并消费
/*basicConsume(String queue, boolean autoAck, Consumer callback)参数:1. queue: 队列名称2. autoAck: 是否⾃动确认, 消费者收到消息之后,⾃动和MQ确认3. callback: 回调对象
*/
DefaultConsumer consumer new DefaultConsumer(channel) {
/*回调⽅法, 当收到消息后, 会⾃动执⾏该⽅法1. consumerTag: 标识2. envelope: 获取⼀些信息, 交换机, 路由key3. properties:配置信息4. body:数据
*/
Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(接收到消息: new String(body));}
};
channel.basicConsume(hello, true, consumer);5.3.2 释放资源
//等待回调函数执⾏完毕之后, 关闭资源
TimeUnit.SECONDS.sleep(5);
//7. 释放资源 消费者相当于是⼀个监听程序, 不需要关闭资源
channel.close();
connection.close();实际上消费者相当于是⼀个监听程序, 不需要关闭资源 5.3.3 运行代码, 观察结果
运行程序, 我们刚才发送的消息, 就收到了 接收到消息: Hello World 如果我们不释放资源, 可以看到响应的 Connection, channel 5.4 附源码
生产者代码
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class RabbitProducer {public static void main(String[] args) throws Exception {// 1. 创建连接⼯⼚ConnectionFactory factory new ConnectionFactory();//2. 设置参数factory.setHost(110.41.51.65);//ip 默认值localhostfactory.setPort(15673); //默认值5672factory.setVirtualHost(bite);//虚拟机名称, 默认 factory.setUsername(study);//⽤⼾名,默认guestfactory.setPassword(study);//密码, 默认guest//3. 创建连接ConnectionConnection connection factory.newConnection();//4. 创建channel通道Channel channel connection.createChannel();//5. 声明队列/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, MapString, Object arguments)1.queue: 队列名称2.durable: 是否持久化, 当mq重启之后, 消息还在3.exclusive:* 是否独占, 只能有⼀个消费者监听队列* 当Connection关闭时, 是否删除队列4.autoDelete: 是否⾃动删除, 当没有Consumer时, ⾃动删除掉5.arguments: ⼀些参数*///如果没有⼀个hello 这样的⼀个队列, 会⾃动创建, 如果有, 则不创建channel.queueDeclare(hello, true, false, false, null);//6. 通过channel发送消息到队列中/*basicPublish(String exchange, String routingKey,AMQP.BasicProperties props, byte[] body)1. exchange: 交换机名称, 简单模式下, 交换机会使⽤默认的2.routingKey: 路由名称, routingKey 队列名称3.props: 配置信息4.body: 发送消息的数据*/String msg Hello World;//使⽤的是内置交换机. 使⽤内置交换机时, routingKey要和队列名称⼀样, 才可以路由到对应的队列上去channel.basicPublish(, hello, null, msg.getBytes());//7.释放资源System.out.println(msg 消息发送成功);channel.close();connection.close();}
}消费者代码
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeUnit;public class RabbitmqConsumer {public static void main(String[] args) throws Exception {// 1. 创建连接⼯⼚ConnectionFactory factory new ConnectionFactory();//2. 设置参数factory.setHost(110.41.51.65);//ip 默认值localhostfactory.setPort(15673); //默认值5672factory.setVirtualHost(bite);//虚拟机名称, 默认 /factory.setUsername(study);//⽤⼾名,默认guestfactory.setPassword(study);//密码, 默认guest//3. 创建连接ConnectionConnection connection factory.newConnection();//4. 创建channel通道Channel channel connection.createChannel();//5. 声明队列/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, MapString, Object arguments)1.queue: 队列名称2.durable: 是否持久化, 当mq重启之后, 消息还在3.exclusive:* 是否独占, 只能有⼀个消费者监听队列* 当Connection关闭时, 是否删除队列4.autoDelete: 是否⾃动删除, 当没有Consumer时, ⾃动删除掉5.arguments: ⼀些参数*///如果没有⼀个hello 这样的⼀个队列, 会⾃动创建, 如果有, 则不创建channel.queueDeclare(hello, true, false, false, null);//6. 接收消息, 并消费/*basicConsume(String queue, boolean autoAck, Consumer callback)参数:1. queue: 队列名称2. autoAck: 是否⾃动确认, 消费者收到消息之后,⾃动和MQ确认3. callback: 回调对象*/DefaultConsumer consumer new DefaultConsumer(channel) {/*回调⽅法, 当收到消息后, 会⾃动执⾏该⽅法1. consumerTag: 标识2. envelope: 获取⼀些信息, 交换机, 路由key3. properties:配置信息4. body:数据*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(接收到消息: new String(body));}};channel.basicConsume(hello, true, consumer);//等待回调函数执⾏完毕之后, 关闭资源TimeUnit.SECONDS.sleep(5);//7. 释放资源 消费者相当于是⼀个监听程序, 不需要关闭资源//顺序不可改变// channel.close();// connection.close();}
}
