0
0

RabbitMQ详细实战教程

产自：后端三郎

时间：刚刚

参考：B站编程不良人

1.MQ安装教程

** Docker安装MQ（没有docker的先安装docker，第一篇是安装docker,第二篇是docker安装MQ）**
三郎之——Centos7安装Docker_后端三郎@ZYJ的博客-CSDN博客Centos7安装Dockerhttps://blog.csdn.net/ZhAoYaJ/article/details/122457714?spm=1001.2014.3001.5501

Docker安装Rabbitmq--快捷实用_后端三郎@ZYJ的博客-CSDN博客Docker安装Rabbitmq--快捷实用https://blog.csdn.net/ZhAoYaJ/article/details/123821638?spm=1001.2014.3001.5501

2.什么是MQ

  1.     MQ(Message Quene):翻译为消息队列，通过典型的生产者和消费者模型，生产者不断向消息队列生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步，而且只关心消息的发送和接收，没有业务逻辑的侵入，轻松实现系统间解耦。别名为消息中间件，通过利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成。

3.MQ有哪些

  1.     当今市面上有很多主流的消息中间件，如老牌的ActiveMQ,RabbitMQ,炙手可热的Kafka,阿里巴巴自主开发的RocketMQ等。

4.不同MQ特点

#1.ActiveMQ特点

  1.     **ActiveMQ是Apache出品，最流行的，能力强劲的开源消息总线。他是一个完全支持JMS规范的消息中间件。丰富的API，多种集群架构模式让ActiveMQ在业界成为老牌的消息中间件，在中小型企业颇受欢迎。**

#2.Kafka特点

  1.     **Kafka是LinkedIn开源的分布式发布-订阅消息系统，目前归属于Apache顶级项目。Kafka主要特点是基于Pull的模式来处理消息消费。**

** 追求高吞吐量，一开始的目的就是用于日志的传输和收集。0.8版本开始支持复制，不支持事务，对消息的重复，丢失，错误没有严格要求。**

** 适合产生大量的互联网服务的数据收集业务。**

#3.RocketMQ特点

  1.     **RocketMQ是阿里开源的消息中间件，它是纯Java开发，具有高吞吐量，高可用性，使用大规模分布式系统应用的特点。RocketMQ思路起源于Kafka，但并不是Kafka的一个Copy，他对消息的可靠传输以及事务做了优化，目前在阿里集团被广泛应用于交易，充值，流计算，消息推送，日志流式处理，binglog分发等场景。**

#4.RabbitMQ特点

  1.     **RabbitMQ是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现，AMQP主要特征是面向消息，队列，路由（包括点对点和发布/订阅），可靠性，安全。        **

** AMQP协议更多用在企业系统内对数据的一致性，稳定性和可靠性要求很高的场景。**

** 对性能和吞吐量的要求还在其次。**

  1. **注：RabbitMQ安全系数高，更加可靠，kafka更适用于日志的处理。**

5.web管理页面介绍

  • Connections：无论生产者还是消费者，都需要与RabbitMQ建立连接后才可以完成消息的生产和消费，在这里可以查看连接情况
  • Channels：通道，建立连接后，会形成通道，消息的投递获取依赖通道。
  • Exchanges：交换机，用来实现消息的路由
  • Queues：队列，即消息队列，消息存放在队列中，等待消费，消费后被移除队列。
  • Admin：用户管理，用户权限设置，新增用户等操作。

5.1Admin用户和虚拟主机管理

（1）添加用户

上面的Tags选项，其实是指定用户的角色，可选的有以下几个：

  • 超级管理员(Administrator)

    1.   可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。

  • 监控者(Monitoring)

    1.   可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)

  • 策略制定者(policymaker)

    1.   可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。

  • 普通管理者(Management)

    1.   仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。

  • 其他

    1.   无法登陆管理控制台，通常就是普通的生产者和消费者。

（2）创建虚拟主机

解释：****为了让各个用户可以互不干扰的工作，RabbitMQ添加了虚拟主机（Virtual Hosts）的概念。其实就是一个独立的访问路径，不同用户使用不同路径，各自有自己的队列、交换机，互相不会影响。

（3）虚拟主机和用户绑定

** 进入虚拟主机配置**

** 绑定用户和虚拟主机 **

** 用户和虚拟主机绑定成功**

6.RabbitMQ支持的消息模型

第一种模型：直连

第二种模型：work quene

第三种模型：Publish/Subscribe

第四种模型：Routing

第五种模型：Topic

6.1引入依赖

  1.         <dependency>
  2.             <groupId>com.rabbitmq</groupId>
  3.             <artifactId>amqp-client</artifactId>
  4.             <version>5.7.2</version>
  5.         </dependency>

6.1.1第一种模型(直连)

P：生产者，就是要发送消息的程序。

C：消费者，会一直等待消息的到来。

queue：消息队列，图中红色部分，类似一个邮箱，可以缓存消息，生产者向其中投递消息，消费者从其中取出消息。

开发生产者

  1. public static void main(String[] args) throws IOException, TimeoutException {
  2.         //创建连接MQ的连接工厂对象
  3.         ConnectionFactory connectionFactory = new ConnectionFactory();
  4.         //设置连接rabbitmq主机
  5.         connectionFactory.setHost("8.141.53.148");
  6.         //设置端口号
  7.         connectionFactory.setPort(5672);
  8.         //设置访问虚拟主机的用户名
  9.         connectionFactory.setUsername("guest");
  10.         //设置访问虚拟主机的密码
  11.         connectionFactory.setPassword("guest");
  12.         //设置连接哪个虚拟主机
  13.         connectionFactory.setVirtualHost("/zyj");
  14.         //获取连接对象
  15.         Connection connection = connectionFactory.newConnection();
  16.         //获取连接中的通道
  17.         Channel channel = connection.createChannel();
  18.         //通道绑定消息队列
  19.         //参数1: 队列名称（不存在的情况下自动创建）
  20.         //参数2：用来定义队列是否要持久化 true持久化 false不持久化，如果不持久化在重新启动rabbitmq之后队列会消失
  21.         //参数3：是否独占队列 true独占  false不独占
  22.         //参数4：是否在消费完成后自动删除队列 true自动删除 false不删除
  23.         //参数5：额外附加参数
  24.         channel.queueDeclare("hello",false,false,false,null);
  25.         //发布消息
  26.         //参数1：交换机名称
  27.         //参数2：队列名称
  28.         //参数3：传递消息额外设置，此参数可以设置消息的持久化，不设置持久化重启之后消息会消失，设置为MessageProperties.PERSISTENT_TEXT_PLAIN即是消息持久化
  29.         channel.basicPublish("","hello", MessageProperties.PERSISTENT_TEXT_PLAIN,"hello rabbitmq".getBytes());
  30.         //关闭通道
  31.         channel.close();
  32.         //关闭主机
  33.         connection.close();
  34.     }

** 注：虚拟需要开通5672的端口，云服务器需要安全组配置5672的端口。**

  1.     执行代码，创建队列，向队列中发送消息

开发消费者

  1. public static void main(String[] args) throws Exception {
  2.         //创建连接工厂
  3.         ConnectionFactory connectionFactory = new ConnectionFactory();
  4.         //设置连接rabbitmq主机
  5.         connectionFactory.setHost("8.141.53.148");
  6.         //设置端口号
  7.         connectionFactory.setPort(5672);
  8.         //设置访问虚拟主机的用户名
  9.         connectionFactory.setUsername("guest");
  10.         //设置访问虚拟主机的密码
  11.         connectionFactory.setPassword("guest");
  12.         //设置连接哪个虚拟主机
  13.         connectionFactory.setVirtualHost("/zyj");
  14.         //获取连接对象
  15.         Connection connection = connectionFactory.newConnection();
  16.         //获取连接中的通道
  17.         Channel channel = connection.createChannel();
  18.         //通道绑定消息队列
  19.         //参数1: 队列名称（不存在的情况下自动创建）
  20.         //参数2：用来定义队列是否要持久化 true持久化 false不持久化
  21.         //参数3：是否独占队列 true独占  false不独占
  22.         //参数4：是否在消费完成后自动删除队列 true自动删除 false不删除
  23.         //参数5：额外附加参数
  24.         channel.queueDeclare("hello",false,false,false,null);
  25.         //消费消息
  26.         //参数1：消息队列名称
  27.         //参数2：开始消息的自动确认机制
  28.         //参数3：消费时的回调接口
  29.         channel.basicConsume("hello",true,new DefaultConsumer(channel){
  30.             @Override//最后一个参数：消息队列中取出的消息
  31.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
  32.                 System.out.println("==============="+new String(body));
  33.             }
  34.         });
  35.         //需要不断的监测队列，不推荐关闭(因为我们需要不停的监控回调接口的处理)
  36.         //关闭不输出消费者信息，不关闭可以打印出消费着信息
  37.         //channel.close();
  38.         //connection.close();
  39.     }
  1.     执行代码，获取队列中的消息进行消费，输出队列中的参数。

代码优化封装工具类

  1. package com.example.demo.test.studyRabbitmq;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  7. public class RabbitMqUtils {
  9.     private static ConnectionFactory connectionFactory;
  11.     //static静态代码块的作用是它是在类加载的时候执行
  12.     static{
  13.         //创建连接MQ的连接工厂对象
  14.         connectionFactory = new ConnectionFactory();
  15.         //设置连接rabbitmq主机
  16.         connectionFactory.setHost("8.141.53.148");
  17.         //设置端口号
  18.         connectionFactory.setPort(5672);
  19.         //设置访问虚拟主机的用户名
  20.         connectionFactory.setUsername("guest");
  21.         //设置访问虚拟主机的密码
  22.         connectionFactory.setPassword("guest");
  23.         //设置连接哪个虚拟主机
  24.         connectionFactory.setVirtualHost("/zyj");
  25.     }
  27.     //提供连接对象的方法
  28.     public static Connection getConnection(){
  29.         try {
  30.             //获取连接对象
  31.             return connectionFactory.newConnection();
  32.         }catch (Exception e){
  33.             e.printStackTrace();
  34.         }
  35.         return null;
  36.     }
  38.     //关闭通道和关闭连接对象的方法
  39.     public static void closeConnectionAndChannel(Connection connection, Channel channel){
  40.         try {
  41.             if(channel!=null)    channel.close();
  42.             if(connection!=null)  connection.close();
  43.         } catch (Exception e) {
  44.             e.printStackTrace();
  45.         }
  46.     }
  48. }

生产者代码

  1. package com.example.demo.test.studyRabbitmq;
  3. import com.rabbitmq.client.*;
  5. import java.io.IOException;
  6. import java.util.concurrent.TimeoutException;
  8. public class RabbitmqPublishTest {
  10.     public static void main(String[] args) throws IOException, TimeoutException {
  11.         //调用工具类
  12.         Connection connection = RabbitMqUtils.getConnection();
  13.         //获取连接中的通道
  14.         Channel channel = connection.createChannel();
  15.         //通道绑定消息队列
  16.         //参数1: 队列名称（不存在的情况下自动创建）
  17.         //参数2：用来定义队列是否要持久化 true持久化 false不持久化，如果不持久化在重新启动rabbitmq之后队列会消失
  18.         //参数3：是否独占队列 true独占  false不独占
  19.         //参数4：是否在消费完成后自动删除队列 true自动删除 false不删除
  20.         //参数5：额外附加参数
  21.         channel.queueDeclare("hello",false,false,false,null);
  22.         //发布消息
  23.         //参数1：交换机名称
  24.         //参数2：队列名称
  25.         //参数3：传递消息额外设置，此参数可以设置消息的持久化，不设置持久化重启之后消息会消失，设置为MessageProperties.PERSISTENT_TEXT_PLAIN即是消息持久化
  26.         //参数4：消息的具体内容
  27.         channel.basicPublish("","hello", null,"hello rabbitmq".getBytes());
  28.         //关闭通道和主机
  29.         RabbitMqUtils.closeConnectionAndChannel(connection,channel);
  30.     }
  32. }

消费者代码

  1. package com.example.demo.test.studyRabbitmq;
  3. import com.rabbitmq.client.*;
  5. public class RabbitmqConsumeTest {
  7.     public static void main(String[] args) throws Exception {
  8.         //调用工具类
  9.         Connection connection = RabbitMqUtils.getConnection();
  10.         //获取连接中的通道
  11.         Channel channel = connection.createChannel();
  12.         //通道绑定消息队列
  13.         //参数1: 队列名称（不存在的情况下自动创建）
  14.         //参数2：用来定义队列是否要持久化 true持久化 false不持久化
  15.         //参数3：是否独占队列 true独占  false不独占
  16.         //参数4：是否在消费完成后自动删除队列 true自动删除 false不删除
  17.         //参数5：额外附加参数
  18.         channel.queueDeclare("hello",false,false,false,null);
  19.         //消费消息
  20.         //参数1：消息队列名称
  21.         //参数2：开始消息的自动确认机制
  22.         //参数3：消费时的回调接口
  23.         channel.basicConsume("hello",true,new DefaultConsumer(channel){
  24.             @Override//最后一个参数：消息队列中取出的消息
  25.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
  26.                 System.out.println("==============="+new String(body));
  27.             }
  28.         });
  29.         //需要不断的监测队列，不推荐关闭(因为我们需要不停的监控回调接口的处理)
  30. //        channel.close();
  31. //        connection.close();
  32.     }
  34. }

6.1.2第二种模型(work quene)

P:生产者

C1：消费者1

C2：消费者2

在消息处理比较耗时的时候，可能产生的消息的速度远远大于消息消费的速度，长此以往，消息就会堆积的越来越多，无法及时处理，此时就可以使用第二种模型 work quene ,让多个消费者绑定一个队列，共同消费队列中的消息。

开发生产者

  1. package com.example.demo.test.studyRabbitmq.workQuene;
  3. import com.example.demo.test.studyRabbitmq.RabbitMqUtils;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  7. import java.io.IOException;
  9. public class Provider {
  11.     public static void main(String[] args) throws IOException {
  12.         //获取连接对象
  13.         Connection connection = RabbitMqUtils.getConnection();
  14.         //获取对象通道
  15.         Channel channel = connection.createChannel();
  16.         //通过通道生命队列
  17.         channel.queueDeclare("work",false,false,false,null);
  18.         //循环生产10条消息
  19.         for (int i = 0; i < 10; i++) {
  20.             System.out.println(i);
  21.             //生产消息
  22.             channel.basicPublish("","work",null,(i+"hello word").getBytes());
  23.         }
  24.         RabbitMqUtils.closeConnectionAndChannel(connection,channel);
  25.     }
  27. }

开发消费者1

  1. package com.example.demo.test.studyRabbitmq.workQuene;
  3. import com.example.demo.test.studyRabbitmq.RabbitMqUtils;
  4. import com.rabbitmq.client.*;
  6. import java.io.IOException;
  8. public class Customer1 {
  10.     public static void main(String[] args) throws IOException {
  11.         //获取连接对象
  12.         Connection connection = RabbitMqUtils.getConnection();
  13.         //获取对象通道
  14.         Channel channel = connection.createChannel();
  15.         //通过通道绑定队列
  16.         channel.queueDeclare("work",false,false,false,null);
  17.         //消费消息
  18.         channel.basicConsume("work",true,new DefaultConsumer(channel){
  19.             @Override//最后一个参数：消息队列中取出的消息
  20.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
  21.                 System.out.println("消费者1:"+new String(body));
  22.             }
  23.         });
  24.     }
  26. }

开发消费者2

  1. package com.example.demo.test.studyRabbitmq.workQuene;
  3. import com.example.demo.test.studyRabbitmq.RabbitMqUtils;
  4. import com.rabbitmq.client.*;
  6. import java.io.IOException;
  8. public class Customer2 {
  10.     public static void main(String[] args) throws IOException {
  11.         //获取连接对象
  12.         Connection connection = RabbitMqUtils.getConnection();
  13.         //获取对象通道
  14.         Channel channel = connection.createChannel();
  15.         //通过通道绑定队列
  16.         channel.queueDeclare("work",false,false,false,null);
  17.         //消费消息
  18.         channel.basicConsume("work",true,new DefaultConsumer(channel){
  19.             @Override//最后一个参数：消息队列中取出的消息
  20.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
  21.                 System.out.println("消费者2:"+new String(body));
  22.             }
  23.         });
  24.     }
  26. }

测试结果

总结

默认这种方式是按照顺序将每条消息发送给消费者，每个消费者会收到相同的数量，这种分发消息的方式称为循环。（缺点：假设一个消费者处理快，一个处理慢，会拖垮我们的项目，理想效果是能者多劳，处理快的多处理点，慢的少处理点）

能者多劳

消息确认机制：队列一股脑将10条消息发送给消费者，队列自动确认消息消费完毕（告诉了消费者就认为消费者处理完毕），队列就会删除队列中的数据。但是此时，消费者1处理第二条消息宕机了，队列也会认为处理完毕了消息，就会丢失剩余的消息。如果避免这种情况发生，首先就是关闭消息自动确认机制，不能一次性将消息全部消费。

其次设置每次通道传送给消费者几条消息，当通道关闭将所有消息一股脑的全部发给消费者的时候，也就实现了能者多劳，通道中一条数据，谁执行的快，谁执行下一个消息。

最后每条消息消费完毕之后进行手动确认（好处：完成一条确认一条，保证消息永不丢失）

修改完的消费者1

为了测试效果更加明显，让消费者1处理消息前睡眠1秒或者更多，效果更加明显

  1. package com.example.demo.test.studyRabbitmq.workQuene;
  3. import com.example.demo.test.studyRabbitmq.RabbitMqUtils;
  4. import com.rabbitmq.client.*;
  6. import java.io.IOException;
  8. public class Customer1 {
  10.     public static void main(String[] args) throws IOException {
  11.         //获取连接对象
  12.         Connection connection = RabbitMqUtils.getConnection();
  13.         //获取对象通道
  14.         Channel channel = connection.createChannel();
  15.         //设置每次通道传递一条消息
  16.         channel.basicQos(1);
  17.         //通过通道绑定队列
  18.         channel.queueDeclare("work",false,false,false,null);
  19.         //消费消息
  20.         channel.basicConsume("work",false,new DefaultConsumer(channel){
  21.             @Override//最后一个参数：消息队列中取出的消息
  22.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  23.                 try {
  24.                     Thread.sleep(1000);
  25.                     System.out.println("消费者1:"+new String(body));
  26.                     //消息手动确认机制：参数1：手动确认标识，参数2：是否开启多条消息确认（我们的通道中只有一条，所以false）
  27.                     channel.basicAck(envelope.getDeliveryTag(),false);
  28.                 } catch (InterruptedException e) {
  29.                     e.printStackTrace();
  30.                 }
  32.             }
  33.         });
  34.     }
  36. }

修改完的消费者2

  1. package com.example.demo.test.studyRabbitmq.workQuene;
  3. import com.example.demo.test.studyRabbitmq.RabbitMqUtils;
  4. import com.rabbitmq.client.*;
  6. import java.io.IOException;
  8. public class Customer2 {
  10.     public static void main(String[] args) throws IOException {
  11.         //获取连接对象
  12.         Connection connection = RabbitMqUtils.getConnection();
  13.         //获取对象通道
  14.         Channel channel = connection.createChannel();
  15.         //设置每次通道传递一条消息
  16.         channel.basicQos(1);
  17.         //通过通道绑定队列
  18.         channel.queueDeclare("work",false,false,false,null);
  19.         //消费消息
  20.         channel.basicConsume("work",false,new DefaultConsumer(channel){
  21.             @Override//最后一个参数：消息队列中取出的消息
  22.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  23.                 System.out.println("消费者2:"+new String(body));
  24.                 //消息手动确认机制：参数1：手动确认标识，参数2：是否开启多条消息确认（我们的通道中只有一条，所以false）
  25.                 channel.basicAck(envelope.getDeliveryTag(),false);
  26.             }
  27.         });
  28.     }
  30. }

测试结果

至此：完美实现能者多劳

6.1.3第三种模型(publish/subscribe)

这种模式称为广播模式

可以有多个消费者

每个消费者都有自己的队列

每个队列都需要绑定到交换机

生产者发送消息，只能发送到交换机，由交换机绝对发送给哪个队列，生产者无法决定

交换机把消息发送给所有绑定过的队列

实现了一条消息被多个消费者消费

生产者代码

  1. package com.example.demo.test.studyRabbitmq.fanout;
  3. import com.example.demo.test.studyRabbitmq.RabbitMqUtils;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  7. import java.io.IOException;
  9. public class Provider {
  11.     public static void main(String[] args) throws IOException {
  12.         //获取连接对象
  13.         Connection connection = RabbitMqUtils.getConnection();
  14.         //获取对象通道
  15.         Channel channel = connection.createChannel();
  16.         //将通道声明交换机
  17.         channel.exchangeDeclare("logs","fanout");
  18.         //发送消息到交换机
  19.         channel.basicPublish("logs","",null,"fanout type".getBytes());
  20.         //关闭连接
  21.         RabbitMqUtils.closeConnectionAndChannel(connection,channel);
  23.     }
  25. }

消费者代码1

  1. package com.example.demo.test.studyRabbitmq.fanout;
  3. import com.example.demo.test.studyRabbitmq.RabbitMqUtils;
  4. import com.rabbitmq.client.*;
  6. import java.io.IOException;
  8. public class Customer1 {
  10.     public static void main(String[] args) throws IOException {
  11.         //获取连接对象
  12.         Connection connection = RabbitMqUtils.getConnection();
  13.         //获取对象通道
  14.         Channel channel = connection.createChannel();
  15.         //通道绑定交换机
  16.         channel.exchangeDeclare("logs","fanout");
  17.         //临时队列
  18.         String queue = channel.queueDeclare().getQueue();
  19.         //绑定交换机和队列
  20.         channel.queueBind(queue,"logs","");
  21.         //消费消息
  22.         channel.basicConsume(queue,true,new DefaultConsumer(channel){
  23.             @Override
  24.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body){
  25.                 System.out.println("消费者1："+new String(body));
  26.             }
  27.         });
  30.     }
  32. }

消费者代码2

  1. package com.example.demo.test.studyRabbitmq.fanout;
  3. import com.example.demo.test.studyRabbitmq.RabbitMqUtils;
  4. import com.rabbitmq.client.*;
  6. import java.io.IOException;
  8. public class Customer2 {
  10.     public static void main(String[] args) throws IOException {
  11.         //获取连接对象
  12.         Connection connection = RabbitMqUtils.getConnection();
  13.         //获取对象通道
  14.         Channel channel = connection.createChannel();
  15.         //通道绑定交换机
  16.         channel.exchangeDeclare("logs","fanout");
  17.         //临时队列
  18.         String queue = channel.queueDeclare().getQueue();
  19.         //绑定交换机和队列
  20.         channel.queueBind(queue,"logs","");
  21.         //消费消息
  22.         channel.basicConsume(queue,true,new DefaultConsumer(channel){
  23.             @Override
  24.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body){
  25.                 System.out.println("消费者2："+new String(body));
  26.             }
  27.         });
  29.     }
  31. }

测试结果

6.1.4第四种模型(Routing)

在某些场景下，我们希望不同的消息被不同的队列进行消费，就需要用到这种模型

在Routing的模型下也是（direct路由模式）：

  • 队列与交换机的绑定，不能是任意绑定，而是需要指定一个Routing key（路由key）
  • 消息的发送方在向Exchange发消息时，也必须指定消息的Routing key
  • Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routingkey进行判断，只有队列的Routingkey和消息的Routingkey完全一致，才能接收到消息。

生产者代码

  1. package com.example.demo.test.studyRabbitmq.routing;
  3. import com.example.demo.test.studyRabbitmq.RabbitMqUtils;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  7. import java.io.IOException;
  9. public class Provider {
  11.     public static void main(String[] args) throws IOException {
  12.         //获取连接对象
  13.         Connection connection = RabbitMqUtils.getConnection();
  14.         //获取连接通道对象
  15.         Channel channel = connection.createChannel();
  16.         //将通道声明交换机 参数1交换机名称  参数2路由模式
  17.         channel.exchangeDeclare("logs-routing","direct");
  18.         //发送消息  参数2 Routing-key名称
  19.         channel.basicPublish("logs-routing","info",null,"direct路由模式生产者测试".getBytes());
  20.         //关闭资源
  21.         RabbitMqUtils.closeConnectionAndChannel(connection,channel);
  22.     }
  24. }

消费者代码1

  1. package com.example.demo.test.studyRabbitmq.routing;
  3. import com.example.demo.test.studyRabbitmq.RabbitMqUtils;
  4. import com.rabbitmq.client.*;
  6. import java.io.IOException;
  8. public class Customer1 {
  10.     public static void main(String[] args) throws IOException {
  11.         //获取连接对象
  12.         Connection connection = RabbitMqUtils.getConnection();
  13.         //获取连接通道对象
  14.         Channel channel = connection.createChannel();
  15.         //声明交换机和交换机类型
  16.         channel.exchangeDeclare("logs-routing","direct");
  17.         //创建一个临时队列
  18.         String queue = channel.queueDeclare().getQueue();
  19.         //基于Routing key 绑定交换机和队列,绑定key为error的队列,此处可以绑定多个Routingkey，如下我就绑定了两个info和error
  20.         channel.queueBind(queue,"logs-routing","info");
  21.         channel.queueBind(queue,"logs-routing","error");
  22.         //获取消费消息
  23.         channel.basicConsume(queue,true,new DefaultConsumer(channel){
  24.             @Override
  25.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  26.                 System.out.println("消费者1的消息："+new String(body));
  27.             }
  28.         });
  29.     }
  31. }

消费者代码2

  1. package com.example.demo.test.studyRabbitmq.routing;
  3. import com.example.demo.test.studyRabbitmq.RabbitMqUtils;
  4. import com.rabbitmq.client.*;
  6. import java.io.IOException;
  8. public class Customer2 {
  10.     public static void main(String[] args) throws IOException {
  11.         //获取连接对象
  12.         Connection connection = RabbitMqUtils.getConnection();
  13.         //获取连接通道对象
  14.         Channel channel = connection.createChannel();
  15.         //声明交换机和交换机类型
  16.         channel.exchangeDeclare("logs-routing","direct");
  17.         //创建一个临时队列
  18.         String queue = channel.queueDeclare().getQueue();
  19.         //基于Routing key 绑定交换机和队列,绑定key为error的队列
  20.         channel.queueBind(queue,"logs-routing","error");
  21.         //获取消费消息
  22.         channel.basicConsume(queue,true,new DefaultConsumer(channel){
  23.             @Override
  24.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  25.                 System.out.println("消费者2的消息："+new String(body));
  26.             }
  27.         });
  28.     }
  29. }

测试结果

测试结果可以看出消费者1成功接收到消息，消费者2没有收到消息

6.1.5第五种模型(Topic)

和路由模式不用的是这种方法不用再写死Routing key，可以使用通配符的方式进行匹配

通配符有两种 * 和 # 两种用法如下：例如

user.* 可以匹配user.name user.sex user.key等

user.# 可以匹配user.name.sex.key

*只能代表一个单词，而#可以代表一个或者多个单词

生产者代码

  1. package com.example.demo.test.studyRabbitmq.topic;
  3. import com.example.demo.test.studyRabbitmq.RabbitMqUtils;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  7. import java.io.IOException;
  9. public class Provider {
  11.     public static void main(String[] args) throws IOException {
  12.         //获取连接对象
  13.         Connection connection = RabbitMqUtils.getConnection();
  14.         //获取连接通道对象
  15.         Channel channel = connection.createChannel();
  16.         //将通道声明交换机 参数1交换机名称  参数2路由模式
  17.         channel.exchangeDeclare("topics","topic");
  18.         //发送消息  参数2 Routing-key名称
  19.         channel.basicPublish("topics","user.login.topic",null,"topic模式生产者测试".getBytes());
  20.         //关闭资源
  21.         RabbitMqUtils.closeConnectionAndChannel(connection,channel);
  22.     }
  24. }

消费者代码1

  1. package com.example.demo.test.studyRabbitmq.topic;
  3. import com.example.demo.test.studyRabbitmq.RabbitMqUtils;
  4. import com.rabbitmq.client.*;
  6. import java.io.IOException;
  8. public class Customer1 {
  10.     public static void main(String[] args) throws IOException {
  11.         //获取连接对象
  12.         Connection connection = RabbitMqUtils.getConnection();
  13.         //获取连接通道对象
  14.         Channel channel = connection.createChannel();
  15.         //声明交换机和交换机类型
  16.         channel.exchangeDeclare("topics","topic");
  17.         //创建一个临时队列
  18.         String queue = channel.queueDeclare().getQueue();
  19.         //基于通配符的方式绑定Routingkey
  20.         channel.queueBind(queue,"topics","user.*");
  21.         //获取消费消息
  22.         channel.basicConsume(queue,true,new DefaultConsumer(channel){
  23.             @Override
  24.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  25.                 System.out.println("消费者1的消息："+new String(body));
  26.             }
  27.         });
  28.     }
  29. }

消费者代码2

  1. package com.example.demo.test.studyRabbitmq.topic;
  3. import com.example.demo.test.studyRabbitmq.RabbitMqUtils;
  4. import com.rabbitmq.client.*;
  6. import java.io.IOException;
  8. public class Customer2 {
  10.     public static void main(String[] args) throws IOException {
  11.         //获取连接对象
  12.         Connection connection = RabbitMqUtils.getConnection();
  13.         //获取连接通道对象
  14.         Channel channel = connection.createChannel();
  15.         //声明交换机和交换机类型
  16.         channel.exchangeDeclare("topics","topic");
  17.         //创建一个临时队列
  18.         String queue = channel.queueDeclare().getQueue();
  19.         //基于通配符的方式绑定Routingkey
  20.         channel.queueBind(queue,"topics","user.#");
  21.         //获取消费消息
  22.         channel.basicConsume(queue,true,new DefaultConsumer(channel){
  23.             @Override
  24.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  25.                 System.out.println("消费者1的消息："+new String(body));
  26.             }
  27.         });
  28.     }
  29. }

测试结果

** 完结 **

标签： rabbitmq kafka 分布式
本文转载自: https://blog.csdn.net/ZhAoYaJ/article/details/123659829
版权归原作者 后端三郎@ZYJ 所有， 如有侵权，请联系我们删除。
登录后发布评论

“RabbitMQ详细实战教程”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？