RabbitMQ之旅(1)

server/2025/3/9 5:05:44/

相信自己,终会成功

目录

主流MQ产品

1.kafaka 

2.RocketMQ

3.RabbitMQ

在xshell上安装RabbitMQ

 RabbitMQ七种工作模式

1.简单模式

​编辑

2.工作队列模式

3.发布/订阅模式

4.路由模式

5.通配符模式

6.RPC模式

AMQP.BasicProperties 设置消息属性的类

7.发布确认模式

代码展示(生产者)

常量命名规范

连接工厂 (ConnectionFactory) 

Channel（通道）

channel.queueDeclare 声明队列

channel.basicPublish 发送消息

channel.exchangeDeclare 声明创建交换机

channel.queueBind() 将队列绑定到交换机

channel.basicQos设置消费者预取限制

channel.basicAck 手动确认消息

代码展示(消费者)

DefaultConsumer：

handleDelivery 方法：

主流MQ产品

1.kafaka 

  1. 特点：高吞吐量、分布式、持久化、支持分区和副本。
  2. 适用场景：日志收集、流处理、实时数据分析等大数据场景。
  3. 优势：高吞吐量和低延迟，适合处理大量数据。
  4. 缺点：配置复杂，对小型项目可能过于重量级。

2.RocketMQ

  1. 特点：分布式、高吞吐量、低延迟、支持事务消息。
  2. 适用场景：电商、金融等需要高可靠性和事务支持的场景。
  3. 优势：支持事务消息，适合金融等高可靠性要求的场景。
  4. 缺点：社区相对较小，文档和资源不如Kafka丰富。

3.RabbitMQ

  1. 特点：轻量级、支持多种消息协议、易于使用和部署。
  2. 适用场景：中小型项目、需要快速上手的场景。
  3. 优势：易于使用，支持多种消息协议，社区活跃。
  4. 缺点：在大规模高并发场景下性能不如Kafka和RocketMQ。

在xshell上安装RabbitMQ

1.更新xshell中最新的软件包列表

sudo apt-get update

2.安装erlang

sudo apt-get install erlang

输入erl,出现下图内容表示安装成功  输入halt().退出即可

 3.安装rabbitmq

sudo apt-get install rabbitmq-server

4.确认安装结果

systemctl status rabbitmq-server

 显示running即可

 5.安装RabbitMQ管理界面

rabbitmq-plugins enable rabbiting_management

6.启动服务

sudo service rabbitmq-server start

在浏览器中输入自己云服务器的端口号+15672即可访问页面

添加用户名和密码 

rabbitmqctl add_user 用户名 密码

给用户权限

rabbitmqctl set_user_tags 用户名 权限等级

 RabbitMQ七种工作模式

P:生产者        C:消费者

queue:队列   X:交换机

在使用绑定的时候为 BindingKey

在发送消息的时候为RoutingKey

官方网站:RabbitMQ Tutorials | RabbitMQ

建立连接需要的信息:

1.IP   2.端口号   3.账号   4.密码   5. 虚拟主机

消费者代码:

1.创建连接  2.创建Channel  3.声明一个队列Queue  4.消费信息  5.释放资源

1.简单模式

一个生产者,一个消费者,点到点模式

2.工作队列模式

一个生产者,多个消费者

假如有十条队列消息,C1和C2是共同消费这10条消息,消息不会重复消费

3.发布/订阅模式

4.路由模式

订阅模式的变化形式

5.通配符模式

6.RPC模式

AMQP.BasicProperties 设置消息属性的类
属性名类型说明
contentTypeString消息内容的 MIME 类型（如 text/plain、application/json）。
contentEncodingString消息内容的编码方式（如 UTF-8）。
headersMap<String, Object>自定义消息头，用于传递额外信息。
deliveryModeInteger消息的持久化模式：1（非持久化）或 2（持久化）。
priorityInteger消息的优先级（0 到 9，数值越大优先级越高）。
correlationIdString关联 ID，通常用于 RPC 模式，匹配请求和响应。
replyToString响应队列的名称，通常用于 RPC 模式。
expirationString消息的过期时间（以毫秒为单位的字符串）。
messageIdString消息的唯一标识符。
timestampDate消息的时间戳。
typeString消息的类型标识符。
userIdString用户 ID，用于验证消息的发送者。
appIdString应用程序 ID，用于标识发送消息的应用程序。
//        AMQP.BasicProperties 是一个不可变类，因此需要通过其内部类 Builder 来创建对象。AMQP.BasicProperties props = new AMQP.BasicProperties().builder().correlationId(correlationID).replyTo(Constants.RPC_RESPONSE_QUEUE).build();

  1. 客户端：

    • 生成唯一的 correlationId。

    • 设置 replyTo 为响应队列的名称。

    • 发送消息到请求队列（Constants.RPC_REQUEST_QUEUE）。

    • 监听响应队列（Constants.RPC_RESPONSE_QUEUE），等待服务器返回结果。

  2. 服务器：

    • 监听请求队列（Constants.RPC_REQUEST_QUEUE）。

    • 处理请求后，将结果发送到客户端指定的响应队列（replyTo）。

    • 在响应消息中设置与请求相同的 correlationId。

  3. 客户端匹配响应：

    • 收到响应后，根据 correlationId 匹配对应的请求。

7.发布确认模式

是RabbitMQ消息可靠性保证的关键 

代码展示(生产者)

下图的代码Constants是自己写的 Java 常量

常量命名规范

1.常量名使用 全大写字母，并用下划线 _ 分隔单词（如 VIRTUALHOST 和 WORK_QUEUE）。

2.这是 Java 中的命名约定，便于区分常量和变量。 

//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST); //云服务器的IP地址connectionFactory.setPort(Constants.PORT); //需要提前开放端口号connectionFactory.setUsername(Constants.USER_NAME);//账号connectionFactory.setPassword(Constants.PASSWORD);  //密码connectionFactory.setVirtualHost(Constants.VIRTUALHOST); //虚拟主机//2. 开启信道Channel channel = connection.createChannel();//3. 声明交换机   使用内置的交换机

连接工厂 (ConnectionFactory) 

是一个设计模式中的“工厂类”，它的目的是隐藏创建连接的复杂细节（比如网络配置、认证信息等）。你可以通过这个工厂对象预先设置连接参数（如服务器地址、端口、用户名、密码等），然后通过它来生成具体的连接对象

Channel（通道）

通道 是建立在连接（Connection）之上的一个轻量级逻辑连接。一个连接（Connection）可以创建多个通道，每个通道可以独立地发送和接收消息。通道的设计是为了复用连接，避免频繁创建和销毁连接的开销。创建通道后，通常会用它来声明队列、发送消息或消费消息

channel.queueDeclare 声明队列
//4.声明队列channel.queueDeclare(Constants.WORK_QUEUE,true,false,false,null);/*** queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,*                                  Map<String, Object> arguments)*  参数说明:*  queue: 队列名称*  durable: 可持久化*  exclusive: 是否独占*  autoDelete: 是否自动删除*  arguments: 参数*/

channel.basicPublish 发送消息
        //5. 发送消息/*** basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)* 参数说明:* exchange: 交换机名称* routingKey: 内置交换机, routingkey和队列名称保持一致* props: 属性配置* body: 消息*/for (int i = 0; i < 10; i++) {String msg = "hello rabbitmq~"+i;channel.basicPublish("","hello", null, msg.getBytes());}


 //6. 资源释放channel.close();connection.close();
channel.exchangeDeclare 声明创建交换机
Exchange.DeclareOk exchangeDeclare(String exchange,              // 交换机名称String type,                  // 交换机类型（direct、fanout、topic、headers）boolean durable,              // 是否持久化boolean autoDelete,           // 是否自动删除boolean internal,            // 是否内部交换机Map<String, Object> arguments // 额外参数
) throws IOException;

channel.queueBind() 将队列绑定到交换机

  • queue：队列名称（如 Constants.PUBLISH_QUEUE1）。

  • exchange：交换机名称（如 Constants.PUBLISH_EXCHANGE）。

  • routingKey：路由键（如 "" 或 "key1"）

channel.queueBind(Constants.PUBLISH_QUEUE1, Constants.PUBLISH_EXCHANGE, "");
channel.queueBind(Constants.PUBLISH_QUEUE2, Constants.PUBLISH_EXCHANGE, "");
//作用：将 Constants.PUBLISH_QUEUE1 和 Constants.PUBLISH_QUEUE2 绑定到 //Constants.PUBLISH_EXCHANGE。
//路由键为空字符串（""），表示所有消息都会被路由到这两个队列（前提是交换机类型支持）。
channel.basicQos设置消费者预取限制
参数名类型说明
prefetchSizeint预取消息的总大小（以字节为单位），通常设置为 0 表示不限制大小。
prefetchCountint预取消息的数量限制（即未确认消息的最大数量）。
globalboolean是否全局生效：true 表示对整个 Channel 生效，false 表示对每个消费者生效。
// 设置每个消费者最多预取 10 条未确认消息
channel.basicQos(10);// 设置整个 Channel 最多预取 100 条未确认消息
channel.basicQos(100, true);// 设置预取消息的总大小不超过 1MB，数量不超过 10 条
channel.basicQos(1024 * 1024, 10, false);
channel.basicAck 手动确认消息
参数名类型说明
deliveryTaglong消息的唯一标识符，由 RabbitMQ 分配。
multipleboolean是否批量确认：true 表示确认所有比 deliveryTag 小的消息；false 表示仅确认当前消息。

使用场景

  • 手动确认模式：当消费者从队列中拉取消息时，如果启用了手动确认模式（autoAck=false），则必须调用 basicAck 来确认消息。

  • 确保消息可靠性：通过手动确认，可以确保消息在处理成功后才会从队列中删除，避免消息丢失。

  • 批量确认：通过设置 multiple=true，可以一次性确认多条消息，提高效率。

代码展示(消费者)

public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT); //需要提前开放端口号connectionFactory.setUsername(Constants.USER_NAME);//账号connectionFactory.setPassword(Constants.PASSWORD);  //密码connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虚拟主机Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();//3. 声明队列   使用内置的交换机//如果队列不存在, 则创建, 如果队列存在, 则不创建channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null);//4. 消费消息DefaultConsumer consumer = new DefaultConsumer(channel){//从队列中收到消息, 就会执行的方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到消息:"+ new String(body));}};channel.basicConsume(Constants.WORK_QUEUE, true, consumer);6. 资源释放channel.close();connection.close();}

DefaultConsumer：

RabbitMQ 提供的默认消费者类。

channel：消费者绑定的通道（Channel）。

匿名内部类：通过 new DefaultConsumer(channel) { ... } 创建一个匿名内部类，并写 handleDelivery 方法。

handleDelivery 方法：

当队列中有消息时，RabbitMQ 会调用此方法将消息传递给消费者

RabbitMQ 支持两种消息确认模式：

  1. 自动确认：

    • 在 basicConsume 方法中将第二个参数设置为 true。

    • 消费者接收到消息后，RabbitMQ 会自动将消息标记为已处理。

    • 示例：

      channel.basicConsume(QUEUE_NAME, true, consumer);

  2. 手动确认：

    • 在 basicConsume 方法中将第二个参数设置为 false。

    • 需要在 handleDelivery 方法中手动调用 channel.basicAck() 确认消息。

    • 示例：

      channel.basicAck(envelope.getDeliveryTag(), false);


http://www.ppmy.cn/server/173584.html

相关文章

设计模式-创建型模式详解

设计模式-创建型模式详解

这里写目录标题 一、基本概念二、单例模式1. 模式特点2. 适用场景3. 实现方法4. 经典示例 三、简单工厂模式1. 模式特点2. 经典示例 四、工厂方法模式五、抽象工厂模式1. 适用场景2. 经典示例 六、建造者模式1. 模式特点2. 一般流程3. 适用场景4. 经典示例 七、原型模式 一、基…
阅读更多...
无人机扩频技术对比！

无人机扩频技术对比！

一、技术原理与核心差异 FHSS&#xff08;跳频扩频&#xff09; 核心原理&#xff1a;通过伪随机序列控制载波频率在多个频点上快速跳变&#xff0c;收发双方需同步跳频序列。信号在某一时刻仅占用窄带频谱&#xff0c;但整体覆盖宽频带。 技术特点&#xff1a; 抗干扰…
阅读更多...
本地部署大语言模型-DeepSeek

本地部署大语言模型-DeepSeek

DeepSeek 是国内顶尖 AI 团队「深度求索」开发的多模态大模型&#xff0c;具备数学推理、代码生成等深度能力&#xff0c;堪称"AI界的六边形战士"。 Hostease AMD 9950X/96G/3.84T NVMe/1G/5IP/RTX4090 GPU服务器提供多种计费模式。 DeepSeek-R1-32B配置 配置项 规…
阅读更多...
用DeepSeek学Android开发：Android初学者遇到的常见问题有哪些？如何解决？

用DeepSeek学Android开发：Android初学者遇到的常见问题有哪些？如何解决？

本书定位于原生Android开发的入门书籍&#xff0c;力求做到看得懂、能练习、能学会&#xff0c;提供视频、源代码、课件、习题、试卷、教案等&#xff0c;免费在线答疑&#xff0c;教学两用 以下答案来自 DeepSeek Q: Android初学者遇到的常见问题有哪些&#xff1f;如何解决…
阅读更多...
STM32_IIC外设工作流程

STM32_IIC外设工作流程

STM32 IC 外设工作流程&#xff08;基于寄存器&#xff09; 在 STM32 中&#xff0c;IC 通信主要通过一系列寄存器控制。理解这些寄存器的作用&#xff0c;能够帮助我们掌握 IC 硬件的运行机制&#xff0c;实现高效的数据传输。本文以 STM32F1&#xff08;如 STM32F103&#x…
阅读更多...
在大型语言模型的提示词设计中，system、user和assistant三个角色的区别与联系

在大型语言模型的提示词设计中，system、user和assistant三个角色的区别与联系

在大型语言模型的提示词设计中，system、user和assistant三个角色承担不同的功能，其区别与联系如下： 1. 角色定义与功能 system（系统指令） 作用：设定模型的整体行为、角色定位和任务框架。例如，“你是一位专业的科技作家”或“仅回答与医疗相关的问题”。特点：在多轮对话…
阅读更多...
uniapp微信小程序vue3自定义tabbar

uniapp微信小程序vue3自定义tabbar

在App.vue隐藏原生tabbar&#xff0c;也可以在pages.json中配置 二选一就好了 创建 CustomTabBar 公共组件 <template><view class"custom-tab-bar" :style"{paddingBottom: safeAreaHeight px}"><view class"tab-bar-item" :…
阅读更多...
Java，Golang，Rust 泛型的大体对比小记

Java，Golang，Rust 泛型的大体对比小记

引子 在我重构JVM尝试实现泛型时&#xff0c;在阅读《深入理解JVM》这本书时&#xff0c;对Java泛型的实现原因和应用场景有了一定了解&#xff0c;书中也将其与C#泛型进行了简单对比。当我去年在学习Golang时&#xff0c;也发现了Golang在1.8后才提供了泛型&#xff0c;并不是…
阅读更多...
最新文章

Copyright @ 2022~2023