RabbitMQ routing模式

2023-12-08 23:19
文章标签 模式 rabbitmq routing

本文主要是介绍RabbitMQ routing模式,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

首先理解一下routing模式

原理同路由器,根据不同的rountKey可以指定发送到某一“IP”地址的接收者,例如发送者指定发送给IP为192.168.1.1的接收者。但是允许存在不同的接收者之间公用同一IP,即不同接收者允许有多个IP地址。

也可以多个queue绑定为同一routKey

还是先上代码

package com.tencent.routing;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;public class EmitLogDirect {private static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "direct");String severity = getSeverity(argv);String message = getMessage(argv);channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());System.out.println(" [x] Sent '" + severity + "':'" + message + "'");channel.close();connection.close();}private static String getSeverity(String[] strings){if (strings.length < 1)return "info";return strings[0];}private static String getMessage(String[] strings){ if (strings.length < 2)//如果数据不够 则默认发送hello Worldreturn "Hello World!";return joinStrings(strings, " ", 1);}private static String joinStrings(String[] strings, String delimiter, int startIndex) {int length = strings.length;if (length == 0 ) return "";if (length < startIndex ) return "";StringBuilder words = new StringBuilder(strings[startIndex]);for (int i = startIndex + 1; i < length; i++) {words.append(delimiter).append(strings[i]);}return words.toString();}
}

     来看一下发送端producing端的代码,开始new了ConnectionFactory并open了相应的Connection和Channel.

     关键代码出现了declare了一个Exchange 

  channel.exchangeDeclare(EXCHANGE_NAME, "direct");

    exchange有directtopicheaders 和 fanout这4种模式,fanout是在Publish/Subscribe模式下使用,即所有的queue都能接受到exchange分发的massage,

现在要讨论的routing模式,则采用的是direct(直发)模式。

  然后channel在publish message的时候

 channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
  这里的severity对应的是下图中的'orange' ‘black’和'green' 也就是routKey

   下图中有2个queue  Q1和Q2 绑定(binding)到了exchange X上面


     也就是说exchange决定了分发到哪个绑定了的queue

package com.tencent.routing;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;public class ReceiveLogsDirect {private static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "direct");String queueName = channel.queueDeclare().getQueue();if (argv.length < 1){System.err.println("Usage: ReceiveLogsDirect [info] [warning] [error]");System.exit(1);}for(String severity : argv){    channel.queueBind(queueName, EXCHANGE_NAME, severity);//argv是用户的输入,指定这个Receive什么severity标签的数据//argv指定error 该ReceivelLogDirect就只能接受binding到了error的Message}System.out.println(" [*] Waiting for messages. To exit press CTRL+C");QueueingConsumer consumer = new QueueingConsumer(channel);channel.basicConsume(queueName, true, consumer);while (true) {QueueingConsumer.Delivery delivery = consumer.nextDelivery();String message = new String(delivery.getBody());String routingKey = delivery.getEnvelope().getRoutingKey();System.out.println(" [x] Received '" + routingKey + "':'" + message + "'");   }}
}

   在consumer端,会为每一个severity创建一个新的绑定,也就是

   for(String severity : argv)

       channel.queueBind(queueName, EXCHAGNE_NAME ,severity)

  参考 http://www.rabbitmq.com/tutorials/tutorial-four-java.html





这篇关于RabbitMQ routing模式的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/471650

相关文章

Springboot使用RabbitMQ实现关闭超时订单(示例详解)

《Springboot使用RabbitMQ实现关闭超时订单(示例详解)》介绍了如何在SpringBoot项目中使用RabbitMQ实现订单的延时处理和超时关闭,通过配置RabbitMQ的交换机、队列和... 目录1.maven中引入rabbitmq的依赖:2.application.yml中进行rabbit

SpringBoot整合Canal+RabbitMQ监听数据变更详解

《SpringBoot整合Canal+RabbitMQ监听数据变更详解》在现代分布式系统中,实时获取数据库的变更信息是一个常见的需求,本文将介绍SpringBoot如何通过整合Canal和Rabbit... 目录需求步骤环境搭建整合SpringBoot与Canal实现客户端Canal整合RabbitMQSp

在JS中的设计模式的单例模式、策略模式、代理模式、原型模式浅讲

1. 单例模式(Singleton Pattern) 确保一个类只有一个实例,并提供一个全局访问点。 示例代码: class Singleton {constructor() {if (Singleton.instance) {return Singleton.instance;}Singleton.instance = this;this.data = [];}addData(value)

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)、

模版方法模式template method

学习笔记,原文链接 https://refactoringguru.cn/design-patterns/template-method 超类中定义了一个算法的框架, 允许子类在不修改结构的情况下重写算法的特定步骤。 上层接口有默认实现的方法和子类需要自己实现的方法

【iOS】MVC模式

MVC模式 MVC模式MVC模式demo MVC模式 MVC模式全称为model(模型)view(视图)controller(控制器),他分为三个不同的层分别负责不同的职责。 View:该层用于存放视图,该层中我们可以对页面及控件进行布局。Model:模型一般都拥有很好的可复用性,在该层中,我们可以统一管理一些数据。Controlller:该层充当一个CPU的功能,即该应用程序

迭代器模式iterator

学习笔记,原文链接 https://refactoringguru.cn/design-patterns/iterator 不暴露集合底层表现形式 (列表、 栈和树等) 的情况下遍历集合中所有的元素

《x86汇编语言:从实模式到保护模式》视频来了

《x86汇编语言:从实模式到保护模式》视频来了 很多朋友留言,说我的专栏《x86汇编语言:从实模式到保护模式》写得很详细,还有的朋友希望我能写得更细,最好是覆盖全书的所有章节。 毕竟我不是作者,只有作者的解读才是最权威的。 当初我学习这本书的时候,只能靠自己摸索,网上搜不到什么好资源。 如果你正在学这本书或者汇编语言,那你有福气了。 本书作者李忠老师,以此书为蓝本,录制了全套视频。 试

利用命令模式构建高效的手游后端架构

在现代手游开发中,后端架构的设计对于支持高并发、快速迭代和复杂游戏逻辑至关重要。命令模式作为一种行为设计模式,可以有效地解耦请求的发起者与接收者,提升系统的可维护性和扩展性。本文将深入探讨如何利用命令模式构建一个强大且灵活的手游后端架构。 1. 命令模式的概念与优势 命令模式通过将请求封装为对象,使得请求的发起者和接收者之间的耦合度降低。这种模式的主要优势包括: 解耦请求发起者与处理者

RabbitMQ使用及与spring boot整合

1.MQ   消息队列(Message Queue,简称MQ)——应用程序和应用程序之间的通信方法   应用:不同进程Process/线程Thread之间通信   比较流行的中间件:     ActiveMQ     RabbitMQ(非常重量级,更适合于企业级的开发)     Kafka(高吞吐量的分布式发布订阅消息系统)     RocketMQ   在高并发、可靠性、成熟度等