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。

相关文章

Spring Security基于数据库验证流程详解

Spring Security基于数据库验证流程详解

Spring Security 校验流程图 相关解释说明（认真看哦） AbstractAuthenticationProcessingFilter 抽象类 /*** 调用 #requiresAuthentication(HttpServletRequest, HttpServletResponse) 决定是否需要进行验证操作。* 如果需要验证，则会调用 #attemptAuthentica
阅读更多...
OpenHarmony鸿蒙开发（ Beta5.0）无感配网详解

OpenHarmony鸿蒙开发（ Beta5.0）无感配网详解

1、简介 无感配网是指在设备联网过程中无需输入热点相关账号信息，即可快速实现设备配网，是一种兼顾高效性、可靠性和安全性的配网方式。 2、配网原理 2.1 通信原理 手机和智能设备之间的信息传递，利用特有的NAN协议实现。利用手机和智能设备之间的WiFi 感知订阅、发布能力，实现了数字管家应用和设备之间的发现。在完成设备间的认证和响应后，即可发送相关配网数据。同时还支持与常规Sof
阅读更多...
6.1.数据结构-c/c++堆详解下篇（堆排序，TopK问题）

6.1.数据结构-c/c++堆详解下篇（堆排序，TopK问题）

上篇：6.1.数据结构-c/c++模拟实现堆上篇（向下，上调整算法，建堆，增删数据）-CSDN博客 本章重点 1.使用堆来完成堆排序 2.使用堆解决TopK问题 目录 一.堆排序 1.1 思路 1.2 代码 1.3 简单测试 二.TopK问题 2.1 思路(求最小)： 2.2 C语言代码（手写堆） 2.3 C++代码（使用优先级队列 priority_queue）
阅读更多...
K8S（Kubernetes）开源的容器编排平台安装步骤详解

K8S（Kubernetes）开源的容器编排平台安装步骤详解

K8S（Kubernetes）是一个开源的容器编排平台，用于自动化部署、扩展和管理容器化应用程序。以下是K8S容器编排平台的安装步骤、使用方式及特点的概述： 安装步骤： 安装Docker：K8S需要基于Docker来运行容器化应用程序。首先要在所有节点上安装Docker引擎。 安装Kubernetes Master：在集群中选择一台主机作为Master节点，安装K8S的控制平面组件，如AP
阅读更多...
RabbitMQ练习（AMQP 0-9-1 Overview）

RabbitMQ练习（AMQP 0-9-1 Overview）

1、What is AMQP 0-9-1 AMQP 0-9-1（高级消息队列协议）是一种网络协议，它允许遵从该协议的客户端（Publisher或者Consumer）应用程序与遵从该协议的消息中间件代理（Broker，如RabbitMQ）进行通信。 AMQP 0-9-1模型的核心概念包括消息发布者（producers/publisher）、消息（messages）、交换机（exchanges）、
阅读更多...
嵌入式Openharmony系统构建与启动详解

嵌入式Openharmony系统构建与启动详解

大家好，今天主要给大家分享一下，如何构建Openharmony子系统以及系统的启动过程分解。 第一：OpenHarmony系统构建      首先熟悉一下，构建系统是一种自动化处理工具的集合，通过将源代码文件进行一系列处理，最终生成和用户可以使用的目标文件。这里的目标文件包括静态链接库文件、动态链接库文件、可执行文件、脚本文件、配置文件等。      我们在编写hellowor
阅读更多...
LabVIEW FIFO详解

LabVIEW FIFO详解

在LabVIEW的FPGA开发中，FIFO（先入先出队列）是常用的数据传输机制。通过配置FIFO的属性，工程师可以在FPGA和主机之间，或不同FPGA VIs之间进行高效的数据传输。根据具体需求，FIFO有多种类型与实现方式，包括目标范围内FIFO（Target-Scoped）、DMA FIFO以及点对点流（Peer-to-Peer）。 FIFO类型 **目标范围FIFO（Target-Sc
阅读更多...
019、JOptionPane类的常用静态方法详解

019、JOptionPane类的常用静态方法详解

目录 JOptionPane类的常用静态方法详解 1. showInputDialog()方法 1.1基本用法 1.2带有默认值的输入框 1.3带有选项的输入对话框 1.4自定义图标的输入对话框 2. showConfirmDialog()方法 2.1基本用法 2.2自定义按钮和图标 2.3带有自定义组件的确认对话框 3. showMessageDialog()方法 3.1
阅读更多...
脏页的标记方式详解

脏页的标记方式详解

脏页的标记方式 一、引言 在数据库系统中，脏页是指那些被修改过但还未写入磁盘的数据页。为了有效地管理这些脏页并确保数据的一致性，数据库需要对脏页进行标记。了解脏页的标记方式对于理解数据库的内部工作机制和优化性能至关重要。 二、脏页产生的过程 当数据库中的数据被修改时，这些修改首先会在内存中的缓冲池（Buffer Pool）中进行。例如，执行一条 UPDATE 语句修改了某一行数据，对应的缓
阅读更多...
ActiveMQ—Queue与Topic区别

ActiveMQ—Queue与Topic区别

Queue与Topic区别 转自：http://blog.csdn.net/qq_21033663/article/details/52458305 队列（Queue）和主题（Topic）是JMS支持的两种消息传递模型：         1、点对点（point-to-point，简称PTP）Queue消息传递模型：         通过该消息传递模型，一个应用程序（即消息生产者）可以
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2