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

Rabbitmq常用的五种工作模式_rabbitmq在工作中常用的模式是哪种

作者：你好赵伟 | 2024-05-19 21:26:10

赞

踩

rabbitmq在工作中常用的模式是哪种

目录

前言

1、simple（简单模式）

（1）简单模式的模型

 （2）搭建测试项目

1、生产者

2、消费者

（3）测试项目

2、 Work queues（工作模式）

（1）工作模式的模型

（2）搭建测试项目

1、生产者

2、消费者 

（3）测试项目

（4） 总结

3、 Publish/Subscribe（发布订阅模式）

（1）发布订阅模型

（2）搭建测试项目

1、生产者

2、消费者

（3）测试项目

 （4）总结

4、Routing（路由模式）

（1）路由模型 

（2）搭建测试项目

1、生产者

2、消费者

（3）测试项目

（4）总结

5、Topics（主题模式） 

（1）主题模式模型

（2）搭建测试项目

1、生产者

2、消费者

（3）测试项目

（4）总结

前言

在介绍Rabbitmq的五种常用的工作模式之前先导入需要的依赖

<dependencies>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.14.2</version>
</dependency>
</dependencies>

然后创建两个模块，一个是生产者模块，一个是消费者模块，方便我们后面演示效果

 

1、simple（简单模式）

（1）简单模式的模型

首先介绍一下，简单模式，在上图的模型中，有以下概念：
 P：生产者，也就是要发送消息的程序
C：消费者，消息的接收者，会一直等待消息到来
queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息

 （2）搭建测试项目

 这里需要注意的是，生产者和消费者不在一个模块中

1、生产者
  1. public class HelloProducer {
  2.     public static void main(String[] args) throws IOException, TimeoutException {
  3.         //创建连接，并设置连接信息
  4.         ConnectionFactory factory = new ConnectionFactory();
  5.         //设置rabbitmq服务的地址
  6.         factory.setHost("192.168.47.135");
  7.         //设置rabbitmq服务的端口号
  8.         factory.setPort(5672);
  9.         //设置连接的账号
  10.         factory.setUsername("ykk");
  11.         //设置连接的密码
  12.         factory.setPassword("ykk");
  13.         //设置虚拟主机
  14.         factory.setVirtualHost("/ykk");
  15.         //获取连接对象
  16.         Connection connection = factory.newConnection();
  17.         //获取channel对象
  18.         Channel channel = connection.createChannel();
  19.         //创建队列
  20.         /**
  21.          * String queue, 队列的名称. 如果该名称不存在 则创建如果存在则不创建
  22.          * boolean durable, 该对象是否持久化 当rabbitmq重启后 队列就会消失
  23.          * boolean exclusive, 该队列是否被一个消费者独占
  24.          * boolean autoDelete,当没有消费者时，该队列是否被自动删除
  25.          * Map<String, Object> arguments: 额外参数的设置
  26.          */
  27.         channel.queueDeclare("hello_queue", true, false, false, null);
  28.         //发送消息
  29.         /**
  30.          * String exchange, 交换机的名称 简单模式没有交换机使用""表示采用默认交换机
  31.          * String routingKey, 路由标识 如果是简单模式起名为队列的名称
  32.          * BasicProperties props, 消息的属性设置。设置为null
  33.          * byte[] body: 消息的内容
  34.          */
  35.         //msg是测试发送的信息
  36.         String msg="hello rabbitmq";
  37.         channel.basicPublish("","hello_queue",null,msg.getBytes());
  38.         //关闭资源
  39.         channel.close();
  40.         connection. Close();
  41.     }
  42. }
2、消费者
  1. public class HelloConsumer {
  2.     public static void main(String[] args) throws IOException, TimeoutException {
  3.         //创建连接，并设置连接信息
  4.         ConnectionFactory factory = new ConnectionFactory();
  5.         //设置rabbitmq服务的地址
  6.         factory.setHost("192.168.47.135");
  7.         //设置rabbitmq服务的端口号
  8.         factory.setPort(5672);
  9.         //设置连接的账号
  10.         factory.setUsername("ykk");
  11.         //设置连接的密码
  12.         factory.setPassword("ykk");
  13.         //设置虚拟主机
  14.         factory.setVirtualHost("/ykk");
  15.         //获取连接对象
  16.         Connection connection = factory.newConnection();
  17.         //获取channel对象
  18.         Channel channel = connection.createChannel();
  19.         //接受队列中的消息
  20.         Consumer consumer=new DefaultConsumer(channel){
  21.             /**
  22.              * @param consumerTag: 消费者的标签
  23.              * @param envelope : 设置 拿到你的交换机 路由key等信息
  24.              * @param properties: 消息的属性对象
  25.              * @param body: 消息的内容
  26.              * @throws IOException
  27.              */
  28.             @Override
  29.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  30.                 //这里我们在控制台打印一下，看下消费者是否接收到了生产者传过来的信息
  31.                 System.out.println("接受的内容:"+new String(body));
  32.                 System.out.println("消费者的标志:"+consumerTag);
  33.                 System.out.println("交换机名称:"+envelope.getExchange());
  34.                 System.out.println("路由key标志:"+envelope.getRoutingKey());
  35.                 System.out.println("消息属性:"+properties);
  36.                 super.handleDelivery(consumerTag, envelope, properties, body);
  37.             }
  38.         };
  39.         /**
  40.          * String queue, 队列名
  41.          * boolean autoAck,是否自动确认。 当rabbitmq把消息发送给消费后，消费端自动确认消息。
  42.          * Consumer callback：回调。 当rabbitmq队列中存在消息则触发该回调
  43.          */
  44.         channel.basicConsume("hello_queue",true,consumer);
  45.  
  46.         //这里需要注意一点，不能关闭connection和channel
  47.     }
  48. }

（3）测试项目

 首先运行HelloProducer项目，然后在RabbitMQ前端控制台可以看到队列中此时存在一个Messages，因为此时没有消费者（消费者没有启动），所以此时消息存在于队列中

 启动HelloConsumer ，控制台打印结果

 

 此时观察RabbitMQ前端控制器队列中的情况
 

 

2、 Work queues（工作模式）
 

（1）工作模式的模型
 

 

 
 
 Work Queues：与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。
 应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。
 

 

（2）搭建测试项目
 

1、生产者
 

  1. public class WorkProducer {
  2.     public static void main(String[] args) throws IOException, TimeoutException {
  3.         //创建连接，并设置连接信息
  4.         ConnectionFactory factory = new ConnectionFactory();
  5.         //设置rabbitmq服务的地址
  6.         factory.setHost("192.168.47.135");
  7.         //设置rabbitmq服务的端口号
  8.         factory.setPort(5672);
  9.         //设置连接的账号
  10.         factory.setUsername("ykk");
  11.         //设置连接的密码
  12.         factory.setPassword("ykk");
  13.         //设置虚拟主机
  14.         factory.setVirtualHost("/ykk");
  15.         //获取连接对象
  16.         Connection connection = factory.newConnection();
  17.         //获取channel对象
  18.         Channel channel = connection.createChannel();
  19.         //创建队列
  20.         /**
  21.          * String queue, 队列的名称. 如果该名称不存在 则创建如果存在则不创建
  22.          * boolean durable, 该对象是否持久化 当rabbitmq重启后 队列就会消失
  23.          * boolean exclusive, 该队列是否被一个消费者独占
  24.          * boolean autoDelete,当没有消费者时，该队列是否被自动删除
  25.          * Map<String, Object> arguments: 额外参数的设置
  26.          */
  27.         channel.queueDeclare("work_queue", true, false, false, null);
  28.         //发送消息
  29.         /**
  30.          * String exchange, 交换机的名称 简单模式没有交换机使用""表示采用默认交换机
  31.          * String routingKey, 路由标识 如果是简单模式起名为队列的名称
  32.          * BasicProperties props, 消息的属性设置。设置为null
  33.          * byte[] body: 消息的内容
  34.          */
  35.         //msg是测试发送的信息(这里和simple模式不一样，我们一次性生产十条消息)
  36.         for (int i = 0; i < 10; i++) {
  37.             String msg="work rabbitmq-----------------"+i;
  38.             channel.basicPublish("","work_queue",null,msg.getBytes());
  39.         }
  40.         //关闭资源
  41.         channel.close();
  42.         connection. Close();
  43.     }
  44. }
 

2、消费者 
 

  1. //消费者1
  2. public class WorkConsumer01 {
  3.     public static void main(String[] args) throws IOException, TimeoutException {
  4. //创建连接，并设置连接信息
  5.         ConnectionFactory factory = new ConnectionFactory();
  6.         //设置rabbitmq服务的地址
  7.         factory.setHost("192.168.47.135");
  8.         //设置rabbitmq服务的端口号
  9.         factory.setPort(5672);
  10.         //设置连接的账号
  11.         factory.setUsername("ykk");
  12.         //设置连接的密码
  13.         factory.setPassword("ykk");
  14.         //设置虚拟主机
  15.         factory.setVirtualHost("/ykk");
  16.         //获取连接对象
  17.         Connection connection = factory.newConnection();
  18.         //获取channel对象
  19.         Channel channel = connection.createChannel();
  20.         //接受队列中的消息
  21.         Consumer consumer=new DefaultConsumer(channel){
  22.             /**
  23.              * @param consumerTag: 消费者的标签
  24.              * @param envelope : 设置 拿到你的交换机 路由key等信息
  25.              * @param properties: 消息的属性对象
  26.              * @param body: 消息的内容
  27.              * @throws IOException
  28.              */
  29.             @Override
  30.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  31.                 //这里我们在控制台打印一下，看下消费者是否接收到了生产者传过来的信息
  32.                 System.out.println("接受的内容:"+new String(body));
  33.             }
  34.         };
  35.         /**
  36.          * String queue, 队列名
  37.          * boolean autoAck,是否自动确认。 当rabbitmq把消息发送给消费后，消费端自动确认消息。
  38.          * Consumer callback：回调。 当rabbitmq队列中存在消息则触发该回调
  39.          */
  40.         channel.basicConsume("hello_queue",true,consumer);
  41.  
  42.         //这里需要注意一点，不能关闭connection和channel
  43.     }
  44. }
  45.  
  46.  
  47. //消费者2
  48. public class WorkConsumer02 {
  49.     public static void main(String[] args) throws IOException, TimeoutException {
  50. //创建连接，并设置连接信息
  51.         ConnectionFactory factory = new ConnectionFactory();
  52.         //设置rabbitmq服务的地址
  53.         factory.setHost("192.168.47.135");
  54.         //设置rabbitmq服务的端口号
  55.         factory.setPort(5672);
  56.         //设置连接的账号
  57.         factory.setUsername("ykk");
  58.         //设置连接的密码
  59.         factory.setPassword("ykk");
  60.         //设置虚拟主机
  61.         factory.setVirtualHost("/ykk");
  62.         //获取连接对象
  63.         Connection connection = factory.newConnection();
  64.         //获取channel对象
  65.         Channel channel = connection.createChannel();
  66.         //接受队列中的消息
  67.         Consumer consumer=new DefaultConsumer(channel){
  68.             /**
  69.              * @param consumerTag: 消费者的标签
  70.              * @param envelope : 设置 拿到你的交换机 路由key等信息
  71.              * @param properties: 消息的属性对象
  72.              * @param body: 消息的内容
  73.              * @throws IOException
  74.              */
  75.             @Override
  76.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  77.                 //这里我们在控制台打印一下，看下消费者是否接收到了生产者传过来的信息
  78.                 System.out.println("接受的内容:"+new String(body));
  79.             }
  80.         };
  81.         /**
  82.          * String queue, 队列名
  83.          * boolean autoAck,是否自动确认。 当rabbitmq把消息发送给消费后，消费端自动确认消息。
  84.          * Consumer callback：回调。 当rabbitmq队列中存在消息则触发该回调
  85.          */
  86.         channel.basicConsume("hello_queue",true,consumer);
  87.  
  88.         //这里需要注意一点，不能关闭connection和channel
  89.     }
  90. }
 

（3）测试项目
 

首先运行生产者，接着运行两个消费者，然后观察idea控制台打印结果
 

消费者1
 

 

 消费者2
 

 

（4） 总结
 

在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。Work Queues 对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。例如：短信服务部署多个，只需要有一个节点成功发送即可。 
 

3、 Publish/Subscribe（发布订阅模式）
 

（1）发布订阅模型
 

 

 
 
在订阅模型中，多了一个 Exchange 角色，而且过程略有变化：
 P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
 C：消费者，消息的接收者，会一直等待消息到来
 Queue：消息队列，接收消息、缓存消息
 Exchange：交换机（X）。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：
 1. Fanout：广播，将消息交给所有绑定到交换机的队列
 2. Direct：定向，把消息交给符合指定routing key 的队列
 3. Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列
 Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！ 
 

 

（2）搭建测试项目
 

1、生产者
 

  1. public class PublishProducer {
  2.     public static void main(String[] args) throws IOException, TimeoutException {
  3.         //创建连接，并设置连接信息
  4.         ConnectionFactory factory = new ConnectionFactory();
  5.         //设置rabbitmq服务的地址
  6.         factory.setHost("192.168.47.135");
  7.         //设置rabbitmq服务的端口号
  8.         factory.setPort(5672);
  9.         //设置连接的账号
  10.         factory.setUsername("ykk");
  11.         //设置连接的密码
  12.         factory.setPassword("ykk");
  13.         //设置虚拟主机
  14.         factory.setVirtualHost("/ykk");
  15.         //获取连接对象
  16.         Connection connection = factory.newConnection();
  17.         //获取channel对象
  18.         Channel channel = connection.createChannel();
  19.         //创建交换机
  20.         /**
  21.          * String exchange, 交换机的名称 如果不存在则创建，存在则不创建
  22.          * BuiltinExchangeType type, 交换机的类型
  23.          * boolean durable: 是否持久化。
  24.          */
  25.         channel.exchangeDeclare("publish_exchange", BuiltinExchangeType.FANOUT,true);
  26.         //创建队列
  27.         /**
  28.          * String queue, 队列的名称. 如果该名称不存在 则创建如果存在则不创建
  29.          * boolean durable, 该对象是否持久化 当rabbitmq重启后 队列就会消失
  30.          * boolean exclusive, 该队列是否被一个消费者独占
  31.          * boolean autoDelete,当没有消费者时，该队列是否被自动删除
  32.          * Map<String, Object> arguments: 额外参数的设置
  33.          */
  34.         //队列1
  35.         channel.queueDeclare("publish_queue01", true, false, false, null);
  36.         //队列2
  37.         channel.queueDeclare("publish_queue02", true, false, false, null);
  38.         //将队列和交换机绑定
  39.         /**
  40.          * String queue
  41.          * String exchange
  42.          * String routingKey:发布订阅模式没有routingKey则写为""
  43.          */
  44.         channel.queueBind("publish_queue01","publish_exchange","");
  45.         channel.queueBind("publish_queue02","publish_exchange","");
  46.         //发送消息
  47.         /**
  48.          * String exchange, 交换机的名称 简单模式没有交换机使用""表示采用默认交换机
  49.          * String routingKey, 路由标识 如果是简单模式起名为队列的名称
  50.          * BasicProperties props, 消息的属性设置。设置为null
  51.          * byte[] body: 消息的内容
  52.          */
  53.             String msg="publish_exchange rabbitmq";
  54.             channel.basicPublish("publish_exchange","",null,msg.getBytes());
  55.         //关闭资源
  56.         channel.close();
  57.         connection. Close();
  58.     }
  59. }
 

2、消费者
 

  1. //消费者1
  2. public class PublishConsumer01 {
  3.     public static void main(String[] args) throws IOException, TimeoutException {
  4.         //创建连接，并设置连接信息
  5.         ConnectionFactory factory = new ConnectionFactory();
  6.         //设置rabbitmq服务的地址
  7.         factory.setHost("192.168.47.135");
  8.         //设置rabbitmq服务的端口号
  9.         factory.setPort(5672);
  10.         //设置连接的账号
  11.         factory.setUsername("ykk");
  12.         //设置连接的密码
  13.         factory.setPassword("ykk");
  14.         //设置虚拟主机
  15.         factory.setVirtualHost("/ykk");
  16.         //获取连接对象
  17.         Connection connection = factory.newConnection();
  18.         //获取channel对象
  19.         Channel channel = connection.createChannel();
  20.  
  21.         //接受队列中的消息
  22.         Consumer consumer=new DefaultConsumer(channel){
  23.             /**
  24.              * @param consumerTag: 消费者的标签
  25.              * @param envelope : 设置 拿到你的交换机 路由key等信息
  26.              * @param properties: 消息的属性对象
  27.              * @param body: 消息的内容
  28.              * @throws IOException
  29.              */
  30.             @Override
  31.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  32.                 //这里我们在控制台打印一下，看下消费者是否接收到了生产者传过来的信息
  33.                 System.out.println("接受的内容:"+new String(body));
  34.             }
  35.         };
  36.         /**
  37.          * String queue, 队列名
  38.          * boolean autoAck,是否自动确认。 当rabbitmq把消息发送给消费后，消费端自动确认消息。
  39.          * Consumer callback：回调。 当rabbitmq队列中存在消息则触发该回调
  40.          */
  41.         channel.basicConsume("publish_queue01",true,consumer);
  42.  
  43.     }
  44. }
  45.  
  46. //消费者2
  47. public class PublishConsumer02 {
  48.     public static void main(String[] args) throws IOException, TimeoutException {
  49.         //创建连接，并设置连接信息
  50.         ConnectionFactory factory = new ConnectionFactory();
  51.         //设置rabbitmq服务的地址
  52.         factory.setHost("192.168.47.135");
  53.         //设置rabbitmq服务的端口号
  54.         factory.setPort(5672);
  55.         //设置连接的账号
  56.         factory.setUsername("ykk");
  57.         //设置连接的密码
  58.         factory.setPassword("ykk");
  59.         //设置虚拟主机
  60.         factory.setVirtualHost("/ykk");
  61.         //获取连接对象
  62.         Connection connection = factory.newConnection();
  63.         //获取channel对象
  64.         Channel channel = connection.createChannel();
  65.  
  66.         //接受队列中的消息
  67.         Consumer consumer=new DefaultConsumer(channel){
  68.             /**
  69.              * @param consumerTag: 消费者的标签
  70.              * @param envelope : 设置 拿到你的交换机 路由key等信息
  71.              * @param properties: 消息的属性对象
  72.              * @param body: 消息的内容
  73.              * @throws IOException
  74.              */
  75.             @Override
  76.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  77.                 //这里我们在控制台打印一下，看下消费者是否接收到了生产者传过来的信息
  78.                 System.out.println("接受的内容:"+new String(body));
  79.             }
  80.         };
  81.         /**
  82.          * String queue, 队列名
  83.          * boolean autoAck,是否自动确认。 当rabbitmq把消息发送给消费后，消费端自动确认消息。
  84.          * Consumer callback：回调。 当rabbitmq队列中存在消息则触发该回调
  85.          */
  86.         channel.basicConsume("publish_queue02",true,consumer);
  87.  
  88.     }
  89. }
 

（3）测试项目
 

首先运行生产者，接着运行两个消费者，然后观察idea控制台打印结果
 

消费者1
 

 

消费者2
 

 
 

 （4）总结
 

 
 
交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到。
 2. 发布订阅模式与工作队列模式的区别：
 工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机
 
 
发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)
 发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑 定到默认的交换机
 

 

4、Routing（路由模式）
 

（1）路由模型 
 

 

 
 
队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key），消息的发送方在向 Exchange 发送消息时，也必须指定消息的RoutingKey
 
 
Exchange 不再把消息交给每一个绑定的队列，而是根据消息的Routing Key 进行判断，只有队列的Routingkey 与消息的 Routingkey 完全一致，才会接收到消息
 P：生产者，向 Exchange 发送消息，发送消息时，会指定一个routing key
 X：Exchange（交换机），接收生产者的消息，然后把消息递交给与 routing key 完全匹配的队列
 C1：消费者，其所在队列指定了需要 routing key 为 error 的消息
 C2：消费者，其所在队列指定了需要 routing key 为 info、error、warning 的消息
 

 

（2）搭建测试项目
 

1、生产者
 

  1. public class RoutingProducer {
  2.     public static void main(String[] args) throws IOException, TimeoutException {
  3.         //创建连接，并设置连接信息
  4.         ConnectionFactory factory = new ConnectionFactory();
  5.         //设置rabbitmq服务的地址
  6.         factory.setHost("192.168.47.135");
  7.         //设置rabbitmq服务的端口号
  8.         factory.setPort(5672);
  9.         //设置连接的账号
  10.         factory.setUsername("ykk");
  11.         //设置连接的密码
  12.         factory.setPassword("ykk");
  13.         //设置虚拟主机
  14.         factory.setVirtualHost("/ykk");
  15.         //获取连接对象
  16.         Connection connection = factory.newConnection();
  17.         //获取channel对象
  18.         Channel channel = connection.createChannel();
  19.         //创建交换机
  20.         /**
  21.          * String exchange, 交换机的名称 如果不存在则创建，存在则不创建
  22.          * BuiltinExchangeType type, 交换机的类型
  23.          * boolean durable: 是否持久化。
  24.          */
  25.         channel.exchangeDeclare("router_exchange", BuiltinExchangeType.DIRECT,true);
  26.         //创建队列
  27.         /**
  28.          * String queue, 队列的名称. 如果该名称不存在 则创建如果存在则不创建
  29.          * boolean durable, 该对象是否持久化 当rabbitmq重启后 队列就会消失
  30.          * boolean exclusive, 该队列是否被一个消费者独占
  31.          * boolean autoDelete,当没有消费者时，该队列是否被自动删除
  32.          * Map<String, Object> arguments: 额外参数的设置
  33.          */
  34.         //队列1
  35.         channel.queueDeclare("router_queue01", true, false, false, null);
  36.         //队列2
  37.         channel.queueDeclare("router_queue02", true, false, false, null);
  38.         //将队列和交换机绑定
  39.         /**
  40.          * String queue
  41.          * String exchange
  42.          * String routingKey:发布订阅模式没有routingKey则写为""
  43.          */
  44.         channel.queueBind("router_queue01","router_exchange","error");
  45.         channel.queueBind("router_queue02","router_exchange","error");
  46.         channel.queueBind("router_queue02","router_exchange","info");
  47.         channel.queueBind("router_queue02","router_exchange","warning");
  48.         //发送消息
  49.         /**
  50.          * String exchange, 交换机的名称 简单模式没有交换机使用""表示采用默认交换机
  51.          * String routingKey, 路由标识 如果是简单模式起名为队列的名称
  52.          * BasicProperties props, 消息的属性设置。设置为null
  53.          * byte[] body: 消息的内容
  54.          */
  55.             String msg="router_exchange rabbitmq";
  56.             channel.basicPublish("router_exchange","info",null,msg.getBytes());
  57.         //关闭资源
  58.         channel.close();
  59.         connection.close();
  60.     }
  61. }
 

2、消费者
 

  1. //消费者1
  2. public class RoutingConsumer01 {
  3.     public static void main(String[] args) throws IOException, TimeoutException {
  4.         //创建连接，并设置连接信息
  5.         ConnectionFactory factory = new ConnectionFactory();
  6.         //设置rabbitmq服务的地址
  7.         factory.setHost("192.168.47.135");
  8.         //设置rabbitmq服务的端口号
  9.         factory.setPort(5672);
  10.         //设置连接的账号
  11.         factory.setUsername("ykk");
  12.         //设置连接的密码
  13.         factory.setPassword("ykk");
  14.         //设置虚拟主机
  15.         factory.setVirtualHost("/ykk");
  16.         //获取连接对象
  17.         Connection connection = factory.newConnection();
  18.         //获取channel对象
  19.         Channel channel = connection.createChannel();
  20.  
  21.         //接受队列中的消息
  22.         Consumer consumer=new DefaultConsumer(channel){
  23.             /**
  24.              * @param consumerTag: 消费者的标签
  25.              * @param envelope : 设置 拿到你的交换机 路由key等信息
  26.              * @param properties: 消息的属性对象
  27.              * @param body: 消息的内容
  28.              * @throws IOException
  29.              */
  30.             @Override
  31.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  32.                 //这里我们在控制台打印一下，看下消费者是否接收到了生产者传过来的信息
  33.                 System.out.println("接受的内容:"+new String(body));
  34.             }
  35.         };
  36.         /**
  37.          * String queue, 队列名
  38.          * boolean autoAck,是否自动确认。 当rabbitmq把消息发送给消费后，消费端自动确认消息。
  39.          * Consumer callback：回调。 当rabbitmq队列中存在消息则触发该回调
  40.          */
  41.         channel.basicConsume("router_queue01",true,consumer);
  42.  
  43.     }
  44. }
  45. //消费者2
  46. public class RoutingConsumer02 {
  47.     public static void main(String[] args) throws IOException, TimeoutException {
  48.         //创建连接，并设置连接信息
  49.         ConnectionFactory factory = new ConnectionFactory();
  50.         //设置rabbitmq服务的地址
  51.         factory.setHost("192.168.47.135");
  52.         //设置rabbitmq服务的端口号
  53.         factory.setPort(5672);
  54.         //设置连接的账号
  55.         factory.setUsername("ykk");
  56.         //设置连接的密码
  57.         factory.setPassword("ykk");
  58.         //设置虚拟主机
  59.         factory.setVirtualHost("/ykk");
  60.         //获取连接对象
  61.         Connection connection = factory.newConnection();
  62.         //获取channel对象
  63.         Channel channel = connection.createChannel();
  64.  
  65.         //接受队列中的消息
  66.         Consumer consumer=new DefaultConsumer(channel){
  67.             /**
  68.              * @param consumerTag: 消费者的标签
  69.              * @param envelope : 设置 拿到你的交换机 路由key等信息
  70.              * @param properties: 消息的属性对象
  71.              * @param body: 消息的内容
  72.              * @throws IOException
  73.              */
  74.             @Override
  75.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  76.                 //这里我们在控制台打印一下，看下消费者是否接收到了生产者传过来的信息
  77.                 System.out.println("接受的内容:"+new String(body));
  78.             }
  79.         };
  80.         /**
  81.          * String queue, 队列名
  82.          * boolean autoAck,是否自动确认。 当rabbitmq把消息发送给消费后，消费端自动确认消息。
  83.          * Consumer callback：回调。 当rabbitmq队列中存在消息则触发该回调
  84.          */
  85.         channel.basicConsume("router_queue02",true,consumer);
  86.  
  87.     }
  88. }
 

（3）测试项目
 

首先运行生产者，接着运行两个消费者，然后观察idea控制台打印结果
 

 

 消费者1
 

 

消费者2
 

 

（4）总结
 

 
 
 Routing 模式要求队列在绑定交换机时要指定 routing key，消息会转发到符合 routing key 的队列。
 

 

5、Topics（主题模式） 
 

（1）主题模式模型
 

 

 
 
Topic 类型与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型Exchange 可以让队列在绑定Routing key 的时候使用通配符！
 Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert
 通配符规则：# 匹配一个或多个词，* 匹配不多不少恰好1个词，例如：item.# 能够匹配 item.insert.abc 或者 item.insert，item.*只能匹配 item.insert 
 

 

（2）搭建测试项目
 

1、生产者
 

  1. public class TopicsProducer {
  2.     public static void main(String[] args) throws IOException, TimeoutException {
  3.         //创建连接，并设置连接信息
  4.         ConnectionFactory factory = new ConnectionFactory();
  5.         //设置rabbitmq服务的地址
  6.         factory.setHost("192.168.47.135");
  7.         //设置rabbitmq服务的端口号
  8.         factory.setPort(5672);
  9.         //设置连接的账号
  10.         factory.setUsername("ykk");
  11.         //设置连接的密码
  12.         factory.setPassword("ykk");
  13.         //设置虚拟主机
  14.         factory.setVirtualHost("/ykk");
  15.         //获取连接对象
  16.         Connection connection = factory.newConnection();
  17.         //获取channel对象
  18.         Channel channel = connection.createChannel();
  19.         //创建交换机
  20.         /**
  21.          * String exchange, 交换机的名称 如果不存在则创建，存在则不创建
  22.          * BuiltinExchangeType type, 交换机的类型
  23.          * boolean durable: 是否持久化。
  24.          */
  25.         channel.exchangeDeclare("topic_exchange", BuiltinExchangeType.TOPIC,true);
  26.         //创建队列
  27.         /**
  28.          * String queue, 队列的名称. 如果该名称不存在 则创建如果存在则不创建
  29.          * boolean durable, 该对象是否持久化 当rabbitmq重启后 队列就会消失
  30.          * boolean exclusive, 该队列是否被一个消费者独占
  31.          * boolean autoDelete,当没有消费者时，该队列是否被自动删除
  32.          * Map<String, Object> arguments: 额外参数的设置
  33.          */
  34.         //队列1
  35.         channel.queueDeclare("topic_queue01", true, false, false, null);
  36.         //队列2
  37.         channel.queueDeclare("topic_queue02", true, false, false, null);
  38.         //将队列和交换机绑定
  39.         /**
  40.          * String queue
  41.          * String exchange
  42.          * String routingKey:发布订阅模式没有routingKey则写为""
  43.          */
  44.         channel.queueBind("topic_queue01","topic_exchange","*.orange.*");
  45.         channel.queueBind("topic_queue02","topic_exchange","*.*.rabbit");
  46.         channel.queueBind("topic_queue02","topic_exchange","lazy.#");
  47.         //发送消息
  48.         /**
  49.          * String exchange, 交换机的名称 简单模式没有交换机使用""表示采用默认交换机
  50.          * String routingKey, 路由标识 如果是简单模式起名为队列的名称
  51.          * BasicProperties props, 消息的属性设置。设置为null
  52.          * byte[] body: 消息的内容
  53.          */
  54.             String msg="topic_exchange rabbitmq";
  55.             channel.basicPublish("topic_exchange","lazy.rabbit.orange",null,msg.getBytes());
  56.         //关闭资源
  57.         channel.close();
  58.         connection. Close();
  59.     }
  60. }
 

2、消费者
 

  1. //消费者1
  2. public class TopicsConsumer01 {
  3.     public static void main(String[] args) throws IOException, TimeoutException {
  4.         //创建连接，并设置连接信息
  5.         ConnectionFactory factory = new ConnectionFactory();
  6.         //设置rabbitmq服务的地址
  7.         factory.setHost("192.168.47.135");
  8.         //设置rabbitmq服务的端口号
  9.         factory.setPort(5672);
  10.         //设置连接的账号
  11.         factory.setUsername("ykk");
  12.         //设置连接的密码
  13.         factory.setPassword("ykk");
  14.         //设置虚拟主机
  15.         factory.setVirtualHost("/ykk");
  16.         //获取连接对象
  17.         Connection connection = factory.newConnection();
  18.         //获取channel对象
  19.         Channel channel = connection.createChannel();
  20.  
  21.         //接受队列中的消息
  22.         Consumer consumer=new DefaultConsumer(channel){
  23.             /**
  24.              * @param consumerTag: 消费者的标签
  25.              * @param envelope : 设置 拿到你的交换机 路由key等信息
  26.              * @param properties: 消息的属性对象
  27.              * @param body: 消息的内容
  28.              * @throws IOException
  29.              */
  30.             @Override
  31.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  32.                 //这里我们在控制台打印一下，看下消费者是否接收到了生产者传过来的信息
  33.                 System.out.println("接受的内容:"+new String(body));
  34.             }
  35.         };
  36.         /**
  37.          * String queue, 队列名
  38.          * boolean autoAck,是否自动确认。 当rabbitmq把消息发送给消费后，消费端自动确认消息。
  39.          * Consumer callback：回调。 当rabbitmq队列中存在消息则触发该回调
  40.          */
  41.         channel.basicConsume("topic_queue01",true,consumer);
  42.  
  43.     }
  44. }
  45. //消费者2
  46. public class TopicsConsumer02 {
  47.     public static void main(String[] args) throws IOException, TimeoutException {
  48.         //创建连接，并设置连接信息
  49.         ConnectionFactory factory = new ConnectionFactory();
  50.         //设置rabbitmq服务的地址
  51.         factory.setHost("192.168.47.135");
  52.         //设置rabbitmq服务的端口号
  53.         factory.setPort(5672);
  54.         //设置连接的账号
  55.         factory.setUsername("ykk");
  56.         //设置连接的密码
  57.         factory.setPassword("ykk");
  58.         //设置虚拟主机
  59.         factory.setVirtualHost("/ykk");
  60.         //获取连接对象
  61.         Connection connection = factory.newConnection();
  62.         //获取channel对象
  63.         Channel channel = connection.createChannel();
  64.  
  65.         //接受队列中的消息
  66.         Consumer consumer=new DefaultConsumer(channel){
  67.             /**
  68.              * @param consumerTag: 消费者的标签
  69.              * @param envelope : 设置 拿到你的交换机 路由key等信息
  70.              * @param properties: 消息的属性对象
  71.              * @param body: 消息的内容
  72.              * @throws IOException
  73.              */
  74.             @Override
  75.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  76.                 //这里我们在控制台打印一下，看下消费者是否接收到了生产者传过来的信息
  77.                 System.out.println("接受的内容:"+new String(body));
  78.             }
  79.         };
  80.         /**
  81.          * String queue, 队列名
  82.          * boolean autoAck,是否自动确认。 当rabbitmq把消息发送给消费后，消费端自动确认消息。
  83.          * Consumer callback：回调。 当rabbitmq队列中存在消息则触发该回调
  84.          */
  85.         channel.basicConsume("topic_queue02",true,consumer);
  86.  
  87.     }
  88. }
 

（3）测试项目
 

首先运行生产者，接着运行两个消费者，然后观察idea控制台打印结果
 

 

消费者1有结果，消费者2无结果
 

 

（4）总结
 

 
 
Topic 主题模式可以实现 Pub/Sub 发布与订阅模式和 Routing 路由模式的功能，只是 Topic 在配置routing key 的时候可以使用通配符，显得更加灵活。 
 


                

        
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/你好赵伟/article/detail/594716
推荐阅读
相关标签
Copyright ©  2003-2013 www.wpsshop.cn  版权所有，并保留所有权利。

  
闽ICP备14008679号