文章目录

• 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值的话就是预取值,可以事先规定好给该队列分配几条数据