1. Toy模板网首页
  2. > Toy博客

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

10月前 作者：杨宸杨 分类：Toy博客 阅读(42) 违法举报

这篇具有很好参考价值的文章主要介绍了RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

HelloWord

在下图中，“P”是我们的生产者，“C”是我们的消费者。中间的框是一个队列-RabbitMO.代表使用者保留的消息缓冲区
RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）
第一步：导入依赖

<?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);
    }
}

工作队列

RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）
因为你为了确保同一条消息被其中一个工作线程接收到了之后，其它工作就不能消费的到了
三者之间的关系必须是竞争的关系
因为
RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）
这部分代码来来回回都是重复的，所以我们可以抽取连接工厂工具类

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);
    }
}

启动两个工作线程

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

工作队列（生产者代码）

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 （HelloWord 消息应答 持久化 不公平分发 预取值）
RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）
RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）

消息应答

我们都知道消费者它完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了，会发生什么情况？
为了保证消息在发送过程中不丢失，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这三个消息依然不会被确认收到消息应答
RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）

消息自动重新入队

如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或TCP连接丢失)，导致消息未发送ACK确认，RabbitMQ将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。
RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）

手动应答代码

消息手动应答（生产者）

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 （HelloWord 消息应答 持久化 不公平分发 预取值）
RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）
RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）

RabbitMQ持久化

刚刚我们已经看到了如何处理任务不丢失的情况，但是如何保障当Rabbi1MQJ服务停掉以后消息生产者发送过来的消息不丢失。默认情况下RahbitMQ退出或由于某种原因崩溃时，它忽视队列和消息，除非告知它不要这样做。确保消息不会丢尖需要做两件事:我们需要将队列和消息都标记为持久化。
当队列持久化的时候
RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）
此处会显示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 （HelloWord 消息应答 持久化 不公平分发 预取值）
将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉RabbitMQ将消息保存到磁盘，但是这里依然存在当消息刚准备存储在磁盘的时候但是还没有存储完，消息还在缓存的一个间隔点。此时并没有真正写入磁盘。持久性保证并不强

不公平分发

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

预取值

RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）
这里如果是0的话是轮训分发，1的话是不公平分发，其它大于1值的话就是预取值,可以事先规定好给该队列分配几条数据
RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）文章来源地址https://www.toymoban.com/news/detail-412861.html

到了这里，关于RabbitMQ （HelloWord 消息应答 持久化 不公平分发 预取值）的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处： 如若内容造成侵权/违法违规/事实不符，请点击违法举报进行投诉反馈，一经查实，立即删除！

分享到：
领支付宝红包 赞助服务器费用

相关文章

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包，优惠每天领

二维码1

领取红包

二维码2

领红包