一、认识消息队列

1.1 什么是消息队列

消息：应用中传递的数据，可以是文本，对象等待。
消息队列：是一种应用间的通信方式，用于上下游传递消息，FIFO先入先出，消息发送上下游只需要依赖mq不用依赖其他服务。

1.2 消息队列的作用

  1. **峰

假如在一个订单系统中当应用服务器处理数据最大峰值是每秒一万单，当高峰期超过一万单就比较吃力，那么就可以使用消息队列，将一部分订单缓冲到队列中，后面再处理，让用户下单后十几秒后才收到下单成功的操作。

  1. **

当服务中A系统调用B系统，B需要花费很长的时间才能执行完成，一般情况下需要等待B执行完成后通知A，或者A定时查询B是否执行完成，使用MQ就可以让A监听B，只要B完成后发生一条消息被A监听到，这样A就能及时得到异步处理的结果；或者当A调用B系统,B系统整体逻辑与A没有关联，这个时候A系统可以将发送给B系统的任务发送到队列中，可以继续执行剩下的事情而不是等待。

  1. **

假如一个创建订单的过程需要调用订单系统，支付系统，库存系统等等；如果每个系统都是耦合在一起，任何一个子系统发生故障或者雪崩，整个下单就会中断，如果将各个子系统分开中间用MQ连接，当中间某个系统故障也会把下单进行下去，任务会存在队列中，故障被修复后会拉取队列中的任务，提高系统可用性。

1.3 mq的分类


 RabbitMQ ActiveMQ RocketMQ Kafka
公司/社区 Rabbit Apache 阿里 Apache
开发语言 Erlang Java Java Scala&Java
协议支持 AMQP，XMPP，SMTP，STOMP OpenWire,STOMP，REST,XMPP,AMQP 自定义协议 自定义协议
可用性 高 一般 高 高
单机吞吐量 一般 差 高 非常高
消息延迟 微秒级 毫秒级 毫秒级 毫秒以内
消息可靠性 高 一般 高 一般
社区活跃度 高 中 中 高

二、rabbitMQ简介

官网传送门
RabbitMQ是一个消息中间件：它接受并转发消息。你可以把它当做一个快递站点，当你要发送一个包裹时，你把你的包裹放到快递站，快递员最终会把你的快递送到收件人那里，按照这种逻辑RabbitMQ是一个快递站，一个快递员帮你传递快件。RabbitMQ与快递站的主要区别在于，它不处理快件而是接收，存储和转发消息数据。

2.1 rabbitMQ四大概念

生产者

生产数据，发送消息的程序就是生产者。

交换机

rabbitmq中重要的组件，接收来自生产者发送的消息；将消息推送到队列中，交换机必须知道如果处理这些消息，是将消息推送到单个队列还是多个队列，还是丢弃，这个由交换机的类型决定。

队列

队列是RabbitMQ内部使用的一种数据结构，尽管消息流经RabbitMQ和应用程序，但它们只能存储在队列中。队列仅受主机的内存和磁盘限制的约束，本质上是一个大的消息缓冲区。许多生产者可以将消息发送到一个队列，许多消费者可以尝试从一个队列接收数据。

消费者

消费与接收具有相似的含义。消费者大多时候是一个等待接收消息的程序。请注意生产者，消费者和消息中间件很多时候并不在同一机器上。同一个应用程序既可以是生产者又是可以是消费者。

2.2 rabbitMQ工作原理

生产端生产消息通过连接把消息传到信道里面，信道里的消息再传给交换机，交换机根据routing key将消息路由到不同的队列中，队列收到消息后就将消息通过连接传到消费端的信道里，然后消费者就收到消息了。
image.png

Broker：接收和分发消息的应用，RabbitMQ Server就是Message Broker
Virtual host：出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的namespace概念。当多个不同的用户使用同一个RabbitMQ server提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等
Connection：publisher／consumer和broker之间的TCP连接
Channel：如果每一次访问 RabbitMQ 都建立一个Connection，在消息量大的时候建立 TCP Connection的开销将是巨大的，效率也较低。Channel是在connection内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的channel进行通讯，AMQP method包含了channel id 帮助客户端和message broker 识别 channel，所以channel之间是完全隔离的。Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销
Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的类型有：直连direct (point-to-point), 主题topic (publish-subscribe) and 广播fanout (multicast)
Queue：消息最终被送到这里等待consumer取走
Binding：exchange和queue之间的虚拟连接，binding中可以包含routing key，Binding信息被保存到exchange中的查询表中，用于message的分发依据

2.3 docker 安装rabbitMQ

官方指导 带有management表示带有控制台

  1. docker run -di --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.9-management

image.png

三、简单入门

集成springboot的 pom依赖

  1. <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter-amqp</artifactId>
  4.  </dependency>

3.1 hello world

使用amqp 最基本的生产消费
生产者

  1. package com.rem.rabbitmq.ee.Aquick;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  7. /**
  8.  * 生产者
  9.  *
  10.  * @author Rem
  11.  * @date 2021-12-12
  12.  */
  14. public class Producer {
  16.     private final static String QUEUE_NAME = "hello";
  18.     public static void main(String[] args) throws Exception {
  19.         ConnectionFactory factory = new ConnectionFactory();
  20.         factory.setHost("106.14.72.13");
  21.         factory.setPort(5672);
  22.         factory.setUsername("rem");
  23.         factory.setPassword("123456");
  24.         Connection connection = factory.newConnection();
  25.         Channel channel = connection.createChannel();
  26.         try {
  27.             /**
  28.              *创建队列
  29.              *
  30.              *queue 队列名称
  31.              *durable 队列里的消息是否持久化 (服务重启是否存在) 默认在内存中，
  32.              *exclusive 如果我们声明一个独占队列（仅限于此连接），则为 true (只能有一个消费者)
  33.              * autoDelete 如果我们声明一个自动删除队列，则为 true（服务器将在不再使用时将其删除）
  34.              *arguments 队列的其他属性（构造参数）
  35.              */
  36.             channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  37.             String message = "hello world";
  38.             /**
  39.              *发送消息
  40.              *
  41.              *exchange 发送到哪个交换机 默认交换机 用空表示
  42.              *routingKey 路由键
  43.              *props 消息的其他属性 -路由标头等
  44.              * message 消息内容
  45.              */
  46.             channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
  47.             System.out.println("消息发送完毕");
  48.         } finally {
  49.             channel.close();
  50.             connection.close();
  51.         }
  53.     }
  54. }

消费者

  1. package com.rem.rabbitmq.ee.Aquick;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  6. import com.rabbitmq.client.ConnectionFactory;
  7. import com.rabbitmq.client.DeliverCallback;
  9. /**
  10.  * 消费者
  11.  *
  12.  * @author Rem
  13.  * @date 2021-12-12
  14.  */
  16. public class Consumer {
  19.     private final static String QUEUE_NAME = "hello";
  21.     public static void main(String[] args) throws Exception {
  23.         ConnectionFactory factory = new ConnectionFactory();
  24.         factory.setHost("106.14.72.13");
  25.         factory.setPort(5672);
  26.         factory.setUsername("rem");
  27.         factory.setPassword("123456");
  28.         Connection connection = factory.newConnection();
  29.         Channel channel = connection.createChannel();
  30.         try {
  32.             //消费回调
  33.             DeliverCallback deliverCallback = (consumerTag, message) -> {
  34.                 System.err.println("接收到的消息：" + new String(message.getBody()));
  35.             };
  37.             //取消消费回到 如果在消费的时候队列被删除了
  38.             CancelCallback cancelCallback = consumerTag -> {
  39.                 System.err.println("消费消息被中断");
  40.             };
  41.             /**
  42.              * queue – 队列的名称
  43.              * autoAck – 如果服务器应该考虑消息一旦发送就确认为真； 如果服务器应该期待显式确认，则为 false
  44.              * DeliverCallback – 传递消息时的回调
  45.              * cancelCallback – 消费者取消时的回调
  46.              */
  47.             channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
  48.         } finally {
  49.             channel.close();
  50.             connection.close();
  51.         }
  53.     }
  54. }

3.2 多个消费轮训接收消息

构造一个公共类

  1. /**
  2.      * 获取信道
  3.      *
  4.      * @return
  5.      * @throws Exception
  6.      */
  7.     public static Channel getChannel() throws Exception {
  8.         ConnectionFactory factory = new ConnectionFactory();
  9.         factory.setHost("106.14.72.13");
  10.         factory.setPort(5672);
  11.         factory.setUsername("rem");
  12.         factory.setPassword("123456");
  13.         connection = factory.newConnection();
  14.         channel = connection.createChannel();
  16.         return channel;
  17.     }
  20.     /**
  21.      * 关闭连接
  22.      *
  23.      * @throws Exception
  24.      */
  25.     public static void close() throws Exception {
  26.         if (channel.isOpen()) {
  27.             channel.close();
  28.         }
  29.         if (connection.isOpen()) {
  30.             connection.close();
  31.         }
  32.     }
  34.     public static void main(String[] args) throws Exception {
  35.         Channel channel = RabbitMqUtils.getChannel();
  36.         System.err.println(channel.isOpen());
  37.         Connection connection = channel.getConnection();
  38.         System.err.println(connection);
  39.         close();
  40.     }

生产者

  1. package com.rem.rabbitmq.ee.Brotation;
  3. import com.rabbitmq.client.Channel;
  4. import com.rem.rabbitmq.ee.RabbitMqUtils;
  6. import java.util.Scanner;
  9. /**
  10.  * 分发消息
  11.  * 两个consumer 轮训接收消息
  12.  *
  13.  * @author Rem
  14.  * @date 2021-12-23
  15.  */
  17. public class RotationProducer {
  19.     public static void main(String[] args) throws Exception {
  20.         Channel channel = RabbitMqUtils.getChannel();
  22.         //声明队列
  23.         channel.queueDeclare(RabbitMqUtils.QUEUE_ROTATION, false, false, false, null);
  25.         /**
  26.          *发送消息
  27.          *
  28.          *exchange 发送到哪个交换机 默认交换机 用空表示
  29.          *routingKey 路由键
  30.          *props 消息的其他属性 -路由标头等
  31.          * message 消息内容
  32.          */
  33.         Scanner scanner = new Scanner(System.in);
  34.         while (scanner.hasNext()) {
  35.             String message = scanner.next();
  36.             channel.basicPublish("", RabbitMqUtils.QUEUE_ROTATION, null, message.getBytes());
  37.             System.err.println("发送消息完毕" + message);
  38.         }
  39.     }
  40. }

消费者1

  1. package com.rem.rabbitmq.ee.Brotation;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  9. /**
  10.  * 消费者01
  11.  *
  12.  * @author Rem
  13.  * @date 2021-12-23
  14.  */
  16. public class Consumer01 {
  18.     public static void main(String[] args) throws Exception {
  20.         Channel channel = RabbitMqUtils.getChannel();
  21.         //消费回调
  22.         DeliverCallback deliverCallback = (consumerTag, message) -> {
  23.             System.out.println("接收到的消息：" + new String(message.getBody()));
  24.         };
  26.         //取消消费回调 如果在消费的时候队列被删除了
  27.         CancelCallback cancelCallback = consumerTag -> {
  28.             System.out.println("消费消息被中断");
  29.         };
  31.         System.out.println("消费者01 等待消费");
  32.         channel.basicConsume(RabbitMqUtils.QUEUE_ROTATION, true, deliverCallback, cancelCallback);
  33.     }
  34. }

消费者2

  1. package com.rem.rabbitmq.ee.Brotation;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  9. /**
  10.  * 消费者02
  11.  *
  12.  * @author Rem
  13.  * @date 2021-12-23
  14.  */
  16. public class Consumer02 {
  18.     public static void main(String[] args) throws Exception {
  20.         Channel channel = RabbitMqUtils.getChannel();
  22.         //消费回调
  23.         DeliverCallback deliverCallback = (consumerTag, message) -> {
  24.             System.out.println("接收到的消息：" + new String(message.getBody()));
  25.         };
  27.         //取消消费回调 如果在消费的时候队列被删除了
  28.         CancelCallback cancelCallback = consumerTag -> {
  29.             System.out.println("消费消息被中断");
  30.         };
  32.         System.out.println("消费者02 等待消费");
  33.         channel.basicConsume(RabbitMqUtils.QUEUE_ROTATION, true, deliverCallback, cancelCallback);
  34.     }
  35. }

3.3 消息应答ack

  1. package com.rem.rabbitmq.ee.Cack;
  3. import com.rabbitmq.client.Channel;
  4. import com.rem.rabbitmq.ee.RabbitMqUtils;
  6. import java.util.Scanner;
  9. /**
  10.  * 消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了，会发生什么情况。
  11.  * RabbitMQ一旦向消费者传递了一条消息，便立即将该消息标记为删除。
  12.  * 在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息。以及后续发送给该消费这的消息，因为它无法接收到。
  13.  * 为了保证消息在发送过程中不丢失，rabbitmq引入消息应答机制，
  14.  * 消息应答就是:消费者在接收到消息并且处理该消息之后，告诉rabbitmq它已经处理了，rabbitmq可以把该消息删除了。
  15.  * <p>
  16.  * <p>
  17.  * 消息应答类型:
  18.  * 1) 手动应答
  19.  * ********消费者处理完业务逻辑，手动返回ack（通知）告诉队列处理完了，队列进而删除消息。
  20.  * 2) 自动应答
  21.  * ********不在乎消费者对消息处理是否成功，都会告诉队列删除消息。如果处理消息失败，实现自动补偿（队列投递过去 重新处理）
  22.  * 这种模式需要在高吞吐量和数据传输安全性方面做权衡,
  23.  * ** a.channel或者连接关闭 消息会丢失
  24.  * ** b. 消息过载，并且对消息数量没有限制(或者消费者来不及处理) 消息积压，导致内存耗尽 消费者内存被系统杀死
  25.  * 所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用
  26.  * <p>
  27.  * <p>
  28.  * 消息应答类型方式:
  29.  * A.Channel.basicAck(用于肯定确认)         RabbitMQ已知道该消息并且成功的处理消息，可以将其丢弃了
  30.  * B.Channel.basicNack(用于否定确认)
  31.  * C.Channel.basicReject(用于否定确认) 与Channel.basicNack相比少一个参数,不处理该消息了直接拒绝，可以将其丢弃了
  32.  * <p>
  33.  * <p>
  34.  * Multiple: 批量处理
  35.  * true代表批量应答channel上未应答的消息  包含还未应答的消息都会被确认收到应答消息
  36.  * false 会应答当前tag的消息  之前的消息不会应答
  37.  * <p>
  38.  * <p>
  39.  * 消息自动重新入队
  40.  * 如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或TCP连接丢失)，导致消息未发送ACK确认，
  41.  * RabbitMQ将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。
  42.  * <p>
  43.  * 轮训消费消息时，即使其中有消费者在等待，也会均匀消费  当关闭其中一个消费者时会将为消费的消息转给另外一个消费者
  44.  * <p>
  45.  * <p>
  46.  *
  47.  * @author Rem
  48.  * @date 2021-12-23
  49.  */
  51. public class AckProducer {
  53.     public static void main(String[] args) throws Exception {
  54.         Channel channel = RabbitMqUtils.getChannel();
  56.         //声明队列
  57.         channel.queueDeclare(RabbitMqUtils.QUEUE_ACK, false, false, false, null);
  59.         System.err.println("开始发送消息...");
  60.         Scanner scanner = new Scanner(System.in);
  61.         while (scanner.hasNext()) {
  62.             String message = scanner.next();
  64.             //发送消息
  65.             channel.basicPublish("", RabbitMqUtils.QUEUE_ACK, null, message.getBytes());
  66.             System.err.println("发送消息完毕" + message);
  67.         }
  68.     }
  69. }
  1. package com.rem.rabbitmq.ee.Cack;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  8. import java.util.concurrent.TimeUnit;
  11. /**
  12.  *
  13.  * @author Rem
  14.  * @date 2021-12-23
  15.  */
  17. public class Consumer03 {
  19.     public static void main(String[] args) throws Exception {
  21.         Channel channel = RabbitMqUtils.getChannel();
  22.         //消费回调
  23.         DeliverCallback deliverCallback = (consumerTag, message) -> {
  24.             System.out.println("接收到的消息：" + new String(message.getBody()));
  26.             try {
  27.                 //休眠10s
  28.                 TimeUnit.SECONDS.sleep(10);
  29.             } catch (InterruptedException e) {
  30.                 e.printStackTrace();
  31.             }
  32.             /**
  33.              * DeliveryTag – 来自收到的AMQP.Basic.GetOk或AMQP.Basic.Deliver的标签
  34.              * 多个 – true 确认所有消息，包括提供的交付标签； false 仅确认提供的交付标签
  35.              */
  36.             channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
  37.         };
  39.         //取消消费回调 如果在消费的时候队列被删除了
  40.         CancelCallback cancelCallback = consumerTag -> {
  41.             System.out.println("消费消息被中断");
  42.         };
  43.         System.out.println("消费者03 等待消费");
  45.         //手动应答
  46.         boolean autoAck = false;
  47.         channel.basicConsume(RabbitMqUtils.QUEUE_ACK, autoAck, deliverCallback, cancelCallback);
  48.     }
  49. }
  1. package com.rem.rabbitmq.ee.Cack;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  8. import java.util.concurrent.TimeUnit;
  11. /**
  12.  * @author Rem
  13.  * @date 2021-12-23
  14.  */
  16. public class Consumer04 {
  18.     public static void main(String[] args) throws Exception {
  20.         Channel channel = RabbitMqUtils.getChannel();
  21.         //消费回调
  22.         DeliverCallback deliverCallback = (consumerTag, message) -> {
  23.             System.out.println("接收到的消息：" + new String(message.getBody()));
  25.             try {
  26.                 //休眠1s
  27.                 TimeUnit.SECONDS.sleep(1);
  28.             } catch (InterruptedException e) {
  29.                 e.printStackTrace();
  30.             }
  31.             /**
  32.              * DeliveryTag – 来自收到的AMQP.Basic.GetOk或AMQP.Basic.Deliver的标签
  33.              * 多个 – true 确认所有消息，包括提供的交付标签； false 仅确认提供的交付标签
  34.              */
  35.             channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
  36.         };
  38.         //取消消费回调 如果在消费的时候队列被删除了
  39.         CancelCallback cancelCallback = consumerTag -> {
  40.             System.out.println("消费消息被中断");
  41.         };
  42.         System.out.println("消费者04 等待消费");
  44.         //手动应答
  45.         boolean autoAck = false;
  46.         channel.basicConsume(RabbitMqUtils.QUEUE_ACK, autoAck, deliverCallback, cancelCallback);
  47.     }
  48. }

3.4 队列持久化

  1. package com.rem.rabbitmq.ee.Ddurable;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.MessageProperties;
  5. import com.rem.rabbitmq.ee.RabbitMqUtils;
  7. import java.util.Scanner;
  10. /**
  11.  * 队列持久化  重启mq服务队列依旧存在
  12.  * <p>
  13.  * <p>
  14.  * 如果队列已经持久化 再次创建就会报error
  15.  * channel error; protocol method: #method<channel.close>(reply-code=406,
  16.  * reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'QUEUE_ACK' in vhost '/': received 'true' but current is 'false', class-id=50, method-id=10)
  17.  *
  18.  * @author Rem
  19.  * @date 2021-12-23
  20.  */
  22. public class DurableProducer {
  24.     public static void main(String[] args) throws Exception {
  25.         Channel channel = RabbitMqUtils.getChannel();
  27.         /**
  28.          *创建队列
  29.          *
  30.          *queue 队列名称
  31.          *durable 队列里的消息是否持久化 (服务重启是否存在) 默认在内存中，  true 持久化
  32.          *exclusive 如果我们声明一个独占队列（仅限于此连接），则为 true
  33.          *autoDelete 如果我们声明一个自动删除队列，则为 true（服务器将在不再使用时将其删除）
  34.          *arguments 队列的其他属性（构造参数）
  35.          */
  36.         channel.queueDeclare(RabbitMqUtils.QUEUE_DURABLE, true, false, false, null);
  38.         System.err.println("开始发送消息...");
  39.         Scanner scanner = new Scanner(System.in);
  40.         while (scanner.hasNext()) {
  41.             String message = scanner.next();
  43.             /**
  44.              *发送消息
  45.              *
  46.              *exchange 发送到哪个交换机 默认交换机 用空表示
  47.              *routingKey 路由键
  48.              *props 消息的其他属性 -路由标头等
  49.              *
  50.              *MessageProperties.PERSISTENT_TEXT_PLAIN: 消息持久化
  51.              *
  52.              *
  53.              *message 消息内容
  54.              */
  55.             channel.basicPublish("", RabbitMqUtils.QUEUE_DURABLE, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
  56.             System.err.println("发送消息完毕" + message);
  57.         }
  58.     }
  59. }
  1. package com.rem.rabbitmq.ee.Ddurable;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  9. /**
  10.  * @author Rem
  11.  * @date 2021-12-23
  12.  */
  14. public class Consumer05 {
  16.     public static void main(String[] args) throws Exception {
  18.         Channel channel = RabbitMqUtils.getChannel();
  19.         //消费回调
  20.         DeliverCallback deliverCallback = (consumerTag, message) -> {
  21.             System.out.println("接收到的消息：" + new String(message.getBody()));
  22.         };
  24.         //取消消费回调 如果在消费的时候队列被删除了
  25.         CancelCallback cancelCallback = consumerTag -> {
  26.             System.out.println("消费消息被中断");
  27.         };
  28.         System.out.println("消费者05 等待消费");
  30.         //手动应答
  31.         boolean autoAck = true;
  32.         channel.basicConsume(RabbitMqUtils.QUEUE_DURABLE, autoAck, deliverCallback, cancelCallback);
  33.     }
  34. }

3.5 不公平分发

  1. package com.rem.rabbitmq.ee.Eunfair;
  3. import com.rabbitmq.client.Channel;
  4. import com.rem.rabbitmq.ee.RabbitMqUtils;
  6. import java.util.Scanner;
  9. /**
  10.  * 不公平分发
  11.  * 在消费者方 设置  channel.basicQos(1); 消费消息则不是轮训分发
  12.  * <p>
  13.  * <p>
  14.  * basicQos 设置的是缓存区大小 0 到 65535 即unack的消息数量
  15.  * 虽然自动应答传输消息速率是最佳的，但是，在这种情况下已传递但尚未处理的消息的数量也会增加，从而增加了消费者的RAM消耗
  16.  * <p>
  17.  * 消费者消费了大量的消息如果没有确认的话，会导致消费者连接节点的内存消耗变大，所以找到合适的预取值是一个反复试验的过程，
  18.  * 不同的负载该值取值也不同100到300范围内的值通常可提供最佳的吞吐量，并且不会给消费者带来太大的风险。预取值为1是最保守的。
  19.  * 当然这将使吞吐量变得很低，特别是消费者连接延迟很严重的情况下，特别是在消费者连接等待时间较长的环境中。对于大多数应用来说，稍微高一点的值将是最佳的
  20.  *
  21.  * @author Rem
  22.  * @date 2021-12-23
  23.  */
  25. public class UnfairProducer {
  27.     public static void main(String[] args) throws Exception {
  28.         Channel channel = RabbitMqUtils.getChannel();
  30.         //声明队列
  31.         channel.queueDeclare(RabbitMqUtils.QUEUE_UNFAIR, false, false, false, null);
  34.         System.err.println("开始发送消息...");
  35.         Scanner scanner = new Scanner(System.in);
  36.         while (scanner.hasNext()) {
  37.             String message = scanner.next();
  39.             //发送消息
  40.             channel.basicPublish("", RabbitMqUtils.QUEUE_UNFAIR, null, message.getBytes());
  41.             System.err.println("发送消息完毕" + message);
  42.         }
  43.     }
  44. }
  1. package com.rem.rabbitmq.ee.Eunfair;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  8. import java.util.concurrent.TimeUnit;
  11. /**
  12.  * @author Rem
  13.  * @date 2021-12-23
  14.  */
  16. public class Consumer06 {
  18.     public static void main(String[] args) throws Exception {
  20.         Channel channel = RabbitMqUtils.getChannel();
  21.         //消费回调
  22.         DeliverCallback deliverCallback = (consumerTag, message) -> {
  23.             System.out.println("接收到的消息：" + new String(message.getBody()));
  25.             try {
  26.                 //休眠10s
  27.                 TimeUnit.SECONDS.sleep(3);
  28.             } catch (InterruptedException e) {
  29.                 e.printStackTrace();
  30.             }
  31.             /**
  32.              * DeliveryTag – 来自收到的AMQP.Basic.GetOk或AMQP.Basic.Deliver的标签
  33.              * 多个 – true 确认所有消息，包括提供的交付标签； false 仅确认提供的交付标签
  34.              */
  35.             channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
  36.         };
  38.         //取消消费回到 如果在消费的时候队列被删除了
  39.         CancelCallback cancelCallback = consumerTag -> {
  40.             System.out.println("消费消息被中断");
  41.         };
  42.         System.out.println("消费者06 等待消费");
  43.         /**
  44.          * 不公平分发  预取值  0 到 65535
  45.          */
  46.         channel.basicQos(1);
  48.         //手动应答
  49.         boolean autoAck = false;
  50.         channel.basicConsume(RabbitMqUtils.QUEUE_UNFAIR, autoAck, deliverCallback, cancelCallback);
  51.     }
  52. }
  1. package com.rem.rabbitmq.ee.Eunfair;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  8. import java.util.concurrent.TimeUnit;
  11. /**
  12.  * @author Rem
  13.  * @date 2021-12-23
  14.  */
  16. public class Consumer07 {
  18.     public static void main(String[] args) throws Exception {
  20.         Channel channel = RabbitMqUtils.getChannel();
  21.         //消费回调
  22.         DeliverCallback deliverCallback = (consumerTag, message) -> {
  23.             System.out.println("接收到的消息：" + new String(message.getBody()));
  25.             try {
  26.                 //休眠1s
  27.                 TimeUnit.SECONDS.sleep(1);
  28.             } catch (InterruptedException e) {
  29.                 e.printStackTrace();
  30.             }
  31.             /**
  32.              * DeliveryTag – 来自收到的AMQP.Basic.GetOk或AMQP.Basic.Deliver的标签
  33.              * 多个 – true 确认所有消息，包括提供的交付标签； false 仅确认提供的交付标签
  34.              */
  35.             channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
  36.         };
  38.         //取消消费回调 如果在消费的时候队列被删除了
  39.         CancelCallback cancelCallback = consumerTag -> {
  40.             System.out.println("消费消息被中断");
  41.         };
  42.         System.out.println("消费者07 等待消费");
  45.         /**
  46.          * 不公平分发
  47.          */
  48.       //  channel.basicQos(1);
  49.         //手动应答
  50.         boolean autoAck = false;
  51.         channel.basicConsume(RabbitMqUtils.QUEUE_UNFAIR, autoAck, deliverCallback, cancelCallback);
  52.     }
  53. }

四、发布确认

单个发布确认

  1. package com.rem.rabbitmq.ee.Fconfirm;
  3. import com.rabbitmq.client.Channel;
  4. import com.rem.rabbitmq.ee.RabbitMqUtils;
  7. /**
  8.  * 发布确认
  9.  * 1.要求队列必须持久胡
  10.  * 2.要求队列里的消息必须持久化
  11.  * 3.发布确认
  12.  * <p>
  13.  * <p>
  14.  * 发布确认原理
  15.  * 所有在该信道上面发布的消息都将会被指派一个唯一的ID，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，
  16.  * 这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出
  17.  * <p>
  18.  * <p>
  19.  * 单个发布确认 （同步） 也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发  发布速度特别的慢
  20.  * 同步等待确认，简单，但吞吐量非常有限。
  21.  * waitForConfirms和waitForConfirmsOrDie的区别是异常后  waitForConfirmsOrDie信道被关闭
  22.  *
  23.  * @author Rem
  24.  * @date 2021-12-24
  25.  */
  27. public class SingleConfirmProducer {
  29.     private final static int MESSAGE_COUNT = 1005;
  31.     public static void main(String[] args) throws Exception {
  33.         try (Channel channel = RabbitMqUtils.getChannel();) {
  35.             //声明队列
  36.             channel.queueDeclare(RabbitMqUtils.QUEUE_CONFIRM, true, false, false, null);
  38.             /**
  39.              * 启用发布确认
  40.              */
  41.             channel.confirmSelect();
  43.             System.err.println("开始发送消息...");
  44.             long begin = System.currentTimeMillis();
  46.             for (int i = 0; i < MESSAGE_COUNT; i++) {
  47.                 String message = i + ":confirm";
  48.                 channel.basicPublish("", RabbitMqUtils.QUEUE_CONFIRM, null, message.getBytes());
  50.                 /**
  51.                  * 等到自上次调用以来发布的所有消息都被代理确认或确认。请注意，在非确认通道上调用时，waitForConfirms 会抛出 IllegalStateException。
  52.                  *
  53.                  * return 是否所有消息都被确认（并且没有消息被确认）服务端返回false或超时时间内未返回，生产者可以消息重发
  54.                  */
  56.                 // channel.waitForConfirmsOrDie(3000);
  57.                 boolean flag = channel.waitForConfirms();
  58.                 if (flag) {
  59.                     System.out.println(i + "消息发送成功");
  60.                 } else {
  61.                     System.out.println(i + "消息发送失败");
  62.                 }
  63.             }
  65.             long end = System.currentTimeMillis();
  66.             System.err.println("发布" + MESSAGE_COUNT + "个单独确认消息,耗时" + (end - begin) + "ms");
  67.         } finally {
  68.             RabbitMqUtils.close();
  69.         }
  71.     }
  72. }

批量发布确认

  1. package com.rem.rabbitmq.ee.Fconfirm;
  3. import com.rabbitmq.client.Channel;
  4. import com.rem.rabbitmq.ee.RabbitMqUtils;
  7. /**
  8.  * 发布确认
  9.  * 1.要求队列必须持久胡
  10.  * 2.要求队列里的消息必须持久化
  11.  * 3.发布确认
  12.  * <p>
  13.  * <p>
  14.  * 发布确认原理
  15.  * 所有在该信道上面发布的消息都将会被指派一个唯一的ID，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，
  16.  * 这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出
  17.  * <p>
  18.  * 批量确认  每发100个确认一次  这种方案仍然是同步的，也一样阻塞消息的发布
  19.  * 批量同步等待确认，简单，合理的吞吐量，一旦出现问题但很难推断出是哪条消息出现了问题。
  20.  *
  21.  * @author Rem
  22.  * @date 2021-12-24
  23.  */
  25. public class BatchConfirmProducer {
  27.     private final static int MESSAGE_COUNT = 1005;
  29.     public static void main(String[] args) throws Exception {
  31.         try (Channel channel = RabbitMqUtils.getChannel();) {
  33.             //声明队列
  34.             channel.queueDeclare(RabbitMqUtils.QUEUE_CONFIRM, true, false, false, null);
  35.             /**
  36.              * 启用发布确认
  37.              */
  38.             channel.confirmSelect();
  40.             System.err.println("开始发送消息...");
  41.             //批量确认消息大小
  42.             int batchSize = 100;
  43.             //未确认消息个数
  44.             int outstandingMessageCount = 0;
  45.             long begin = System.currentTimeMillis();
  47.             for (int i = 0; i < MESSAGE_COUNT; i++) {
  48.                 String message = i + ":confirm";
  49.                 channel.basicPublish("", RabbitMqUtils.QUEUE_CONFIRM, null, message.getBytes());
  50.                 /**
  51.                  * 等到自上次调用以来发布的所有消息都被代理确认或确认。请注意，在非确认通道上调用时，waitForConfirms 会抛出 IllegalStateException。
  52.                  *
  53.                  * return 是否所有消息都被确认（并且没有消息被确认）服务端返回false或超时时间内未返回，生产者可以消息重发
  54.                  */
  55.                 outstandingMessageCount++;
  56.                 if (outstandingMessageCount == batchSize) {
  57.                     channel.waitForConfirms();
  58.                     outstandingMessageCount = 0;
  59.                     System.out.println(i + "消息批量" + batchSize + "发送成功");
  61.                 }
  62.             }
  64.             /**
  65.              * 最后未被确认的 再次确认
  66.              */
  67.             if (outstandingMessageCount > 0) {
  68.                 channel.waitForConfirms();
  69.                 System.out.println("last 消息发送成功");
  70.             }
  72.             long end = System.currentTimeMillis();
  73.             System.err.println("发布" + MESSAGE_COUNT + "个批量确认消息,耗时" + (end - begin) + "ms");
  74.         } finally {
  75.             RabbitMqUtils.close();
  76.         }
  78.     }
  79. }

异步发布确认

  1. package com.rem.rabbitmq.ee.Fconfirm;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.ConfirmCallback;
  5. import com.rem.rabbitmq.ee.RabbitMqUtils;
  7. import java.util.concurrent.ConcurrentNavigableMap;
  8. import java.util.concurrent.ConcurrentSkipListMap;
  11. /**
  12.  * 发布确认
  13.  * 1.要求队列必须持久胡
  14.  * 2.要求队列里的消息必须持久化
  15.  * 3.发布确认
  16.  * <p>
  17.  * <p>
  18.  * 发布确认原理
  19.  * 所有在该信道上面发布的消息都将会被指派一个唯一的ID，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，
  20.  * 这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出
  21.  * <p>
  22.  * <p>
  23.  * 异步确认发布  是通过函数回调来保证是否投递成功 可靠和效率都最高，实现比较复杂  性价比最高
  24.  * confirm模式最大的好处在于他是异步的
  25.  * 当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息
  26.  *
  27.  * @author Rem
  28.  * @date 2021-12-24
  29.  */
  31. public class AsyncConfirmProducer {
  33.     private final static int MESSAGE_COUNT = 1005;
  35.     public static void main(String[] args) throws Exception {
  37.         try (Channel channel = RabbitMqUtils.getChannel();) {
  38.             //声明队列
  39.             channel.queueDeclare(RabbitMqUtils.QUEUE_CONFIRM, true, false, false, null);
  40.             /**
  41.              * 启用发布确认
  42.              */
  43.             channel.confirmSelect();
  44.             System.err.println("开始发送消息...");
  45.             /**
  46.              * 线程安全有序的哈希表，适用于高并发
  47.              *  1.轻松的将序列号与消息关联
  48.              *  2.轻松批量删除条目 只要给到序号
  49.              *  3.支持高并发
  50.              *
  51.              */
  52.             ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
  53.             /**
  54.              * 发布确认回调  消息确认成功
  55.              * deliveryTag 消息的标记
  56.              * multiple 是否为批量
  57.              */
  58.             ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
  59.                 ConcurrentNavigableMap<Long, String> confirmed = null;
  60.                 if (multiple) {
  61.                     //返回的是小于等于当前序列号的确认消息 是一个map
  62.                     confirmed = outstandingConfirms.headMap(deliveryTag);
  63.                 } else {
  64.                     outstandingConfirms.remove(deliveryTag);
  65.                 }
  66.                 if (confirmed != null) {
  67.                     confirmed.clear();
  68.                 }
  69.                 System.out.println("成功的消息" + deliveryTag);
  70.             };
  71.             /**
  72.              * 发布确认回调 消息确认失败
  73.              */
  74.             ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
  75.                 System.out.println("未确认的编号" + deliveryTag + "消息:" + outstandingConfirms.get(deliveryTag));
  76.             };
  77.             /**
  78.              * 消息监听器
  79.              */
  80.             channel.addConfirmListener(ackCallback, nackCallback);
  82.             long begin = System.currentTimeMillis();
  83.             for (int i = 0; i < MESSAGE_COUNT; i++) {
  84.                 String message = i + ":confirm";
  86.                 channel.basicPublish("", RabbitMqUtils.QUEUE_CONFIRM, null, message.getBytes());
  87.                 outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
  88.             }
  89.             long end = System.currentTimeMillis();
  90.             System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息,耗时" + (end - begin) + "ms");
  91.         } finally {
  92.             RabbitMqUtils.close();
  93.         }
  95.     }
  96. }

五、认识rabbitMQ交换机

广播交换机

  1. package com.rem.rabbitmq.ee.Gexchanges.fanout;
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rem.rabbitmq.ee.RabbitMqUtils;
  7. import java.util.Scanner;
  9. /**
  10.  * 交换机-广播模式
  11.  * 将接收到的所有消息广播到它知道的所有队列中
  12.  *
  13.  * @author Rem
  14.  * @date 2021-12-24
  15.  */
  17. public class FanoutExchangeProducer {
  20.     public static void main(String[] args) throws Exception {
  22.         Channel channel = RabbitMqUtils.getChannel();
  23.         /**
  24.          * 声明一个交换机
  25.          * exchange: 交换机的名称
  26.          * type:交换机的类型
  27.          *
  28.          */
  29.         channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_FANOUT, BuiltinExchangeType.FANOUT);
  31.         System.err.println("开始发送消息...");
  32.         Scanner scanner = new Scanner(System.in);
  33.         while (scanner.hasNext()) {
  34.             String message = scanner.next();
  35.             //发送消息
  36.             channel.basicPublish(RabbitMqUtils.EXCHANGE_FANOUT, "", null, message.getBytes());
  37.             System.err.println("发送消息完毕" + message);
  38.         }
  39.     }
  40. }
  1. package com.rem.rabbitmq.ee.Gexchanges.fanout;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  9. /**
  10.  * @author Rem
  11.  * @date 2021-12-23
  12.  */
  14. public class Consumer08 {
  16.     public static void main(String[] args) throws Exception {
  18.         Channel channel = RabbitMqUtils.getChannel();
  20.         //在生产者已经创建过
  21.         //channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_FANOUT, BuiltinExchangeType.FANOUT);
  23.         /**
  24.          * 创建一个临时队列  队列名称随机
  25.          * 当消费者断开和该队列的连接时，队列自动删除
  26.          */
  27.         String queue = channel.queueDeclare().getQueue();
  28.         /**
  29.          * 将临时队列绑定到交换机
  30.          */
  31.         channel.queueBind(queue, RabbitMqUtils.EXCHANGE_FANOUT, "");
  33.         //消费回调
  34.         DeliverCallback deliverCallback = (consumerTag, message) -> System.out.println("接收到的消息：" + new String(message.getBody()));
  36.         //取消消费回调 如果在消费的时候队列被删除了
  37.         CancelCallback cancelCallback = consumerTag -> System.out.println("消费消息被中断");
  39.         System.out.println("消费者08 等待消费");
  40.         channel.basicConsume(queue, true, deliverCallback, cancelCallback);
  41.     }
  42. }
  1. package com.rem.rabbitmq.ee.Gexchanges.fanout;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  9. /**
  10.  * @author Rem
  11.  * @date 2021-12-23
  12.  */
  14. public class Consumer09 {
  16.     public static void main(String[] args) throws Exception {
  18.         Channel channel = RabbitMqUtils.getChannel();
  20.         /**
  21.          * 创建一个临时队列  队列名称随机
  22.          * 当消费者断开和该队列的连接时，队列自动删除
  23.          */
  24.         String queue = channel.queueDeclare().getQueue();
  25.         /**
  26.          * 将临时队列绑定到交换机
  27.          */
  28.         channel.queueBind(queue, RabbitMqUtils.EXCHANGE_FANOUT, "");
  30.         //消费回调
  31.         DeliverCallback deliverCallback = (consumerTag, message) -> System.out.println("接收到的消息：" + new String(message.getBody()));
  33.         //取消消费回调 如果在消费的时候队列被删除了
  34.         CancelCallback cancelCallback = consumerTag -> System.out.println("消费消息被中断");
  36.         System.out.println("消费者09 等待消费");
  37.         channel.basicConsume(queue, true, deliverCallback, cancelCallback);
  38.     }
  39. }

直连交换机

  1. package com.rem.rabbitmq.ee.Gexchanges.direct;
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rem.rabbitmq.ee.RabbitMqUtils;
  7. import java.util.HashMap;
  8. import java.util.Map;
  9. import java.util.Scanner;
  11. /**
  12.  * 交换机-直连模式
  13.  * 将消息发送到绑定的routingkey的队列中去 没有绑定的消息会被丢弃
  14.  *
  15.  * @author Rem
  16.  * @date 2021-12-24
  17.  */
  19. public class DirectExchangeProducer {
  22.     public static void main(String[] args) throws Exception {
  24.         Channel channel = RabbitMqUtils.getChannel();
  25.         /**
  26.          * 声明一个交换机
  27.          * exchange: 交换机的名称
  28.          * type:交换机的类型
  29.          *
  30.          */
  31.         channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_DIRECT, BuiltinExchangeType.DIRECT);
  33.         System.err.println("开始发送消息...");
  34.         Scanner scanner = new Scanner(System.in);
  37.         /**
  38.          * 创建多个bindingKey
  39.          * 没有与routingKey绑定的消费者 消息发出后会丢失
  40.          */
  41.         Map<String, String> routingKeyMap = new HashMap<>();
  42.         routingKeyMap.put("1", "green");
  43.         routingKeyMap.put("2", "blue");
  44.         routingKeyMap.put("3", "red");
  45.         routingKeyMap.put("4", "pink");
  48.         while (scanner.hasNext()) {
  49.             String message = scanner.next();
  50.             String routingKey;
  51.             for (Map.Entry<String, String> entry : routingKeyMap.entrySet()) {
  52.                 String key = entry.getKey();
  53.                 if (message.equals(key)) {
  54.                     routingKey = entry.getValue();
  55.                     //发送消息
  56.                     channel.basicPublish(RabbitMqUtils.EXCHANGE_DIRECT, routingKey, null, message.getBytes());
  57.                     System.err.println("发送消息完毕" + message);
  58.                 }
  59.             }
  60.         }
  61.     }
  62. }
  1. package com.rem.rabbitmq.ee.Gexchanges.direct;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  9. /**
  10.  * @author Rem
  11.  * @date 2021-12-23
  12.  */
  14. public class Consumer10 {
  16.     public static void main(String[] args) throws Exception {
  18.         Channel channel = RabbitMqUtils.getChannel();
  20.         /**
  21.          * 创建一个临时队列  队列名称随机
  22.          * 当消费者断开和该队列的连接时，队列自动删除
  23.          */
  24.         String queue = channel.queueDeclare().getQueue();
  25.         /**
  26.          * 将临时队列绑定到交换机
  27.          */
  28.         channel.queueBind(queue, RabbitMqUtils.EXCHANGE_DIRECT, "green");
  29.         channel.queueBind(queue, RabbitMqUtils.EXCHANGE_DIRECT, "red");
  31.         //消费回调
  32.         DeliverCallback deliverCallback = (consumerTag, message) -> System.out.println("接收到的消息：" + new String(message.getBody()));
  34.         //取消消费回调 如果在消费的时候队列被删除了
  35.         CancelCallback cancelCallback = consumerTag -> System.out.println("消费消息被中断");
  37.         System.out.println("消费者10 等待消费");
  38.         channel.basicConsume(queue, true, deliverCallback, cancelCallback);
  39.     }
  42. }
  1. package com.rem.rabbitmq.ee.Gexchanges.direct;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  9. /**
  10.  * @author Rem
  11.  * @date 2021-12-23
  12.  */
  14. public class Consumer11 {
  16.     public static void main(String[] args) throws Exception {
  18.         Channel channel = RabbitMqUtils.getChannel();
  20.         /**
  21.          * 创建一个临时队列  队列名称随机
  22.          * 当消费者断开和该队列的连接时，队列自动删除
  23.          */
  24.         String queue = channel.queueDeclare().getQueue();
  25.         /**
  26.          * 将临时队列绑定到交换机
  27.          */
  28.         channel.queueBind(queue, RabbitMqUtils.EXCHANGE_DIRECT, "blue");
  30.         //消费回调
  31.         DeliverCallback deliverCallback = (consumerTag, message) -> System.out.println("接收到的消息：" + new String(message.getBody()));
  33.         //取消消费回调 如果在消费的时候队列被删除了
  34.         CancelCallback cancelCallback = consumerTag -> System.out.println("消费消息被中断");
  36.         System.out.println("消费者11 等待消费");
  37.         channel.basicConsume(queue, true, deliverCallback, cancelCallback);
  40.     }
  41. }

主题交换机

  1. package com.rem.rabbitmq.ee.Gexchanges.topic;
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rem.rabbitmq.ee.RabbitMqUtils;
  7. import java.util.HashMap;
  8. import java.util.Map;
  9. import java.util.Scanner;
  11. /**
  12.  * 交换机-主题模式  必须是一个单词列表，以点号分隔开 最多不能超过255个字节
  13.  * routingkey 匹配规则
  14.  * *(星号)可以代替一个单词
  15.  * #(井号)可以替代零个或多个单词
  16.  * <p>
  17.  * <p>
  18.  * 将消息发送到绑定的routingkey的队列中去 没有绑定的消息会被丢弃
  19.  *
  20.  * @author Rem
  21.  * @date 2021-12-24
  22.  */
  24. public class TopicExchangeProducer {
  27.     public static void main(String[] args) throws Exception {
  29.         Channel channel = RabbitMqUtils.getChannel();
  30.         /**
  31.          * 声明一个交换机
  32.          * exchange: 交换机的名称
  33.          * type:交换机的类型
  34.          *
  35.          */
  36.         channel.exchangeDeclare(RabbitMqUtils.EXCHANGE_TOPIC, BuiltinExchangeType.TOPIC);
  38.         System.err.println("开始发送消息...");
  39.         Scanner scanner = new Scanner(System.in);
  42.         /**
  43.          * 创建多个bindingKey
  44.          * 没有与routingKey绑定的消费者 消息发出后会丢失
  45.          */
  46.         Map<String, String> routingKeyMap = new HashMap<>();
  47.         routingKeyMap.put("1", "green.blue.red.pink");
  48.         routingKeyMap.put("2", "green.blue.red");
  49.         routingKeyMap.put("3", "green.blue");
  50.         routingKeyMap.put("4", "green");
  51.         routingKeyMap.put("5", "blue.red.pink");
  52.         routingKeyMap.put("6", "blue.red");
  53.         routingKeyMap.put("7", "blue");
  54.         routingKeyMap.put("8", "red.pink");
  55.         routingKeyMap.put("9", "pink");
  58.         while (scanner.hasNext()) {
  59.             String message = scanner.next();
  60.             String routingKey;
  61.             for (Map.Entry<String, String> entry : routingKeyMap.entrySet()) {
  62.                 String key = entry.getKey();
  63.                 if (message.equals(key)) {
  64.                     routingKey = entry.getValue();
  65.                     //发送消息
  66.                     channel.basicPublish(RabbitMqUtils.EXCHANGE_TOPIC, routingKey, null, message.getBytes());
  67.                     System.err.println("发送消息完毕" + message);
  68.                 }
  69.             }
  70.         }
  71.     }
  72. }
  1. package com.rem.rabbitmq.ee.Gexchanges.topic;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  9. /**
  10.  * @author Rem
  11.  * @date 2021-12-23
  12.  */
  14. public class Consumer12 {
  16.     public static void main(String[] args) throws Exception {
  18.         Channel channel = RabbitMqUtils.getChannel();
  20.         /**
  21.          * 创建一个临时队列  队列名称随机
  22.          * 当消费者断开和该队列的连接时，队列自动删除
  23.          */
  24.         String queue = channel.queueDeclare().getQueue();
  25.         /**
  26.          * 将临时队列绑定到交换机
  27.          */
  28.         channel.queueBind(queue, RabbitMqUtils.EXCHANGE_TOPIC, "*.blue");
  29.         channel.queueBind(queue, RabbitMqUtils.EXCHANGE_TOPIC, "#.red");
  31.         //消费回调
  32.         DeliverCallback deliverCallback = (consumerTag, message) -> System.out.println("接收到的消息：" + new String(message.getBody()));
  34.         //取消消费回调 如果在消费的时候队列被删除了
  35.         CancelCallback cancelCallback = consumerTag -> System.out.println("消费消息被中断");
  37.         System.out.println("消费者12 等待消费");
  38.         channel.basicConsume(queue, true, deliverCallback, cancelCallback);
  39.     }
  40. }
  1. package com.rem.rabbitmq.ee.Gexchanges.topic;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  9. /**
  10.  * @author Rem
  11.  * @date 2021-12-23
  12.  */
  14. public class Consumer13 {
  16.     public static void main(String[] args) throws Exception {
  18.         Channel channel = RabbitMqUtils.getChannel();
  20.         /**
  21.          * 创建一个临时队列  队列名称随机
  22.          * 当消费者断开和该队列的连接时，队列自动删除
  23.          */
  24.         String queue = channel.queueDeclare().getQueue();
  25.         /**
  26.          * 将临时队列绑定到交换机
  27.          */
  28.         channel.queueBind(queue, RabbitMqUtils.EXCHANGE_TOPIC, "green.*");
  29.         channel.queueBind(queue, RabbitMqUtils.EXCHANGE_TOPIC, "#.blue.#");
  31.         //消费回调
  32.         DeliverCallback deliverCallback = (consumerTag, message) -> System.out.println("接收到的消息：" + new String(message.getBody()));
  34.         //取消消费回调 如果在消费的时候队列被删除了
  35.         CancelCallback cancelCallback = consumerTag -> System.out.println("消费消息被中断");
  37.         System.out.println("消费者13 等待消费");
  38.         channel.basicConsume(queue, true, deliverCallback, cancelCallback);
  41.     }
  42. }
  1. package com.rem.rabbitmq.ee.Gexchanges.topic;
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rem.rabbitmq.ee.RabbitMqUtils;
  9. /**
  10.  * @author Rem
  11.  * @date 2021-12-23
  12.  */
  14. public class Consumer14 {
  16.     public static void main(String[] args) throws Exception {
  18.         Channel channel = RabbitMqUtils.getChannel();
  20.         /**
  21.          * 创建一个临时队列  队列名称随机
  22.          * 当消费者断开和该队列的连接时，队列自动删除
  23.          */
  24.         String queue = channel.queueDeclare().getQueue();
  25.         /**
  26.          * 将临时队列绑定到交换机
  27.          */
  28.         channel.queueBind(queue, RabbitMqUtils.EXCHANGE_TOPIC, "pink");
  30.         //消费回调
  31.         DeliverCallback deliverCallback = (consumerTag, message) -> System.out.println("接收到的消息：" + new String(message.getBody()));
  33.         //取消消费回调 如果在消费的时候队列被删除了
  34.         CancelCallback cancelCallback = consumerTag -> System.out.println("消费消息被中断");
  36.         System.out.println("消费者14 等待消费");
  37.         channel.basicConsume(queue, true, deliverCallback, cancelCallback);
  40.     }
  41. }

头部交换机

实际生产中较少使用