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。

相关文章

mac中资源库在哪? macOS资源库文件夹详解

mac中资源库在哪? macOS资源库文件夹详解

《mac中资源库在哪?macOS资源库文件夹详解》经常使用Mac电脑的用户会发现，找不到Mac电脑的资源库，我们怎么打开资源库并使用呢？下面我们就来看看macOS资源库文件夹详解... 在 MACOS 系统中，「资源库」文件夹是用来存放操作系统和 App 设置的核心位置。虽然平时我们很少直接跟它打交道，但了
阅读更多...
关于Maven中pom.xml文件配置详解

关于Maven中pom.xml文件配置详解

《关于Maven中pom.xml文件配置详解》pom.xml是Maven项目的核心配置文件,它描述了项目的结构、依赖关系、构建配置等信息,通过合理配置pom.xml,可以提高项目的可维护性和构建效率... 目录1. POM文件的基本结构1.1 项目基本信息2. 项目属性2.1 引用属性3. 项目依赖4. 构
阅读更多...
Rust 数据类型详解

Rust 数据类型详解

《Rust数据类型详解》本文介绍了Rust编程语言中的标量类型和复合类型,标量类型包括整数、浮点数、布尔和字符,而复合类型则包括元组和数组,标量类型用于表示单个值,具有不同的表示和范围,本文介绍的非... 目录一、标量类型（Scalar Types）1. 整数类型（Integer Types）1.1 整数字
阅读更多...
Java操作ElasticSearch的实例详解

Java操作ElasticSearch的实例详解

《Java操作ElasticSearch的实例详解》Elasticsearch是一个分布式的搜索和分析引擎,广泛用于全文搜索、日志分析等场景,本文将介绍如何在Java应用中使用Elastics... 目录简介环境准备1. 安装 Elasticsearch2. 添加依赖连接 Elasticsearch1. 创
阅读更多...
Redis缓存问题与缓存更新机制详解

Redis缓存问题与缓存更新机制详解

《Redis缓存问题与缓存更新机制详解》本文主要介绍了缓存问题及其解决方案,包括缓存穿透、缓存击穿、缓存雪崩等问题的成因以及相应的预防和解决方法,同时,还详细探讨了缓存更新机制,包括不同情况下的缓存更... 目录一、缓存问题1.1 缓存穿透1.1.1 问题来源1.1.2 解决方案1.2 缓存击穿1.2.1
阅读更多...
PyTorch使用教程之Tensor包详解

PyTorch使用教程之Tensor包详解

《PyTorch使用教程之Tensor包详解》这篇文章介绍了PyTorch中的张量（Tensor）数据结构,包括张量的数据类型、初始化、常用操作、属性等,张量是PyTorch框架中的核心数据结构,支持... 目录1、张量Tensor2、数据类型3、初始化（构造张量）4、常用操作5、常用属性5.1 存储（st
阅读更多...
Python 中 requests 与 aiohttp 在实际项目中的选择策略详解

Python 中 requests 与 aiohttp 在实际项目中的选择策略详解

《Python中requests与aiohttp在实际项目中的选择策略详解》本文主要介绍了Python爬虫开发中常用的两个库requests和aiohttp的使用方法及其区别,通过实际项目案... 目录一、requests 库二、aiohttp 库三、requests 和 aiohttp 的比较四、requ
阅读更多...
Python实现阶乘的四种写法

Python实现阶乘的四种写法

《Python实现阶乘的四种写法》本文主要介绍了Python实现阶乘的六种写法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧... 目录第一种：推导式+循环遍历列表内每个元素相乘第二种：调用functools模块reduce的php累计
阅读更多...
VUE动态绑定class类的三种常用方式及适用场景详解

VUE动态绑定class类的三种常用方式及适用场景详解

《VUE动态绑定class类的三种常用方式及适用场景详解》文章介绍了在实际开发中动态绑定class的三种常见情况及其解决方案,包括根据不同的返回值渲染不同的class样式、给模块添加基础样式以及根据设... 目录前言1.动态选择class样式（对象添加：情景一）2.动态添加一个class样式（字符串添加：情
阅读更多...
Python在固定文件夹批量创建固定后缀的文件(方法详解)

Python在固定文件夹批量创建固定后缀的文件(方法详解)

《Python在固定文件夹批量创建固定后缀的文件(方法详解)》文章讲述了如何使用Python批量创建后缀为.md的文件夹,生成100个,代码中需要修改的路径、前缀和后缀名,并提供了注意事项和代码示例,... 目录1. python需求的任务2. Python代码的实现3. 代码修改的位置4. 运行结果5.
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2