devbox
Cpp五条
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RabbitMQ-Java 简单使用_java处理rabbitmq

作者：Cpp五条 | 2024-04-08 08:55:35

赞

踩

java处理rabbitmq

RabbitMQ-Java 入门案例

参考非常详细的博主教程：https://www.cnblogs.com/dtdx/p/14362760.html
SpringBoot+Java 版教程：https://blog.csdn.net/lgl782519197/article/details/113775569

00、环境搭建

实现步骤：

1、IDEDA内构建一个maven工程（jdk1.8）
2：导入rabbitmq的maven依赖
3：启动rabbitmq-server服务
4：定义生产者、定义消费者
5：观察消息的在rabbitmq-server服务中的过程

1、构建一个maven工程

2、引入rabbitmq-maven依赖

  1. <!-- Java原生依赖 -->
  2. <dependency>
  3.     <groupId>com.rabbitmq</groupId>
  4.     <artifactId>amqp-client</artifactId>
  5.     <version>5.10.0</version>
  6. </dependency>
  1. <!-- Spring依赖 -->
  2. <dependency>
  3.     <groupId>org.springframework.amqp</groupId>
  4.     <artifactId>spring-amqp</artifactId>
  5.     <version>2.2.5.RELEASE</version>
  6. </dependency>
  7. <dependency>
  8.     <groupId>org.springframework.amqp</groupId>
  9.     <artifactId>spring-rabbit</artifactId>
  10.     <version>2.2.5.RELEASE</version>
  11. </dependency>
  1. <!-- SpringBoot依赖 -->
  2. <dependency>
  3.     <groupId>org.springframework.boot</groupId>
  4.     <artifactId>spring-boot-starter-amqp</artifactId>
  5. </dependency>

按需选择即可。
番外：rabbitmq和spring同属一个公司开放的产品，所以他们的支持也是非常完善，这也是为什么推荐使用rabbitmq的一个原因。

01、Simple 简单队列


简单理解
如果把使用 RabbitMQ 进行消息发送的过程比喻成邮寄邮件。那么简单队列的场景是，只有一个邮箱、一个邮局、一个投递员，。消息通过 RabbitMQ 进行一对一发送，发送过程最简单。

简单队列模型示意图


代码实战
一个生产者，一个消费者，一个队列（一个默认的交换机）

 

 

1、定义生产者

  1. package com.example.simple;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6. import java.io.IOException;
  7. import java.util.concurrent.TimeoutException;
  8.  
  9. /**
  10.  * @author:
  11.  * @description: Producer 简单队列生产者
  12.  */
  13. public class Producer {
  14.     public static void main(String[] args) throws Exception {
  15.         // 1: 创建连接工厂，设置连接属性
  16.         ConnectionFactory connectionFactory = new ConnectionFactory();
  17.         connectionFactory.setHost("127.0.0.1");
  18.         connectionFactory.setPort(5672);
  19.         connectionFactory.setVirtualHost("/");
  20.         connectionFactory.setUsername("guest");
  21.         connectionFactory.setPassword("guest");
  22.  
  23.         // 2: 从连接工厂中获取
  24.         Connection connection = connectionFactory.newConnection("生产者");
  25.         // 3: 从连接中打开通道channel
  26.         Channel channel = connection.createChannel();
  27.  
  28.         // 4: 通过创建交换机，声明队列，绑定关系，路由key,发送消息，和接收消息
  29.         /*
  30.          *  申明队列：如果队列不存在会自动创建。
  31.          *  注意：
  32.          *  1.Rabbitmq不允许创建两个相同的队列名称，否则会报错。
  33.          *  2.队列声明可以放在生产者、消费者或web页面上创建。但是在消费者启动监听之前队列一定要创建好。
  34.          *    如果要先启动消费者，建议把声明队列放在消费者端。否在消费者监听队列不存在会报异常。
  35.          *    如果要先启动生产者，建议在生产者端声明队列。虽然发送消息时队列不存不会报错，但第一次发送时队列不存在相当于白发送了。
  36.          *
  37.          *  @params1： queue 队列的名称
  38.          *  @params2： durable 队列是否持久化（即存盘），false = 非持久化 true = 持久化，非持久化会存盘吗？会存盘，但是会随从重启服务会丢失
  39.          *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
  40.          *  @params4： autoDelete 是否自动删除，当此队列的连接数为0时，此队列会销毁（无论队列中是否还有数据）
  41.          *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等
  42.          * */
  43.         channel.queueDeclare("simple-queue1", false, false, false, null);
  44.         // 5： 准备发送消息的内容
  45.         String message = "你好，消息队列！！！";
  46.         // 6: 发送消息给队列queue1
  47.         /*
  48.          * @params1: 交换机exchange
  49.          * @params2: 队列名称、路由key(routing)
  50.          * @params3: 属性配置
  51.          * @params4: 发送消息的内容
  52.          **/
  53.         // 面试题：可以存在没有交换机的队列吗？不可能，虽然没有指定交换机但是一定会存在一个默认的交换机。
  54.         channel.basicPublish("", "simple-queue1", null, message.getBytes());
  55.         System.out.println("消息发送成功!");
  56.  
  57.         // 最后关闭通关和连接
  58.         channel.close();
  59.         connection.close();
  60.     }
  61. }


2、定义消费者

  1. package com.example.simple;
  2.  
  3. import com.rabbitmq.client.*;
  4. import java.io.IOException;
  5.  
  6. /**
  7.  * @author:
  8.  * @description: Producer 简单队列消费者
  9.  */
  10. public class Consumer {
  11.     public static void main(String[] args) throws Exception {
  12.         // 1: 创建连接工厂，设置连接属性
  13.         ConnectionFactory connectionFactory = new ConnectionFactory();
  14.         connectionFactory.setHost("127.0.0.1");
  15.         connectionFactory.setPort(5672);
  16.         connectionFactory.setVirtualHost("/");
  17.         connectionFactory.setUsername("guest");
  18.         connectionFactory.setPassword("guest");
  19.  
  20.         /*
  21.          * 2: 从连接工厂中获取/创建连接（断点到此步可以发现web界面Connection下会出现此连接信息）
  22.          * 3: 从连接中获取通道channel（断点到此步可以发现web界面Channel下会出现此连接信息）
  23.          */
  24.         try (Connection connection = connectionFactory.newConnection("消费者");
  25.              Channel channel = connection.createChannel()){
  26.  
  27.             // 4: 通过创建交换机，声明队列，绑定关系，路由key,发送消息，和接收消息（声明队列可以在生产者或者消费者端）
  28.             //channel.queueDeclare("queue1", false, false, false, null);
  29.  
  30.             // 接收消息，监听对应的队列名即可
  31.             /*
  32.              *  @params1： queue 队列的名称
  33.              *  @params2： autoAck 指定是否自动ACK （true，接收到消息后，会立即告诉RabbitMQ）
  34.              *  @params3： deliverCallback 指定消费回调,开启监听队列queue1
  35.              *  @params4： cancelCallback 消费失败回调
  36.              * */
  37.             channel.basicConsume("simple-queue1", true, new DeliverCallback() {
  38.                 @Override
  39.                 public void handle(String consumerTag, Delivery delivery) throws IOException {
  40.                     System.out.println("收到消息是" + new String(delivery.getBody(), "UTF-8"));
  41.                 }
  42.             }, new CancelCallback() {
  43.                 @Override
  44.                 public void handle(String s)  {
  45.                     System.out.println("接受失败了...");
  46.                 }
  47.             });
  48.  
  49.             // 让程序停止，好接收消费
  50.             System.out.println("开始接受消息");
  51.             System.in.read();
  52.         }
  53.  
  54.     }
  55. }


3、先运行生产者，然后运行消费者。最后消费者控制台输出：

  1. 开始接受消息
  2. 收到消息是你好，消息队列！！！


02、Work 工作队列


简单理解
在前面我们学了Simple Queue 模型，Simple Queue 模型消息生产者和消费者是一 一对应的，消息由生产者发送到队列并由消费者消费。假设有一种消息，消息生产者一次性发送了10条消息，消费者消费一条消息执行耗时1秒，如果使用简单队列模式总共耗时10秒；如果使用工作队列模式有3个消费者共同消费这些消息理想情况下只需要3秒就可以处理完这些消息。

工作队列模型示意图

 

 

代码实战-轮询模式（Round-Robin）
1、定义生产者

  1. package com.example.work.round;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6.  
  7. public class Producer {
  8.     public static void main(String[] args) throws Exception {
  9.         // 1: 创建连接工厂，设置连接属性
  10.         ConnectionFactory connectionFactory = new ConnectionFactory();
  11.         connectionFactory.setHost("127.0.0.1");
  12.         connectionFactory.setPort(5672);
  13.         connectionFactory.setVirtualHost("/");
  14.         connectionFactory.setUsername("guest");
  15.         connectionFactory.setPassword("guest");
  16.  
  17.         // 2: 创建连接，获取通道
  18.         Connection connection = connectionFactory.newConnection("生产者");
  19.         Channel channel = connection.createChannel();
  20.  
  21.         // 3: 声明队列
  22.         channel.queueDeclare("work-queue1", false, false, false, null);
  23.         // 4: 循环发送消息
  24.         for (int i = 1; i <= 20; i++) {
  25.             channel.basicPublish("", "work-queue1", null, ("work-轮询模式:"+ i).getBytes());
  26.         }
  27.         System.out.println("消息发送成功!");
  28.  
  29.         // 最后关闭通关和连接
  30.         channel.close();
  31.         connection.close();
  32.     }
  33. }


2、定义消费者1

  1. package com.example.work.round;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. public class Work1 {
  6.     public static void main(String[] args) throws Exception {
  7.         // 1: 创建连接工厂，设置连接属性
  8.         ConnectionFactory connectionFactory = new ConnectionFactory();
  9.         connectionFactory.setHost("127.0.0.1");
  10.         connectionFactory.setPort(5672);
  11.         connectionFactory.setVirtualHost("/");
  12.         connectionFactory.setUsername("guest");
  13.         connectionFactory.setPassword("guest");
  14.  
  15.         // 2: 创建连接，获取通道
  16.         Connection connection = connectionFactory.newConnection("消费者1");
  17.         Channel channel = connection.createChannel();
  18.  
  19.         // 接收消息，监听对应的队列名即可
  20.         channel.basicConsume("work-queue1", true, (consumerTag, delivery) ->
  21.                 System.out.println("收到消息是" + new String(delivery.getBody(), "UTF-8")),
  22.                 consumerTag  -> {
  23.         });
  24.  
  25.     }
  26. }


3、定义消费者2

  1. package com.example.work.round;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. public class Work2 {
  6.     public static void main(String[] args) throws Exception {
  7.         // 1: 创建连接工厂，设置连接属性
  8.         ConnectionFactory connectionFactory = new ConnectionFactory();
  9.         connectionFactory.setHost("127.0.0.1");
  10.         connectionFactory.setPort(5672);
  11.         connectionFactory.setVirtualHost("/");
  12.         connectionFactory.setUsername("guest");
  13.         connectionFactory.setPassword("guest");
  14.  
  15.         // 2: 创建连接，获取通道
  16.         Connection connection = connectionFactory.newConnection("消费者2");
  17.         Channel channel = connection.createChannel();
  18.  
  19.         // 接收消息，监听对应的队列名即可
  20.         channel.basicConsume("work-queue1", true, (consumerTag, delivery) ->
  21.                 System.out.println("收到消息是" + new String(delivery.getBody(), "UTF-8")),
  22.                 consumerTag  -> {
  23.         });
  24.  
  25.     }
  26. }


查看控制台输出：

消费者1：

  1. 收到消息是work-轮询模式:1
  2. 收到消息是work-轮询模式:3
  3. 收到消息是work-轮询模式:5
  4. 收到消息是work-轮询模式:7
  5. 收到消息是work-轮询模式:9
  6. 收到消息是work-轮询模式:11
  7. 收到消息是work-轮询模式:13
  8. 收到消息是work-轮询模式:15
  9. 收到消息是work-轮询模式:17
  10. 收到消息是work-轮询模式:19


消费者2：

  1. 收到消息是work-轮询模式:2
  2. 收到消息是work-轮询模式:4
  3. 收到消息是work-轮询模式:6
  4. 收到消息是work-轮询模式:8
  5. 收到消息是work-轮询模式:10
  6. 收到消息是work-轮询模式:12
  7. 收到消息是work-轮询模式:14
  8. 收到消息是work-轮询模式:16
  9. 收到消息是work-轮询模式:18
  10. 收到消息是work-轮询模式:20


代码实战-公平分发（Fair-Dispatch）

也可以称为：Work模式的”能者多劳“模式

1、定义生产者

  1. package com.example.work.fairr;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. public class Producer {
  6.     public static void main(String[] args) throws Exception {
  7.         // 1: 创建连接工厂，设置连接属性
  8.         ConnectionFactory connectionFactory = new ConnectionFactory();
  9.         connectionFactory.setHost("127.0.0.1");
  10.         connectionFactory.setPort(5672);
  11.         connectionFactory.setVirtualHost("/");
  12.         connectionFactory.setUsername("guest");
  13.         connectionFactory.setPassword("guest");
  14.  
  15.         // 2: 创建连接，获取通道
  16.         Connection connection = connectionFactory.newConnection("生产者");
  17.         Channel channel = connection.createChannel();
  18.  
  19.         // 3: 声明队列
  20.         channel.queueDeclare("work-queue2", false, false, false, null);
  21.         for (int i = 1; i <= 20; i++) {
  22.             channel.basicPublish("", "work-queue2", null, ("work-公平分发:"+ i).getBytes());
  23.         }
  24.         System.out.println("消息发送成功!");
  25.  
  26.         // 最后关闭通关和连接
  27.         channel.close();
  28.         connection.close();
  29.     }
  30. }


2、定义消费者1

  1. package com.example.work.fairr;
  2.  
  3. import com.rabbitmq.client.*;
  4. import java.util.concurrent.TimeUnit;
  5.  
  6. public class Work1 {
  7.     public static void main(String[] args) throws Exception {
  8.         // 1: 创建连接工厂，设置连接属性
  9.         ConnectionFactory connectionFactory = new ConnectionFactory();
  10.         connectionFactory.setHost("127.0.0.1");
  11.         connectionFactory.setPort(5672);
  12.         connectionFactory.setVirtualHost("/");
  13.         connectionFactory.setUsername("guest");
  14.         connectionFactory.setPassword("guest");
  15.  
  16.         // 2: 创建连接，获取通道
  17.         Connection connection = connectionFactory.newConnection("消费者1");
  18.         Channel channel = connection.createChannel();
  19.  
  20.         // 同一时刻服务器只会发送一条消息给消费者
  21.         channel.basicQos(1);
  22.  
  23.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
  24.             try {
  25.                 // 加一个睡眠，模拟消费者1消费慢
  26.                 TimeUnit.MILLISECONDS.sleep(1000);
  27.                 System.out.println("Work1-收到消息是" + new String(delivery.getBody(), "UTF-8"));
  28.                 // 确认消息消费，参数1：确认队列中哪个具体消息，参数2：是否开启多个消息同时确认
  29.                 channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
  30.             } catch (InterruptedException e) {
  31.                 e.printStackTrace();
  32.             }
  33.         };
  34.  
  35.         // 接收消息，开启手动确认消费消息机制，设置autoAck为false，防止消息一下子都进入消费者
  36.         channel.basicConsume("work-queue2", false, deliverCallback,consumerTag  -> {
  37.         });
  38.  
  39.     }
  40. }


3、定义消费者2

  1. package com.example.work.fairr;
  2.  
  3. import com.rabbitmq.client.*;
  4. import java.util.concurrent.TimeUnit;
  5.  
  6. public class Work2 {
  7.     public static void main(String[] args) throws Exception {
  8.         // 1: 创建连接工厂，设置连接属性
  9.         ConnectionFactory connectionFactory = new ConnectionFactory();
  10.         connectionFactory.setHost("127.0.0.1");
  11.         connectionFactory.setPort(5672);
  12.         connectionFactory.setVirtualHost("/");
  13.         connectionFactory.setUsername("guest");
  14.         connectionFactory.setPassword("guest");
  15.  
  16.         // 2: 创建连接，获取通道
  17.         Connection connection = connectionFactory.newConnection("消费者2");
  18.         Channel channel = connection.createChannel();
  19.  
  20.         // 同一时刻服务器只会发送一条消息给消费者
  21.         channel.basicQos(1);
  22.  
  23.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
  24.             try {
  25.                 // 加一个睡眠，模拟消费者1消费慢
  26.                 TimeUnit.MILLISECONDS.sleep(100);
  27.                 System.out.println("Work2-收到消息是" + new String(delivery.getBody(), "UTF-8"));
  28.                 // 确认消息消费，参数1：确认队列中哪个具体消息，参数2：是否开启多个消息同时确认
  29.                 channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
  30.             } catch (InterruptedException e) {
  31.                 e.printStackTrace();
  32.             }
  33.         };
  34.  
  35.         // 接收消息，开启确认机制，设置autoAck为false
  36.         channel.basicConsume("work-queue2", false, deliverCallback, consumerTag  -> {
  37.         });
  38.  
  39.     }
  40. }


与一般情况不同的是如下三个步骤（在消费端配置）：

  1. // 1.同一时刻服务器只会发一条消息给消费者
  2. channel.basicQos(1);
  3.  
  4. // 2.开启这行 表示使用手动确认模式
  5. channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
  6.  
  7. // 3.同时改为手动确认：监听队列，false表示手动返回完成状态，true表示自动
  8. channel.basicConsume(QUEUE_NAME, false, consumer);


查看控制台输出结果：

消费者1：

  1. Work1-收到消息是work-公平分发:2
  2. Work1-收到消息是work-公平分发:12

消费者2：

  1. Work2-收到消息是work-公平分发:1
  2. Work2-收到消息是work-公平分发:3
  3. Work2-收到消息是work-公平分发:4
  4. Work2-收到消息是work-公平分发:5
  5. Work2-收到消息是work-公平分发:6
  6. Work2-收到消息是work-公平分发:7
  7. Work2-收到消息是work-公平分发:8
  8. Work2-收到消息是work-公平分发:9
  9. Work2-收到消息是work-公平分发:10
  10. Work2-收到消息是work-公平分发:11
  11. Work2-收到消息是work-公平分发:13
  12. Work2-收到消息是work-公平分发:14
  13. Work2-收到消息是work-公平分发:15
  14. Work2-收到消息是work-公平分发:16
  15. Work2-收到消息是work-公平分发:17
  16. Work2-收到消息是work-公平分发:18
  17. Work2-收到消息是work-公平分发:19
  18. Work2-收到消息是work-公平分发:20


03、Fanout 发布订阅


Pub/Sub模型简单理解
在前面我们学习了简单队列模型和工作队列模型，其实不管是简单队列模型还是工作队列模式只用到了 RabbitMQ 中的 Queue 队列，消息直接发送到队列中，消费者直接从队列中获取消息。

现在我们学习另一种消息的处理方式，消息生产者将消息发送到 Exchange 交换机中，再由交换机将消息投递到 Queue队列中，交换机和队列的关系是多对多的关系，表示一个交换机可以通过一定的规矩将消息投递到多个队列中，同样一个队列也可以接收多个交换机投递的消息

交换机投递消息到队列由哪几种方式？

Fanout： 不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。Fanout交换机转发消息是最快的
Direct：处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。这是一个完整的匹配。如果一个队列绑定到该交换机上要求路由键 test，则只有被标记为test的消息才被转发，不会转发test.aaa，也不会转发dog.123，只会转发test。
Topic： 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号#匹配一个或多个词，符号* 匹配不多不少一个词。因此audit.#能够匹配到audit.irs.corporate，但是audit.* 只会匹配到audit.irs
Headers：Headers类型的exchange使用的比较少，它也是忽略routingKey的一种路由方式。是使用Headers来匹配的。Headers是一个键值对，可以定义成Hashtable。发送者在发送的时候定义一些键值对，接收者也可以再绑定时候传入一些键值对，两者匹配的话，则对应的队列就可以收到消息。匹配有两种方式all和any。这两种方式是在接收端必须要用键值"x-mactch"来定义。all代表定义的多个键值对都要满足，而any则代码只要满足一个就可以了。fanout，direct，topic exchange的routingKey都需要要字符串形式的，而headers exchange则没有这个要求，因为键值对的值可以是任何类型

发布订阅模型示例图

 

 代码实战

一个生产者，一个交换机，三个队列，三个消费者
特点：发布与订阅模式，是一种广播机制，它是没有路由key的模式。

1、定义生产者

  1. package com.example.fanout;
  2.  
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  6. import com.rabbitmq.client.ConnectionFactory;
  7.  
  8. public class Producer {
  9.     public static void main(String[] args) throws Exception {
  10.         // 1: 创建连接工厂，设置连接属性
  11.         ConnectionFactory connectionFactory = new ConnectionFactory();
  12.         connectionFactory.setHost("127.0.0.1");
  13.         connectionFactory.setPort(5672);
  14.         connectionFactory.setVirtualHost("/");
  15.         connectionFactory.setUsername("guest");
  16.         connectionFactory.setPassword("guest");
  17.  
  18.         // 2: 创建连接，获取通道
  19.         Connection connection = connectionFactory.newConnection("生产者");
  20.         Channel channel = connection.createChannel();
  21.  
  22.         // 3: 定义队列名称-随机生成队列名、交换机名、路由key（fanout模式时路由key为空字符串）
  23.         String exchangeName = "fanout-exchange";
  24.  
  25.         // 4-1: 声明交换机（也可通过web页面创建）
  26.         channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT);
  27.         // 4-2: 声明队列
  28.         channel.queueDeclare("fanout-queue1", false, false, false, null);
  29.         channel.queueDeclare("fanout-queue2", false, false, false, null);
  30.         channel.queueDeclare("fanout-queue3", false, false, false, null);
  31.         // 4-3: 绑定交换机和队列
  32.         channel.queueBind("fanout-queue1", exchangeName, "");
  33.         channel.queueBind("fanout-queue2", exchangeName, "");
  34.         channel.queueBind("fanout-queue3", exchangeName, "");
  35.  
  36.         // 5: @params1: 交换机名  @params2 队列/路由key @params 属性配置  @params4 消息内容
  37.         channel.basicPublish(exchangeName, "", null, "你好，消息队列！".getBytes("UTF-8"));
  38.         System.out.println("消息发送成功!");
  39.  
  40.         // 最后关闭通关和连接
  41.         channel.close();
  42.         connection.close();
  43.     }
  44. }


2、定义消费者

  1. package com.example.fanout;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6. import com.rabbitmq.client.DeliverCallback;
  7.  
  8. public class Consumer {
  9.  
  10.     private static Runnable runnable = () -> {
  11.         // 1: 创建连接工厂，设置连接属性
  12.         ConnectionFactory connectionFactory = new ConnectionFactory();
  13.         connectionFactory.setHost("127.0.0.1");
  14.         connectionFactory.setPort(5672);
  15.         connectionFactory.setVirtualHost("/");
  16.         connectionFactory.setUsername("guest");
  17.         connectionFactory.setPassword("guest");
  18.  
  19.         Connection connection = null;
  20.         Channel channel = null;
  21.         try {
  22.             // 2: 创建连接，获取通道（消费者一般不增加自动关闭）
  23.             connection = connectionFactory.newConnection("消费者");
  24.             channel = connection.createChannel();
  25.             // 获取队列的名称
  26.             final String queueName = Thread.currentThread().getName();
  27.  
  28.             // 6： 定义接受消息的回调
  29.             DeliverCallback deliverCallback = (consumerTag, delivery) ->
  30.                     System.out.println(queueName + "：收到消息是：" + new String(delivery.getBody(), "UTF-8"));
  31.  
  32.             channel.basicConsume(queueName, true, deliverCallback, consumerTag  -> {
  33.             });
  34.         } catch (Exception e) {
  35.             e.printStackTrace();
  36.             System.out.println("发送消息出现异常...");
  37.         } finally {
  38.             // 此处应该是需要关闭通道的，不过为了测试咱们不关闭了
  39.         }
  40.     };
  41.  
  42.     public static void main(String[] args) {
  43.         // 启动三个线程去执行
  44.         new Thread(runnable, "fanout-queue1").start();
  45.         new Thread(runnable, "fanout-queue2").start();
  46.         new Thread(runnable, "fanout-queue3").start();
  47.     }
  48. }


消费者控制台输出：

  1. fanout-queue3：收到消息是：你好，消息队列！
  2. fanout-queue1：收到消息是：你好，消息队列！
  3. fanout-queue2：收到消息是：你好，消息队列！


04、Direct 路由模式


简单理解
Direct Exchange 消息模型 是交换机将消息投递到队列中的另一种方式，需要将交换机类型设置为 direct。那么什么是 direct 类型交换机？

其中Direct：处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。这是一个完整的匹配。如果一个队列绑定到该交换机上要求路由键 test，则只有被标记为test的消息才被转发，不会转发test.aaa，也不会转发dog.123，只会转发test。

路由消息模型示意图

 

 Direct 消息流转过程

  • 生产者向Exchange发送消息。
  • 队列使用路由密钥绑定到Exchange。
  • 通常，使用相同/不同的路由密钥有多个队列绑定到Exchange。
  • 发送到Exchange的消息包含路由密钥。根据路由密钥，消息将转发到一个或多个队列。
  • 订阅队列的使用者接收消息并进行处理。


代码实战

一个生产者，一个交换机，两个队列，两个消费者
特点：Direct模式是fanout模式上的一种叠加，增加了路由RoutingKey的模式。

1、定义生产者

  1. package com.example.direct;
  2.  
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  6. import com.rabbitmq.client.ConnectionFactory;
  7.  
  8. public class Producer {
  9.     public static void main(String[] args) throws Exception {
  10.         // 1: 创建连接工厂，设置连接属性
  11.         ConnectionFactory connectionFactory = new ConnectionFactory();
  12.         connectionFactory.setHost("127.0.0.1");
  13.         connectionFactory.setPort(5672);
  14.         connectionFactory.setVirtualHost("/");
  15.         connectionFactory.setUsername("guest");
  16.         connectionFactory.setPassword("guest");
  17.  
  18.         // 2: 创建连接，获取通道
  19.         Connection connection = connectionFactory.newConnection("生产者");
  20.         Channel channel = connection.createChannel();
  21.  
  22.         // 3: 定义队列名称-随机生成队列名、交换机名、路由key（fanout模式时路由key为空字符串）
  23.         String exchangeName = "direct-exchange";
  24.  
  25.         // 4-1: 声明交换机
  26.         channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
  27.         // 4-2: 声明队列
  28.         channel.queueDeclare("direct-info", false, false, false, null);
  29.         channel.queueDeclare("direct-error", false, false, false, null);
  30.         // 4-3: 绑定交换机和队列
  31.         channel.queueBind("direct-info", exchangeName, "info");
  32.         channel.queueBind("direct-error", exchangeName, "error");
  33.  
  34.         // 5: @params1: 交换机名  @params2 队列/路由key @params 属性配置  @params4 消息内容
  35.         channel.basicPublish(exchangeName, "info", null, "你好，消息队列！".getBytes("UTF-8"));
  36.         channel.basicPublish(exchangeName, "error", null, "你好，消息队列！".getBytes("UTF-8"));
  37.  
  38.         System.out.println("消息发送成功!");
  39.  
  40.         // 最后关闭通关和连接
  41.         channel.close();
  42.         connection.close();
  43.     }
  44. }


2、定义消费者

  1. package com.example.direct;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6. import com.rabbitmq.client.DeliverCallback;
  7.  
  8. public class Consumer {
  9.  
  10.     private static Runnable runnable = () -> {
  11.         // 1: 创建连接工厂，设置连接属性
  12.         ConnectionFactory connectionFactory = new ConnectionFactory();
  13.         connectionFactory.setHost("127.0.0.1");
  14.         connectionFactory.setPort(5672);
  15.         connectionFactory.setVirtualHost("/");
  16.         connectionFactory.setUsername("guest");
  17.         connectionFactory.setPassword("guest");
  18.  
  19.         Connection connection = null;
  20.         Channel channel = null;
  21.         try {
  22.             // 2: 创建连接，获取通道（消费者一般不增加自动关闭）
  23.             connection = connectionFactory.newConnection("消费者");
  24.             channel = connection.createChannel();
  25.  
  26.             // 获取队列的名称
  27.             final String queueName = Thread.currentThread().getName();
  28.  
  29.             // 3：定义接受消息的回调
  30.             channel.basicConsume(queueName, true, (consumerTag, delivery) ->
  31.                     System.out.println(queueName + "：收到消息是：" + new String(delivery.getBody()),
  32.                     consumerTag  -> {
  33.             });
  34.         } catch (Exception e) {
  35.             e.printStackTrace();
  36.             System.out.println("发送消息出现异常...");
  37.         } finally {
  38.             // 此处应该是需要关闭通道的，不过为了测试咱们不关闭了
  39.         }
  40.     };
  41.  
  42.     public static void main(String[] args) {
  43.         // 启动三个线程去执行
  44.         new Thread(runnable, "direct-info").start();
  45.         new Thread(runnable, "direct-error").start();
  46.     }
  47. }

消费者控制台输出：

  1. direct-error：收到消息是：你好，消息队列！
  2. direct-info：收到消息是：你好，消息队列！

05、Topic 主题模式

简单理解
Topic Exchange 消息模型需要指定路由器类型设置为Topic，那么什么是 Topic 类型？

Topic： 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号#匹配一个或多个词，符号* 匹配不多不少一个词。因此audit.#能够匹配到audit.irs.corporate，但是audit.* 只会匹配到audit.irs

主题消息模型示意图

 

 Topic 中的路由键设置规则

  • Topic Exchange中的路由关键字必须包含零个或多个由 . 点分隔的单词，例如health.education。
  • Topic Exchange中的路由键通常称为路由模式。
  • 路由键允许只包含 星号（*）和 井号 （#）的正则表达式组成
  • 星号（*****）表示正好允许一个字。
  • 同样，井号（＃）表示允许的单词数为零或更多。
  • 点号（.）表示–单词定界符。多个关键术语用点定界符分隔。
  • 如果路由模式为**health.\***，则意味着以第一个单词为首的路由键运行状况发送的任何消息都将到达队列。例如，health.education将到达此队列，但sports.health将不起作用。

Topic 中的消息流转过程

  • 一个Queue队列通过路由键（P）绑定到 Exchange。
  • Producer 将带有P路由键（K）的消息发送到 Topic Exchange。
  • 如果P与K匹配，则消息被传递到队列。路由密钥匹配的确定如下所述。
  • 订阅队列的使用者将收到消息。

代码实战

一个生产者，一个交换机，两个队列，两个消费者
特点：Topic模式是direct模式上的一种叠加，增加了模糊路由RoutingKey的模式。

生产者创建Topic的exchange并且绑定到队列中，绑定可以通过*和#关键字，对指定RoutingKey内容，编写时注意格式 xxx.xxx.xxx 去别写。* -》代表xxx，# -》代表多个xxx.xxx，在发送消息时，指定具体的RoutingKey到底时什么
简单的说：* 表示一个，# 表示 0个 或者 多个

1、定义生产者

  1. package com.example.topic;
  2.  
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  6. import com.rabbitmq.client.ConnectionFactory;
  7.  
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  10.  
  11. public class Producer {
  12.     public static void main(String[] args) throws IOException, TimeoutException {
  13.         // 1: 创建连接工厂，设置连接属性
  14.         ConnectionFactory connectionFactory = new ConnectionFactory();
  15.         connectionFactory.setHost("127.0.0.1");
  16.         connectionFactory.setPort(5672);
  17.         connectionFactory.setVirtualHost("/");
  18.         connectionFactory.setUsername("guest");
  19.         connectionFactory.setPassword("guest");
  20.  
  21.         // 2: 创建连接，获取通道
  22.         Connection connection = connectionFactory.newConnection("生产者");
  23.         Channel channel = connection.createChannel();
  24.  
  25.         // 3: 定义队列名称-随机生成队列名、交换机名、路由key（fanout模式时路由key为空字符串）
  26.         String exchangeName = "topic-exchange";
  27.  
  28.         // 4-1: 声明交换机（也可通过web页面创建）
  29.         channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);
  30.  
  31.         // 4-2: 声明队列
  32.         channel.queueDeclare("topic-queue1", false, false, false, null);
  33.         channel.queueDeclare("topic-queue2", false, false, false, null);
  34.         channel.queueDeclare("topic-queue3", false, false, false, null);
  35.  
  36.         // 4-3: 绑定交换机和队列
  37.         channel.queueBind("topic-queue1", exchangeName, "#.order");
  38.         channel.queueBind("topic-queue2", exchangeName, "*.user");
  39.  
  40.         // 5: @params1: 交换机名  @params2 队列/路由key @params 属性配置  @params4 消息内容
  41.         // topic-queue1、topic-queue2、topic-queue3
  42.         channel.basicPublish(exchangeName, "com.course.order", null, "routingKey：#.order".getBytes("UTF-8"));
  43.         // topic-queue3
  44.         channel.basicPublish(exchangeName, "com.order.user", null, "routingKey：#.user".getBytes("UTF-8"));
  45.  
  46.         System.out.println("消息发送成功!");
  47.  
  48.         // 最后关闭通关和连接
  49.         channel.close();
  50.         connection.close();
  51.     }
  52. }

2、定义消费者

  1. package com.example.topic;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6. import com.rabbitmq.client.DeliverCallback;
  7.  
  8. public class Consumer {
  9.  
  10.     private static Runnable runnable = () -> {
  11.         // 1: 创建连接工厂，设置连接属性
  12.         ConnectionFactory connectionFactory = new ConnectionFactory();
  13.         connectionFactory.setHost("127.0.0.1");
  14.         connectionFactory.setPort(5672);
  15.         connectionFactory.setVirtualHost("/");
  16.         connectionFactory.setUsername("guest");
  17.         connectionFactory.setPassword("guest");
  18.  
  19.         Connection connection = null;
  20.         Channel channel = null;
  21.         try {
  22.             // 2: 创建连接，获取通道（消费者一般不增加自动关闭）
  23.             connection = connectionFactory.newConnection("消费者");
  24.             channel = connection.createChannel();
  25.             // 获取队列的名称
  26.             final String queueName = Thread.currentThread().getName();
  27.  
  28.             // 6： 定义接受消息的回调
  29.             channel.basicConsume(queueName, true,
  30.                     (consumerTag, delivery) -> System.out.println(queueName + "：收到消息是：" + new String(delivery.getBody(), "UTF-8")),
  31.                     consumerTag  -> {
  32.             });
  33.         } catch (Exception e) {
  34.             e.printStackTrace();
  35.             System.err.println("发送消息出现异常...");
  36.         } finally {
  37.             // 此处应该是需要关闭通道的，不过为了测试咱们不关闭了
  38.         }
  39.     };
  40.  
  41.     public static void main(String[] args) {
  42.         // 启动三个线程去执行
  43.         new Thread(runnable, "topic-queue1").start();
  44.         new Thread(runnable, "topic-queue2").start();
  45.         new Thread(runnable, "topic-queue3").start();
  46.     }
  47. }

消费者控制台输出：

topic-queue1：收到消息是：routingKey：#.order


06、Headers 模式


Headers Exchange 模型需要指定路由器类型设置为Headers，那么什么是 HeadersHeaders 类型？

Headers：Headers类型的exchange使用的比较少，它也是忽略routingKey的一种路由方式。是使用Headers来匹配的。Headers是一个键值对，可以定义成Hashtable。发送者在发送的时候定义一些键值对，接收者也可以再绑定时候传入一些键值对，两者匹配的话，则对应的队列就可以收到消息。匹配有两种方式all和any。这两种方式是在接收端必须要用键值"x-mactch"来定义。all代表定义的多个键值对都要满足，而any则代码只要满足一个就可以了。fanout，direct，topic exchange的routingKey都需要要字符串形式的，而headers exchange则没有这个要求，因为键值对的值可以是任何类型

Headers 消息模型示意图

 Headers 中消息流转过程

  • 一个或多个队列使用标头属性（H）绑定（链接）到标头交换。
  • 生产者将带有标头属性（MH）的消息发送到此Exchange。
  • 如果MH与H匹配，则消息将转发到队列。
  • 监听队列的使用者接收消息并对其进行处理。

代码实现
1、生产者：

  1. package com.example.header;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.util.HashMap;
  6. import java.util.Map;
  7.  
  8. public class Producer {
  9.     public static void main(String[] args) throws Exception {
  10.         // 1: 创建连接工厂，设置连接属性
  11.         ConnectionFactory connectionFactory = new ConnectionFactory();
  12.         connectionFactory.setHost("127.0.0.1");
  13.         connectionFactory.setPort(5672);
  14.         connectionFactory.setVirtualHost("/");
  15.         connectionFactory.setUsername("guest");
  16.         connectionFactory.setPassword("guest");
  17.  
  18.         // 2: 创建连接，获取通道
  19.         Connection connection = connectionFactory.newConnection("生产者");
  20.         Channel channel = connection.createChannel();
  21.  
  22.         // 3: 定义队列名称-随机生成队列名、交换机名、路由key
  23.  
  24.         /**
  25.          * 定义交换机、队列 和 绑定队列和交换机并绑定配置参数
  26.          */
  27.         // 4-1: 声明交换机
  28.         channel.exchangeDeclare("header-exchange", BuiltinExchangeType.HEADERS);
  29.         // 4-2: 声明队列
  30.         channel.queueDeclare("header-queue-noe", false, false, false, null);
  31.         channel.queueDeclare("header-queue-two", false, false, false, null);
  32.         // 4-3: 绑定交换机和队列并加上配置header匹配规则
  33.         Map<String, Object> oneArgs = new HashMap<>();
  34.         // 匹配其中一个key/value 就能匹配成功
  35.         oneArgs.put("x-match", "any");
  36.         oneArgs.put("h1", "Header1");
  37.         oneArgs.put("h2", "Header2");
  38.         channel.queueBind("header-queue-noe", "header-exchange", "", oneArgs);
  39.         Map<String, Object> twoArgs = new HashMap<>();
  40.         // 必须匹配所有key/value 才能匹配成功
  41.         twoArgs.put("x-match", "all");
  42.         twoArgs.put("h1", "Header1");
  43.         twoArgs.put("h2", "Header2");
  44.         channel.queueBind("header-queue-two", "header-exchange", "", twoArgs);
  45.  
  46.  
  47.         /**
  48.          * 定义需要发送的消息和属性
  49.          */
  50.         Map<String, Object> headerMap = new HashMap<>();
  51.         headerMap.put("h1", "Header1");
  52.         headerMap.put("h3", "Header3");
  53.  
  54.         AMQP.BasicProperties basicPropertiesOne = new AMQP.BasicProperties()
  55.                 .builder().headers(headerMap).build();
  56.         channel.basicPublish("header-exchange", "", basicPropertiesOne, "Header Exchange example 1".getBytes("UTF-8"));
  57.         System.out.println("h1,h3：消息发送成功!");
  58.  
  59.         headerMap.put("h2", "Header2");
  60.         AMQP.BasicProperties basicPropertiesTwo = new AMQP.BasicProperties()
  61.                 .builder().headers(headerMap).build();
  62.         channel.basicPublish("header-exchange", "", basicPropertiesTwo, "Header Exchange example 2".getBytes("UTF-8"));
  63.         System.out.println("h1,h2,h3：消息发送成功!");
  64.  
  65.         // 最后关闭通关和连接
  66.         channel.close();
  67.         connection.close();
  68.     }
  69. }

2、消费者：

  1. package com.example.header;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6. import com.rabbitmq.client.DeliverCallback;
  7.  
  8. public class Consumer {
  9.  
  10.     private static Runnable runnable = () -> {
  11.         // 1: 创建连接工厂，设置连接属性
  12.         ConnectionFactory connectionFactory = new ConnectionFactory();
  13.         Connection connection = null;
  14.         Channel channel = null;
  15.         try {
  16.             // 2: 创建连接，获取通道（消费者一般不增加自动关闭）
  17.             connection = connectionFactory.newConnection("消费者");
  18.             channel = connection.createChannel();
  19.             // 获取队列的名称
  20.             final String queueName = Thread.currentThread().getName();
  21.  
  22.             // 6： 定义接受消息的回调
  23.             channel.basicConsume(queueName, true, ((consumerTag, message) -> {
  24.                 System.out.println(queueName + "：收到消息是: " + new String(message.getBody()));
  25.             }), consumerTag -> {
  26.                 System.out.println(consumerTag);
  27.             });
  28.         } catch (Exception e) {
  29.             e.printStackTrace();
  30.             System.out.println("发送消息出现异常...");
  31.         } finally {
  32.             // 此处应该是需要关闭通道的，不过为了测试咱们不关闭了
  33.         }
  34.     };
  35.  
  36.     public static void main(String[] args) {
  37.         // 启动三个线程去执行
  38.         new Thread(runnable, "header-queue-noe").start();
  39.         new Thread(runnable, "header-queue-two").start();
  40.     }
  41. }

消费者控制台输出：

  1. header-queue-noe：收到消息是: Header Exchange example 1
  2. header-queue-two：收到消息是: Header Exchange example 2
  3. header-queue-noe：收到消息是: Header Exchange example 2
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Cpp五条/article/detail/384262
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号