RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）

文章目录

  • HelloWord
  • 工作队列
    • 工作线程代码
    • 启动两个工作线程
    • 工作队列（生产者代码）
    • 工作队列（结果成功）
  • 消息应答
    • 自动应答
    • 手动消息应答
    • multiple的解释
    • 消息自动重新入队
    • 手动应答代码
      • 消息手动应答（生产者）
      • 消息手动应答（消费者）
      • 消息手动应答（结果成功）
  • RabbitMQ持久化
    • 队列实现持久化
    • 消息实现持久化
  • 不公平分发
  • 预取值

HelloWord

在下图中，“P”是我们的生产者，“C”是我们的消费者。中间的框是一个队列-RabbitMO.代表使用者保留的消息缓冲区
在这里插入图片描述
第一步：导入依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.yc</groupId>
  <artifactId>rabbitmq-hello</artifactId>
  <version>1.0-SNAPSHOT</version>
  <!--指定jdk编译版本-->
  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <configuration>
          <source>8</source>
          <target>8</target>
        </configuration>
      </plugin>
    </plugins>
  </build>
  <dependencies>
    <!--rabbitmq依赖客户端-->
    <dependency>
      <groupId>com.rabbitmq</groupId>
      <artifactId>amqp-client</artifactId>
      <version>5.0.0</version>
    </dependency>
    <!--操作文件源的一个依赖-->
    <dependency>
      <groupId>commons-io</groupId>
      <artifactId>commons-io</artifactId>
      <version>2.6</version>
    </dependency>
  </dependencies>
</project>

第二步：创建生产者

//生产者:发消息
public class Producer {
    //队列名称
    public static final String QUEUE_NAME = "hello";

    //发消息
    public static void main(String[] args) throws IOException, TimeoutException {
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //工厂IP 连接RabbitMQ的队列
        factory.setHost("192.168.80.128");
        //用户名
        factory.setUsername("admin");
        //密码
        factory.setPassword("123");

        //创建连接
        Connection connection = factory.newConnection();
        //获取信道
        Channel channel = connection.createChannel();
        //生成一个队列
        /*
        * 1.队列名称
        * 2.队列里面的消息是否持久化（磁盘）默认情况消息存储在内存中
        * 3.该队列是否只供一个消费者进行消费，是否进行消息共享，true可以多个消费者消费false:只能一个消费者消费
         *4.是否自动剧除最后一个消贫者端开连接以后该队列是否自动鹏除 true自动鹏除false不自动翮除
         * 5.其它参数*/
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //发消息
        String message = "hello world";//初次使用

        /*发送一个消息
        * 1.发送到哪个交换机
        * 2.路由的Key值是哪个，本次是队列的名称
        * 3.其它参数信息
        * 4.发送消息的消息体*/
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());

        System.out.println("消息发送完毕");
    }
}

第三步：创建消费者

//消费者 接收消息的
public class Consumer {
    //队列名称
    public static final String QUEUE_NAME="hello";
    //接收消息
    public static void main(String[] args) throws IOException, TimeoutException {
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.80.128");
        factory.setUsername("admin");
        factory.setPassword("123");
        Connection connection = factory.newConnection();

        Channel channel = connection.createChannel();

        //声明
        DeliverCallback deliverCallback=(consumerTag, message)->{
            System.out.println(new String(message.getBody()));
        };

        //取消消息时的回调
        CancelCallback cancelCallback = consumerTag->{
            System.out.println("消费消息被中断");
        };
        //消费者消费消息
        //1.消费哪个队列
        //2.消费成功之后是否要自动应答
        //3.消费者未成功消费的回调
        //4.消费者取消消费的回调
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

工作队列

在这里插入图片描述
因为你为了确保同一条消息被其中一个工作线程接收到了之后，其它工作就不能消费的到了
三者之间的关系必须是竞争的关系
因为
在这里插入图片描述
这部分代码来来回回都是重复的，所以我们可以抽取连接工厂工具类

public class RabbitMqUtils {
    //得到一个连接的channel
    public static Channel getChannel() throws Exception{
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.80.128");
        factory.setUsername("admin");
        factory.setPassword("123");
        Connection connection = factory.newConnection();

        Channel channel = connection.createChannel();

        return channel;
    }
}

工作线程代码

//这是一个工作线程（相当于之前的消费者）
public class Worker01 {
    //队列的名称
    public  static final String QUEUE_NAME = "hello";

    //接收消息
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();

        DeliverCallback deliverCallback=(consumerTag, message)->{
            System.out.println(new String(message.getBody()));
        };

        //取消消息时的回调
        CancelCallback cancelCallback = consumerTag->{
            System.out.println("消费消息被中断");
        };

        //消息的接收
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

启动两个工作线程

在这里插入图片描述

工作队列（生产者代码）

public class Task01 {
    //队列名称
    public static final String QUEUE_NAME = "hello";

    //发送大量消息
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMqUtils.getChannel();

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //从控制台当中接受信息
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()){
            String message = scanner.next();
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            System.out.println("发送消息完成");
        }
    }
}

工作队列（结果成功）

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

消息应答

我们都知道消费者它完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了，会发生什么情况？
为了保证消息在发送过程中不丢失，rabbitmq_引入消息应答机制，消息应答就是:消费者在接收到消息并且处理该消息之后，告诉rabbitmq它已经处理了，rabbitmq可以把该消息删除了。

自动应答

消息发送后，立即被认为已经传送成功了，这种模式需要在高吞吐量和数据传输安全性方面做权衡，使得内存耗尽，最终这些消费者线程被操作系统杀死，这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。

手动消息应答

  • A.Channel. basicAck(用于肯定确认)
    RabbitMQ已知道该消息并且成功的处理消息，可以将其丢弃了
  • B.Channel. basicNack(用于否定确认)
  • C.Channel. basicReject(用于否定确认)
    与Channel. basicNack相比少一个参数不处理该消息了直接拒绝，可以将其丢弃了

multiple的解释

multiple 的 true和 false 代表不同意思
true 代表批量应答channel上未应答的消息
比如说channel上有传送tag 的消息5,6,7,8 当前tag是8那么此时5-8的这些还未应答的消息都会被确认收到消息应答
false同上面相比
只会应答 tag=8的消息5,6,7这三个消息依然不会被确认收到消息应答
在这里插入图片描述

消息自动重新入队

如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或TCP连接丢失)，导致消息未发送ACK确认，RabbitMQ将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。
在这里插入图片描述

手动应答代码

消息手动应答（生产者）

public class Task2 {
    //队列名称
    public static final String TASK_QUEUE_NAME = "ack_queue";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();

        //声明队列
        channel.queueDeclare(TASK_QUEUE_NAME,false,false,false,null);
        //从控制台中输入信息
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()){
            String message = scanner.next();
            channel.basicPublish("",TASK_QUEUE_NAME,null,message.getBytes());
            System.out.println("生产者发出消息："+message);
        }
    }
}

消息手动应答（消费者）

public class Work03 {
    //队列名称
    public static final String TASK_QUEUE_NAME="ack_queue";

    //接收消息
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        System.out.println("C1等待接收消息处理时间较短");

        DeliverCallback deliverCallback = (consumeTag,message)->{
            //沉睡1S
            try {
                SleepUtils.sleep(1);
                System.out.println("接收到的消息:"+new String(message.getBody(),"UTF-8"));
                //手动应答
                //1.消息的标记 tag
                //2. 是否批量应答 false：不批量应答通信道中的消息 true：批量
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };
        //采用手动应答
        boolean antoAck = false;
        channel.basicConsume(TASK_QUEUE_NAME,antoAck,deliverCallback,(consumerTag->{
            System.out.println(consumerTag+"消费者取消消费接口回调逻辑");
        }));
    }
}

public class Work04 {
    //队列名称
    public static final String TASK_QUEUE_NAME="ack_queue";

    //接收消息
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        System.out.println("C2等待接收消息处理时间较短");

        DeliverCallback deliverCallback = (consumeTag,message)->{
            //沉睡1S
            try {
                SleepUtils.sleep(30);
                System.out.println("接收到的消息:"+new String(message.getBody(),"UTF-8"));
                //手动应答
                //1.消息的标记 tag
                //2. 是否批量应答 false：不批量应答通信道中的消息 true：批量
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };
        //采用手动应答
        boolean antoAck = false;
        channel.basicConsume(TASK_QUEUE_NAME,antoAck,deliverCallback,(consumerTag->{
            System.out.println(consumerTag+"消费者取消消费接口回调逻辑");
        }));
    }
}

消息手动应答（结果成功）

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

RabbitMQ持久化

刚刚我们已经看到了如何处理任务不丢失的情况，但是如何保障当Rabbi1MQJ服务停掉以后消息生产者发送过来的消息不丢失。默认情况下RahbitMQ退出或由于某种原因崩溃时，它忽视队列和消息，除非告知它不要这样做。确保消息不会丢尖需要做两件事:我们需要将队列和消息都标记为持久化。
当队列持久化的时候
在这里插入图片描述
此处会显示D，这个时候即使重启rabbitmq队列消息也依然存在,但是需要注意的就是如果之前声明的队列不是持久化的，需要把原先队列先删除，或者重新创建一个持久化的队列，不然就会出现错误

队列实现持久化


public class Task2 {
    //队列名称
    public static final String TASK_QUEUE_NAME = "ack_queue";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();

        //声明队列
        boolean durable = true;//需要让Queue进行持久化
        channel.queueDeclare(TASK_QUEUE_NAME,durable,false,false,null);
        //从控制台中输入信息
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()){
            String message = scanner.next();
            channel.basicPublish("",TASK_QUEUE_NAME,null,message.getBytes());
            System.out.println("生产者发出消息："+message);
        }
    }
}

消息实现持久化

我们上面说到的队列持久化，只能保证队列不丢失，但不能保证消息不丢失，所以我们还需要给消息添加一个持久化，要想让消息实现持久化需要在消息生产者修改代码，MessageProperties.PERSISTENT_TEXT_PLAIN添加这个属性
在这里插入图片描述
将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉RabbitMQ将消息保存到磁盘，但是这里依然存在当消息刚准备存储在磁盘的时候但是还没有存储完，消息还在缓存的一个间隔点。此时并没有真正写入磁盘。持久性保证并不强

不公平分发

在最开始的时候我们学习到RabbitMQ分发消息采用的轮训分发,但是在某种场景下这种策略并不是很好，比方说有两个消费者在处理任务，其中有个消费者1处理任务的速度非常快，而另外一个消费者2处理速度却很慢，这个时候我们还是采用轮训分发的化就会到这处理速度快的这个消费者很大一部分时间处于空闲状态，而处理慢的那个消费者一直在干活,这种分配方式在这种情况下其实就不太好，但是RabbitMQ并不知道这种情况它依然很公平的进行分发。
为了避免这种情况，我们可以设置参数channel.basicQos(1);
在这里插入图片描述
当这里设置成1的时候，就说明现在这个队列是不公平分发
如何更改不公平分发呢？
在这里插入图片描述
我们只需给消费者设置这样一个参数

预取值

在这里插入图片描述
这里如果是0的话是轮训分发，1的话是不公平分发，其它大于1值的话就是预取值,可以事先规定好给该队列分配几条数据
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/19297.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

分布式系统概念和设计——命名服务设计和落地经验

分布式系统概念和设计——命名服务设计和落地经验

分布式系统概念和设计 通过命名服务&#xff0c;客户进程可以根据名字获取资源或对象的地址等属性。 被命名的实体可以是多种类型&#xff0c;并且可由不同的服务管理。 命名服务 命名是一个分布式系统中的非常基础的问题&#xff0c;名字在分布式系统中代表了广泛的资源&#…
阅读更多...
企业官方网站怎么申请？

企业官方网站怎么申请？

在数字化时代&#xff0c;企业官方网站是展示企业形象、宣传产品和服务的重要窗口。那么&#xff0c;企业官方网站怎么申请呢&#xff1f;下面是一些简单的步骤。 1、选择合适的网站建设平台 目前市面上有许多网站建设平台&#xff0c;企业需要根据自己的需求和预算选择适合自…
阅读更多...
公司新来个卷王，让人崩溃...

公司新来个卷王，让人崩溃...

最近内卷严重&#xff0c;各种跳槽裁员&#xff0c;相信很多小伙伴也在准备今年的面试计划。 在此展示一套学习笔记 / 面试手册&#xff0c;年后跳槽的朋友可以好好刷一刷&#xff0c;还是挺有必要的&#xff0c;它几乎涵盖了所有的软件测试技术栈&#xff0c;非常珍贵&#x…
阅读更多...
电力系统储能调峰、调频模型研究（Matlab代码实现）

电力系统储能调峰、调频模型研究（Matlab代码实现）

&#x1f4a5;&#x1f4a5;&#x1f49e;&#x1f49e;欢迎来到本博客❤️❤️&#x1f4a5;&#x1f4a5; &#x1f3c6;博主优势&#xff1a;&#x1f31e;&#x1f31e;&#x1f31e;博客内容尽量做到思维缜密&#xff0c;逻辑清晰&#xff0c;为了方便读者。 ⛳️座右铭&a…
阅读更多...
算法修炼之练气篇——练气二十一层

算法修炼之练气篇——练气二十一层

博主&#xff1a;命运之光 专栏&#xff1a;算法修炼之练气篇 前言&#xff1a;每天练习五道题&#xff0c;炼气篇大概会练习200道题左右&#xff0c;题目有C语言网上的题&#xff0c;也有洛谷上面的题&#xff0c;题目简单适合新手入门。&#xff08;代码都是命运之光自己写的…
阅读更多...
程序员面试金典16.*

程序员面试金典16.*

文章目录 16.01 交换数字16.02单词频率16.03交点16.04 井字游戏16.05 阶乘尾数16.06 最小差16.07 最大数值16.08 整数的英文表示16.09 运算16.10 生存人数16.11 跳水板16.13 平分正方形16.14 最佳直线&#xff08;待定&#xff09;16.15珠玑妙算16.16部分排序16.17连续数列16.1…
阅读更多...
10:00面试，10:04就出来了 ，问的实在是太...

10:00面试，10:04就出来了 ，问的实在是太...

从外包出来&#xff0c;没想到竟然死在了另一家厂子 自从加入这家公司&#xff0c;每天都在加班&#xff0c;钱倒是给的不少&#xff0c;所以我也就忍了。没想到12月一纸通知&#xff0c;所有人都不许加班&#xff0c;薪资直降30%&#xff0c;顿时有吃不起饭的赶脚。 好在有个…
阅读更多...
OpenPCDet系列 | 6.PointPillars模型分类、回归、角度损失的构建

OpenPCDet系列 | 6.PointPillars模型分类、回归、角度损失的构建

文章目录 模型损失计算1. 分类损失构建1.1 分类损失函数&#xff1a;SigmoidFocalClassificationLoss 2. 回归损失构建2.1 回归损失函数&#xff1a;WeightedSmoothL1Loss 3. 角度损失构建3.1 角度损失函数&#xff1a;WeightedCrossEntropyLoss 4. 总结 模型损失计算 在进行a…
阅读更多...
如何判断CRM软件的好坏?2023年CRM系统排行榜前三名是什么？

如何判断CRM软件的好坏?2023年CRM系统排行榜前三名是什么？

CRM客户管理系统经过20余年的发展&#xff0c;收获了越来越多企业的认可&#xff0c;成为企业数字化转型必不可少的一环。很多企业都有上线CRM软件的计划&#xff0c;但精准的找到一款适合自身的产品十分不易&#xff0c;今天我们就来盘点2023年CRM软件排行榜。 一、CRM的含义…
阅读更多...
【跟着陈七一起学C语言】今天总结：初识C语言

【跟着陈七一起学C语言】今天总结：初识C语言

友情链接&#xff1a;专栏地址 知识总结顺序参考C Primer Plus&#xff08;第六版&#xff09;和谭浩强老师的C程序设计&#xff08;第五版&#xff09;等&#xff0c;内容以书中为标准&#xff0c;同时参考其它各类书籍以及优质文章&#xff0c;以至减少知识点上的错误&#x…
阅读更多...
Speech and Language Processing之word2vec

Speech and Language Processing之word2vec

1、介绍 事实证明&#xff0c;在每一个NLP任务中&#xff0c;密集向量都比稀疏向量工作得更好。虽然我们不能完全理解其中的所有原因&#xff0c;但我们有一些直觉。首先&#xff0c;密集向量可以更成功地作为特征包含在机器学习系统中;例如&#xff0c;如果我们使用100维…
阅读更多...
如何高清视频录制？您只需要这样操作！

如何高清视频录制？您只需要这样操作！

案例&#xff1a;如何录制画质高清的视频&#xff1f; 【我录制了一个视频课程&#xff0c;上传到网上&#xff0c;但是我录制的视频画质不好&#xff0c;影响观感。有没有支持高清录制的录屏工具&#xff1f;有没有小伙伴可以推荐一下&#xff01;在线等&#xff01;】 无论…
阅读更多...
BM61-矩阵最长递增路径

BM61-矩阵最长递增路径

题目 给定一个 n 行 m 列矩阵 matrix &#xff0c;矩阵内所有数均为非负整数。 你需要在矩阵中找到一条最长路径&#xff0c;使这条路径上的元素是递增的。并输出这条最长路径的长度。 这个路径必须满足以下条件&#xff1a; 对于每个单元格&#xff0c;你可以往上&#xff…
阅读更多...
数组的应用

数组的应用

数组的应用 一、数组的定义二、切片替换删除数值元素 二、数组追加元素三、数组与函数相结合 一、数组的定义 相当于一串数据的集合&#xff0c;以空格相间隔的字符串列表&#xff0c;两边用括号括起来 echo ${shuzu[]}中的代表着显示所有的下标内容&#xff0c;当然&#…
阅读更多...
【C++】关联式容器——mapset的使用

【C++】关联式容器——mapset的使用

文章目录 1.关联式容器和键值对1. 关联式容器2. 键值对 2. 树形结构的关联式容器——set1. 模版参数列表2. 默认成员函数3. 迭代器4.容量相关操作5.modify6.其他操作接口 3. 树形结构的关联式容器——map1. 默认成员函数2. 迭代器3. 容量与数据访问4.数据修改5. 其他操作接口 1…
阅读更多...
初识C语言

初识C语言

1. 初识C语言 C语言是一门通用计算机编程语言&#xff0c;广泛应用于底层开发。 C语言是一门面向过程的计算机编程语言&#xff0c;它与C,Java等面向对象的编程语言有所不同。 第一个C语言程序&#xff1a; #include<stdio.h>int main(void) {printf("hello worl…
阅读更多...
MyBatis基础知识点总结

MyBatis基础知识点总结

MyBatis了解 MyBatis 是什么&#xff1f; MyBatis 是支持定制化 SQL、存储过程以及高级映射的优秀的持久层框架 MyBatis 避免了几乎所有的 JDBC 代码和手动设置参数以及获取结果集 MyBatis 可以使用简单的XML或注解用于配置和原始映射&#xff0c;将接口和Java的 POJO&#x…
阅读更多...
JVM-内存结构

JVM-内存结构

✅作者简介&#xff1a;热爱Java后端开发的一名学习者&#xff0c;大家可以跟我一起讨论各种问题喔。 &#x1f34e;个人主页&#xff1a;Hhzzy99 &#x1f34a;个人信条&#xff1a;坚持就是胜利&#xff01; &#x1f49e;当前专栏&#xff1a;JVM &#x1f96d;本文内容&…
阅读更多...
Mybatis一级缓存详解

Mybatis一级缓存详解

目录 一级缓存 一级缓存的组织 一级缓存的生命周期 一级缓存的工作流程 Cache接口的设计以及CacheKey的定义 一级缓存的性能分析 一级缓存与Spring 事务一级缓存存在的弊端 官方文档分析 Spring通过Mybatis调用数据库的过程 一级缓存 对于会话&#xff08;Session&am…
阅读更多...
chanmama响应数据解析

chanmama响应数据解析

0x00目标url aHR0cHM6Ly93d3cuY2hhbm1hbWEuY29tL2F1dGhvckRldGFpbC85OTI0MjExODcxOC9wcm9tb3Rpb24 0x01接口分析 简单的get 但是返回数据被加密了 这里我们就来想想怎么解密这些数据。首先后端发来的数据是加密的&#xff0c;但是我们在前端看到的可不是加密后的数据。前端…
阅读更多...
最新文章

Copyright @ 2022~2023