RabbitMQ四种Exchange：fanout,direct,topic,header详解

2024-05-25 07:48
文章标签 详解 rabbitmq 四种 exchange header direct fanout topic

本文主要是介绍RabbitMQ四种Exchange：fanout,direct,topic,header详解，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

主流消息中间件介绍- RabbitMQ

 

RabbitMQ是使用 Erlang语言开发的开源消息队列系统,基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由(包括点对点和发布,订阅)、可靠性、安全。AMQP协议更多用在企业系统内,对数据一致性、稳定性和可靠性要求很高的场景,对性能和吞吐量的要求还在其次。

RabbitMQ高性能的原因?

Erlang语言最初在于交换机领域的架构模式,这样使得RabbitMQ在 Broker之间进行数据交互的性能是非常优秀的

Erlang的优点: Erlang有着和原生 Socket一样的延迟

什么是AMQP高级消息队列协议?

AMQP定义:是具有现代特征的二进制协议。是一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放
准,为面向消息的中间件设计。

 

AMQP协议模型

 

AMQP核心概念

Server:又称 Broker,接受客户端的连接,实现AMQP实体服务
Connection:连接,应用程序与 Broker的网络连接

Channel:网络信道,几乎所有的操作都在 Channe中进行, Channel是进行消息读写的通道。客户端可建立多个 Channel,每个 Channe代表一个会话任务。

Message:消息,服务器和应用程序之间传送的数据,由 Properties和Body组成。 Properties可以对消息进行修饰,比如消息的优先级、延迟等高级特性;Body则就是消息体内容。

Virtual host:虚拟地址,用于进行逻辑隔离,最上层的消息路由。一个Virtual host里面可以有若干个 Exchange和 Queue,同一个 Virtual Host里面不能有相同名称的 Exchange或 Queue

Exchange:交换机,接收消息,根据路由键转发消息到绑定的队列
Routing key:一个路由规则,虚拟机可用它来确定如何路由一个特定消息
Queue:也称为 Message Queue,消息队列,保存消息并将它们转发给消费者
 

RabbitMQ的整体架构是什么样子的?

 

RabbitMQ消息是如何流转的?

 

急速入门-消息生产与消费

Connection Factory:获取连接工厂
Connection:一个连接
Channel:数据通信信道,可发送和接收消息

Queue:具体的消息存储队列
Producer& Consumer生产和消费者

Procuder 生产

public class Procuder {public static void main(String[] args) throws Exception {//1 创建一个ConnectionFactory, 并进行配置ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 通过连接工厂创建连接Connection connection = connectionFactory.newConnection();//3 通过connection创建一个ChannelChannel channel = connection.createChannel();//4 通过Channel发送数据for(int i=0; i < 5; i++){String msg = "Hello RabbitMQ!";//1 exchange   2 routingKeychannel.basicPublish("", "test001", null, msg.getBytes());}//5 记得要关闭相关的连接channel.close();connection.close();}
}

Consumer消费

public class Consumer {public static void main(String[] args) throws Exception {//1 创建一个ConnectionFactory, 并进行配置ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 通过连接工厂创建连接Connection connection = connectionFactory.newConnection();//3 通过connection创建一个ChannelChannel channel = connection.createChannel();//4 声明（创建）一个队列String queueName = "test001";channel.queueDeclare(queueName, true, false, false, null);//5 创建消费者QueueingConsumer queueingConsumer = new QueueingConsumer(channel);//6 设置Channelchannel.basicConsume(queueName, true, queueingConsumer);while(true){//7 获取消息Delivery delivery = queueingConsumer.nextDelivery();String msg = new String(delivery.getBody());System.err.println("消费端: " + msg);//Envelope envelope = delivery.getEnvelope();}}
}

 

Exchange交换机

 

Exchange交换机属性

Name:交换机名称
Type:交换机类型 direct、 topic、 fanout、 headers
Durability:是否需要持久化,true为持久化

Auto delete:当最后一个绑定到 Exchange上的队列删除后,自动删除该 Exchange
Internal:当前 Exchange是否用于 RabbitMQ内部使用,默认为 False
Arguments:扩展参数,用于扩展AMQP协议自制定化使用

在rabbitmq中，exchange有4个类型：direct，topic，fanout，header。


Direct Exchange---直连routingKey 完全一致匹配

所有发送到 Direct Exchange的消息被转发到 RouteKey中指定的 Queue
注意: Direct模式可以使用 RabbitMQ自带的 Exchange: default Exchange,所以不需要将 Exchange进行任何绑定( binding操作,消息传递时, Routekey必须完全匹配才会被队列接收,否则该消息会被抛弃


DirectExchange

Producer4DirectExchange

public class Producer4DirectExchange {public static void main(String[] args) throws Exception {//1 创建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 创建ConnectionConnection connection = connectionFactory.newConnection();//3 创建ChannelChannel channel = connection.createChannel();  //4 声明String exchangeName = "test_direct_exchange";String routingKey = "test.direct";//5 发送String msg = "Hello World RabbitMQ 4  Direct Exchange Message 111 ... ";channel.basicPublish(exchangeName, routingKey , null , msg.getBytes()); 		}}

 

Consumer4DirectExchange

public class Consumer4DirectExchange {public static void main(String[] args) throws Exception {ConnectionFactory connectionFactory = new ConnectionFactory() ;  connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setAutomaticRecoveryEnabled(true);connectionFactory.setNetworkRecoveryInterval(3000);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();  //4 声明String exchangeName = "test_direct_exchange";String exchangeType = "direct";String queueName = "test_direct_queue";String routingKey = "test.direct";//表示声明了一个交换机channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);//表示声明了一个队列channel.queueDeclare(queueName, false, false, false, null);//建立一个绑定关系:channel.queueBind(queueName, exchangeName, routingKey);//durable 是否持久化消息QueueingConsumer consumer = new QueueingConsumer(channel);//参数：队列名称、是否自动ACK、Consumerchannel.basicConsume(queueName, true, consumer);  //循环获取消息  while(true){  //获取消息，如果没有消息，这一步将会一直阻塞  Delivery delivery = consumer.nextDelivery();  String msg = new String(delivery.getBody());    System.out.println("收到消息：" + msg);  } }
}

Topic

Topic Exchange

所有发送到 Topic Exchange的消息被转发到所有关心 RouteKey中指定 Topic的 Queue上

Exchange将 RouteKey和某 Topic进行模糊匹配,此时队列需要绑定一个 Topic

注意:可以使用通配符进行模糊匹配

 

 

 

Producer4TopicExchange

public class Producer4TopicExchange {public static void main(String[] args) throws Exception {//1 创建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 创建ConnectionConnection connection = connectionFactory.newConnection();//3 创建ChannelChannel channel = connection.createChannel();  //4 声明String exchangeName = "test_topic_exchange";String routingKey1 = "user.save";String routingKey2 = "user.update";String routingKey3 = "user.delete.abc";//5 发送String msg = "Hello World RabbitMQ 4 Topic Exchange Message ...";String msg1 = "Hello1 World RabbitMQ 4 Topic Exchange Message ...";String msg2 = "Hello2 World RabbitMQ 4 Topic Exchange Message ...";channel.basicPublish(exchangeName, routingKey1 , null , msg1.getBytes());channel.basicPublish(exchangeName, routingKey2 , null , msg.getBytes()); 	channel.basicPublish(exchangeName, routingKey3 , null , msg.getBytes()); channel.close();  connection.close();  }}

 

Consumer4TopicExchange

public class Consumer4TopicExchange {public static void main(String[] args) throws Exception {ConnectionFactory connectionFactory = new ConnectionFactory() ;  connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setAutomaticRecoveryEnabled(true);connectionFactory.setNetworkRecoveryInterval(3000);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();  //4 声明String exchangeName = "test_topic_exchange";String exchangeType = "topic";String queueName = "test_topic_queue";String routingKey = "user.#";
//		String routingKey = "user.save";
//		String routingKey = "user.*";// 1 声明交换机 channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);// 2 声明队列channel.queueDeclare(queueName, false, false, false, null);// 3 建立交换机和队列的绑定关系:channel.queueBind(queueName, exchangeName, routingKey);//durable 是否持久化消息QueueingConsumer consumer = new QueueingConsumer(channel);//参数：队列名称、是否自动ACK、Consumerchannel.basicConsume(queueName, true, consumer);  //循环获取消息  while(true){  //获取消息，如果没有消息，这一步将会一直阻塞  Delivery delivery = consumer.nextDelivery();  String msg = new String(delivery.getBody());    System.out.println("收到消息：" + msg);  } }
}

Fanout

Fanout Exchange

不处理路由键,只需要简单的将队列绑定到交换机上
发送到交换机的消息都会被转发到与该交换机绑定的所有队列上
Fanout交换机转发消息是最快的

 

Consumer4FanoutExchange

public class Consumer4FanoutExchange {public static void main(String[] args) throws Exception {ConnectionFactory connectionFactory = new ConnectionFactory() ;  connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setAutomaticRecoveryEnabled(true);connectionFactory.setNetworkRecoveryInterval(3000);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();  //4 声明String exchangeName = "test_fanout_exchange";String exchangeType = "fanout";String queueName = "test_fanout_queue";String routingKey = "";	//不设置路由键channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);channel.queueDeclare(queueName, false, false, false, null);channel.queueBind(queueName, exchangeName, routingKey);//durable 是否持久化消息QueueingConsumer consumer = new QueueingConsumer(channel);//参数：队列名称、是否自动ACK、Consumerchannel.basicConsume(queueName, true, consumer); //循环获取消息  while(true){  //获取消息，如果没有消息，这一步将会一直阻塞  Delivery delivery = consumer.nextDelivery();  String msg = new String(delivery.getBody());    System.out.println("收到消息：" + msg);  } }
}

Producer4FanoutExchange

public class Producer4FanoutExchange {public static void main(String[] args) throws Exception {//1 创建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 创建ConnectionConnection connection = connectionFactory.newConnection();//3 创建ChannelChannel channel = connection.createChannel();  //4 声明String exchangeName = "test_fanout_exchange";//5 发送for(int i = 0; i < 10; i ++) {String msg = "Hello World RabbitMQ 4 FANOUT Exchange Message ...";channel.basicPublish(exchangeName, "", null , msg.getBytes());}channel.close();  connection.close();  }}

Binding-绑定
Exchange和 Exchange、 Queue之间的连接关系
Binding中可以包含 RoutingKey或者参数

Queue-消息队列

消息队列,实际存储消息数据
Durability:是否持久化, Durable:是, Transient:否
auto delete:如选yes,代表当最后一个监听被移除之后,
该 Queue会自动被删除

Message-消息

服务器和应用程序之间传送的数据
本质上就是一段数据,由 Properties和 Payload(Body)组成
常用属性: delivery mode、 headers(自定义属性)

Message-其他属性
content_type、 content_encoding、 priority
correlation_id、 reply_to、 expiration、 message_id

timestamp、type、 user_id、 app_id、 cluster_id

 

message

public class Procuder {public static void main(String[] args) throws Exception {//1 创建一个ConnectionFactory, 并进行配置ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.11.76");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 通过连接工厂创建连接Connection connection = connectionFactory.newConnection();//3 通过connection创建一个ChannelChannel channel = connection.createChannel();Map<String, Object> headers = new HashMap<>();headers.put("my1", "111");headers.put("my2", "222");AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().deliveryMode(2).contentEncoding("UTF-8").expiration("10000").headers(headers).build();//4 通过Channel发送数据for(int i=0; i < 5; i++){String msg = "Hello RabbitMQ!";//1 exchange   2 routingKeychannel.basicPublish("", "test001", properties, msg.getBytes());}//5 记得要关闭相关的连接channel.close();connection.close();}
}

public class Consumer {public static void main(String[] args) throws Exception {//1 创建一个ConnectionFactory, 并进行配置ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.11.76");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 通过连接工厂创建连接Connection connection = connectionFactory.newConnection();//3 通过connection创建一个ChannelChannel channel = connection.createChannel();//4 声明（创建）一个队列String queueName = "test001";channel.queueDeclare(queueName, true, false, false, null);//5 创建消费者QueueingConsumer queueingConsumer = new QueueingConsumer(channel);//6 设置Channelchannel.basicConsume(queueName, true, queueingConsumer);while(true){//7 获取消息Delivery delivery = queueingConsumer.nextDelivery();String msg = new String(delivery.getBody());System.err.println("消费端: " + msg);Map<String, Object> headers = delivery.getProperties().getHeaders();System.err.println("headers get my1 value: " + headers.get("my1"));//Envelope envelope = delivery.getEnvelope();}}
}

 

Virtual host虚拟主机

虚拟地址,用于进行逻辑隔离,最上层的消息路由
一个Virtual  host里面可以有若干个 Exchange和 Queue
同一个 Virtual host里面不能有相同名称的 Exchange或 Queue

 

 

这篇关于RabbitMQ四种Exchange：fanout,direct,topic,header详解的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1000903。

相关文章

SpringBoot3.4配置校验新特性的用法详解

SpringBoot3.4配置校验新特性的用法详解

《SpringBoot3.4配置校验新特性的用法详解》SpringBoot3.4对配置校验支持进行了全面升级,这篇文章为大家详细介绍了一下它们的具体使用,文中的示例代码讲解详细,感兴趣的小伙伴可以参考... 目录基本用法示例定义配置类配置 application.yml注入使用嵌套对象与集合元素深度校验开发
阅读更多...
Python中的Walrus运算符分析示例详解

Python中的Walrus运算符分析示例详解

《Python中的Walrus运算符分析示例详解》Python中的Walrus运算符（:=）是Python3.8引入的一个新特性,允许在表达式中同时赋值和返回值,它的核心作用是减少重复计算,提升代码简... 目录1. 在循环中避免重复计算2. 在条件判断中同时赋值变量3. 在列表推导式或字典推导式中简化逻辑
阅读更多...
Java Stream流使用案例深入详解

Java Stream流使用案例深入详解

《JavaStream流使用案例深入详解》：本文主要介绍JavaStream流使用案例详解,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录前言1. Lambda1.1 语法1.2 没参数只有一条语句或者多条语句1.3 一个参数只有一条语句或者多
阅读更多...
SpringBoot整合mybatisPlus实现批量插入并获取ID详解

SpringBoot整合mybatisPlus实现批量插入并获取ID详解

《SpringBoot整合mybatisPlus实现批量插入并获取ID详解》这篇文章主要为大家详细介绍了SpringBoot如何整合mybatisPlus实现批量插入并获取ID,文中的示例代码讲解详细... 目录【1】saveBATch(一万条数据总耗时：2478ms)【2】集合方式foreach(一万条数
阅读更多...
Python装饰器之类装饰器详解

Python装饰器之类装饰器详解

《Python装饰器之类装饰器详解》本文将详细介绍Python中类装饰器的概念、使用方法以及应用场景,并通过一个综合详细的例子展示如何使用类装饰器,希望对大家有所帮助,如有错误或未考虑完全的地方,望不... 目录1. 引言2. 装饰器的基本概念2.1. 函数装饰器复习2.2 类装饰器的定义和使用3. 类装饰
阅读更多...
MySQL 中的 JSON 查询案例详解

MySQL 中的 JSON 查询案例详解

《MySQL中的JSON查询案例详解》：本文主要介绍MySQL的JSON查询的相关知识,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录mysql 的 jsON 路径格式基本结构路径组件详解特殊语法元素实际示例简单路径复杂路径简写操作符注意MySQL 的 J
阅读更多...
Python ZIP文件操作技巧详解

Python ZIP文件操作技巧详解

《PythonZIP文件操作技巧详解》在数据处理和系统开发中,ZIP文件操作是开发者必须掌握的核心技能,Python标准库提供的zipfile模块以简洁的API和跨平台特性,成为处理ZIP文件的首选... 目录一、ZIP文件操作基础三板斧1.1 创建压缩包1.2 解压操作1.3 文件遍历与信息获取二、进阶技
阅读更多...
一文详解Java异常处理你都了解哪些知识

一文详解Java异常处理你都了解哪些知识

《一文详解Java异常处理你都了解哪些知识》：本文主要介绍Java异常处理的相关资料,包括异常的分类、捕获和处理异常的语法、常见的异常类型以及自定义异常的实现,文中通过代码介绍的非常详细,需要的朋... 目录前言一、什么是异常二、异常的分类2.1 受检异常2.2 非受检异常三、异常处理的语法3.1 try-
阅读更多...
Java中的@SneakyThrows注解用法详解

Java中的@SneakyThrows注解用法详解

《Java中的@SneakyThrows注解用法详解》：本文主要介绍Java中的@SneakyThrows注解用法的相关资料,Lombok的@SneakyThrows注解简化了Java方法中的异常... 目录前言一、@SneakyThrows 简介1.1 什么是 Lombok？二、@SneakyThrows
阅读更多...
Java中字符串转时间与时间转字符串的操作详解

Java中字符串转时间与时间转字符串的操作详解

《Java中字符串转时间与时间转字符串的操作详解》Java的java.time包提供了强大的日期和时间处理功能,通过DateTimeFormatter可以轻松地在日期时间对象和字符串之间进行转换,下面... 目录一、字符串转时间（一）使用预定义格式（二）自定义格式二、时间转字符串（一）使用预定义格式（二）自
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2