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。

相关文章

C++ move 的作用详解及陷阱最佳实践

C++ move 的作用详解及陷阱最佳实践

《C++move的作用详解及陷阱最佳实践》文章详细介绍了C++中的`std::move`函数的作用,包括为什么需要它、它的本质、典型使用场景、以及一些常见陷阱和最佳实践,感兴趣的朋友跟随小编一起看... 目录C++ move 的作用详解一、一句话总结二、为什么需要 move？C++98/03 的痛点⚡C++
阅读更多...
MySQL中between and的基本用法、范围查询示例详解

MySQL中between and的基本用法、范围查询示例详解

《MySQL中betweenand的基本用法、范围查询示例详解》BETWEENAND操作符在MySQL中用于选择在两个值之间的数据,包括边界值,它支持数值和日期类型,示例展示了如何使用BETWEEN... 目录一、between and语法二、使用示例2.1、betwphpeen and数值查询2.2、be
阅读更多...
python中的flask_sqlalchemy的使用及示例详解

python中的flask_sqlalchemy的使用及示例详解

《python中的flask_sqlalchemy的使用及示例详解》文章主要介绍了在使用SQLAlchemy创建模型实例时,通过元类动态创建实例的方式,并说明了如何在实例化时执行__init__方法,... 目录@orm.reconstructorSQLAlchemy的回滚关联其他模型数据库基本操作将数据添
阅读更多...
Java中ArrayList与顺序表示例详解

Java中ArrayList与顺序表示例详解

《Java中ArrayList与顺序表示例详解》顺序表是在计算机内存中以数组的形式保存的线性表,是指用一组地址连续的存储单元依次存储数据元素的线性结构,：本文主要介绍Java中ArrayList与... 目录前言一、Java集合框架核心接口与分类ArrayList二、顺序表数据结构中的顺序表三、常用代码手动
阅读更多...
MySQL快速复制一张表的四种核心方法(包括表结构和数据)

MySQL快速复制一张表的四种核心方法(包括表结构和数据)

《MySQL快速复制一张表的四种核心方法(包括表结构和数据)》本文详细介绍了四种复制MySQL表（结构+数据）的方法,并对每种方法进行了对比分析,适用于不同场景和数据量的复制需求,特别是针对超大表（1... 目录一、mysql 复制表（结构+数据）的 4 种核心方法（面试结构化回答）方法 1：CREATE
阅读更多...
JAVA线程的周期及调度机制详解

JAVA线程的周期及调度机制详解

《JAVA线程的周期及调度机制详解》Java线程的生命周期包括NEW、RUNNABLE、BLOCKED、WAITING、TIMED_WAITING和TERMINATED,线程调度依赖操作系统,采用抢占... 目录Java线程的生命周期线程状态转换示例代码JAVA线程调度机制优先级设置示例注意事项JAVA线程
阅读更多...
详解C++ 存储二进制数据容器的几种方法

详解C++ 存储二进制数据容器的几种方法

《详解C++存储二进制数据容器的几种方法》本文主要介绍了详解C++存储二进制数据容器,包括std::vector、std::array、std::string、std::bitset和std::ve... 目录1.std::vector<uint8_t>（最常用）特点：适用场景：示例：2.std::arra
阅读更多...
C++构造函数中explicit详解

C++构造函数中explicit详解

《C++构造函数中explicit详解》explicit关键字用于修饰单参数构造函数或可以看作单参数的构造函数,阻止编译器进行隐式类型转换或拷贝初始化,本文就来介绍explicit的使用,感兴趣的可以... 目录1. 什么是explicit2. 隐式转换的问题3.explicit的使用示例基本用法多参数构造
阅读更多...
Android使用java实现网络连通性检查详解

Android使用java实现网络连通性检查详解

《Android使用java实现网络连通性检查详解》这篇文章主要为大家详细介绍了Android使用java实现网络连通性检查的相关知识,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录NetCheck.Java（可直接拷贝）使用示例（Activity/Fragment 内）权限要求
阅读更多...
MyBatis中的两种参数传递类型详解(示例代码)

MyBatis中的两种参数传递类型详解(示例代码)

《MyBatis中的两种参数传递类型详解(示例代码)》文章介绍了MyBatis中传递多个参数的两种方式,使用Map和使用@Param注解或封装POJO,Map方式适用于动态、不固定的参数,但可读性和安... 目录✅ android方式一：使用Map<String, Object>✅ 方式二：使用@Param
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2