rabbitmq direct 多个消费者_一文解析 RabbitMQ 最常用的三大模式

326b7a263b51ee38b3ad551ea36120ce.png

Direct 模式

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

ad388b7026de605ce5d8e21b2535345c.png
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class DirectProducer {public static void main(String[] args) throws Exception {//1. 创建一个 ConnectionFactory 并进行设置ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");//2. 通过连接工厂来创建连接Connection connection = factory.newConnection();//3. 通过 Connection 来创建 ChannelChannel channel = connection.createChannel();//4. 声明String exchangeName = "test_direct_exchange";String routingKey = "item.direct";//5. 发送String msg = "this is direct msg";channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());System.out.println("Send message : " + msg);//6. 关闭连接channel.close();connection.close();}
}
import com.rabbitmq.client.*;
import java.io.IOException;public class DirectConsumer {public static void main(String[] args) throws Exception {//1. 创建一个 ConnectionFactory 并进行设置ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");factory.setAutomaticRecoveryEnabled(true);factory.setNetworkRecoveryInterval(3000);//2. 通过连接工厂来创建连接Connection connection = factory.newConnection();//3. 通过 Connection 来创建 ChannelChannel channel = connection.createChannel();//4. 声明String exchangeName = "test_direct_exchange";String queueName = "test_direct_queue";String routingKey = "item.direct";channel.exchangeDeclare(exchangeName, "direct", true, false, null);channel.queueDeclare(queueName, false, false, false, null);//一般不用代码绑定,在管理界面手动绑定channel.queueBind(queueName, exchangeName, routingKey);//5. 创建消费者并接收消息Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body)throws IOException {String message = new String(body, "UTF-8");System.out.println(" [x] Received '" + message + "'");}};//6. 设置 Channel 消费者绑定队列channel.basicConsume(queueName, true, consumer);}
}
Send message : this is direct msg[x] Received 'this is direct msg'
Topic 模式

Topic 模式

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

  • 符号'#" 匹配一个或多个词
  • 符号"*”匹配不多不少一个词

例如:

  • 'log.#"能够匹配到'log.info.oa"
  • "log.*"只会匹配到"log.erro“

b39cb721a3040fd71aac40aee017256e.png
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class TopicProducer {public static void main(String[] args) throws Exception {//1. 创建一个 ConnectionFactory 并进行设置ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");//2. 通过连接工厂来创建连接Connection connection = factory.newConnection();//3. 通过 Connection 来创建 ChannelChannel channel = connection.createChannel();//4. 声明String exchangeName = "test_topic_exchange";String routingKey1 = "item.update";String routingKey2 = "item.delete";String routingKey3 = "user.add";//5. 发送String msg = "this is topic msg";channel.basicPublish(exchangeName, routingKey1, null, msg.getBytes());channel.basicPublish(exchangeName, routingKey2, null, msg.getBytes());channel.basicPublish(exchangeName, routingKey3, null, msg.getBytes());System.out.println("Send message : " + msg);//6. 关闭连接channel.close();connection.close();}
}
import com.rabbitmq.client.*;
import java.io.IOException;public class TopicConsumer {public static void main(String[] args) throws Exception {//1. 创建一个 ConnectionFactory 并进行设置ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");factory.setAutomaticRecoveryEnabled(true);factory.setNetworkRecoveryInterval(3000);//2. 通过连接工厂来创建连接Connection connection = factory.newConnection();//3. 通过 Connection 来创建 ChannelChannel channel = connection.createChannel();//4. 声明String exchangeName = "test_topic_exchange";String queueName = "test_topic_queue";String routingKey = "item.#";channel.exchangeDeclare(exchangeName, "topic", true, false, null);channel.queueDeclare(queueName, false, false, false, null);//一般不用代码绑定,在管理界面手动绑定channel.queueBind(queueName, exchangeName, routingKey);//5. 创建消费者并接收消息Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body)throws IOException {String message = new String(body, "UTF-8");System.out.println(" [x] Received '" + message + "'");}};//6. 设置 Channel 消费者绑定队列channel.basicConsume(queueName, true, consumer);}
}
Send message : this is topc msg[x] Received 'this is topc msg'
[x] Received 'this is topc msg'

Fanout 模式

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

95566e06362d14e2aa6f704419bb141b.png
import com.rabbitmq.client.*;
import java.io.IOException;public class FanoutConsumer {public static void main(String[] args) throws Exception {//1. 创建一个 ConnectionFactory 并进行设置ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");factory.setAutomaticRecoveryEnabled(true);factory.setNetworkRecoveryInterval(3000);//2. 通过连接工厂来创建连接Connection connection = factory.newConnection();//3. 通过 Connection 来创建 ChannelChannel channel = connection.createChannel();//4. 声明String exchangeName = "test_fanout_exchange";String queueName = "test_fanout_queue";String routingKey = "item.#";channel.exchangeDeclare(exchangeName, "fanout", true, false, null);channel.queueDeclare(queueName, false, false, false, null);//一般不用代码绑定,在管理界面手动绑定channel.queueBind(queueName, exchangeName, routingKey);//5. 创建消费者并接收消息Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body)throws IOException {String message = new String(body, "UTF-8");System.out.println(" [x] Received '" + message + "'");}};//6. 设置 Channel 消费者绑定队列channel.basicConsume(queueName, true, consumer);}
}
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class FanoutProducer {public static void main(String[] args) throws Exception {//1. 创建一个 ConnectionFactory 并进行设置ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");//2. 通过连接工厂来创建连接Connection connection = factory.newConnection();//3. 通过 Connection 来创建 ChannelChannel channel = connection.createChannel();//4. 声明String exchangeName = "test_fanout_exchange";String routingKey1 = "item.update";String routingKey2 = "";String routingKey3 = "ookjkjjkhjhk";//任意routingkey//5. 发送String msg = "this is fanout msg";channel.basicPublish(exchangeName, routingKey1, null, msg.getBytes());channel.basicPublish(exchangeName, routingKey2, null, msg.getBytes());channel.basicPublish(exchangeName, routingKey3, null, msg.getBytes());System.out.println("Send message : " + msg);//6. 关闭连接channel.close();connection.close();}
}
Send message : this is fanout msg[x] Received 'this is fanout msg'
[x] Received 'this is fanout msg'
[x] Received 'this is fanout msg'

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/454213.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

程序员成长最快的环境

除开五大或者ThoughtWorks这种要什么有什么,进去做打字也能光耀门楣的不谈。如果是嫁到一个普通软件公司,怎样的环境才能最快的成长呢?首先基本的 公司项目管理水平是必要的;其次是稳健而不保守的公司技术选型和一班能沟通的同事。…

【BZOJ4254】Aerial Tramway 树形DP

【BZOJ4254】Aerial Tramway 题意&#xff1a;给你一座山上n点的坐标&#xff0c;让你在山里建m条缆车&#xff0c;要求缆车两端的高度必须相等&#xff0c;且中间经过的点的高度都小于缆车的高度。并且不能存在一个点位于至少k条缆车的下方。求缆车的最大总长度。 n,m<200,…

C# 读取保存App.config配置文件的完整源码参考

最近出差在北京做一个小项目&#xff0c;项目里需要读取配置文件的小功能&#xff0c;觉得挺有参考意义的就把代码发上来给大家参考一下。我们选择了直接用微软的读取配置文件的方法。 这个是程序的运行设计效果&#xff0c;就是把这些参数可以进行灵活设置&#xff0c;灵活保存…

TensorFlow 简介

TensorFlow介绍 Tagline&#xff1a;An open-source software library for Machine Intelligence.Definition&#xff1a;TensorFlow TM is an open source software library fornumerical computation using data flow graphs.GitHub&#xff1a;https://github.com/tensorfl…

webbrowser设置为相应的IE版本

注册表路径&#xff1a; HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\Microsoft\Internet Explorer\Main\FeatureControl\FEATURE_BROWSER_EMULATION 或者HKEY_LOCAL_MACHINE\SOFTWARE\Microsoft\Internet Explorer\Main\FeatureControl\FEATURE_BROWSER_EMULATION 究竟选择哪一个…

jmeter压力测试_用Jmeter实现对接口的压力测试

一、多个真实用户对接口的压力测试1. 获取多个真实用户的token的两种方法&#xff1a;1)第一种&#xff1a;让开发帮忙生成多个token(多个用户账户生成的token)&#xff0c;导出为csv格式的文件(以下步骤均以该方法为基础)2)第二种&#xff1a;自己设置多个用户账户和密码&…

程序员成长之路(转)

什么时候才能成为一个专业程序员呢&#xff1f;三年还是五年工作经验&#xff1f;其实不用的&#xff0c;你马上就可以了&#xff0c;我没有骗你&#xff0c;因为专业程序员与业余程序员的区别主要在于一种态度&#xff0c;如果缺乏这种态度&#xff0c;拥有十年工作经验也还是…

嵌入式开发——PWM高级定时器

学习目标 加强掌握PWM开发流程理解定时器与通道的关系掌握多通道配置策略掌握互补PWM配置策略掌握定时器查询方式掌握代码抽取优化策略掌握PWM调试方式学习内容 需求 点亮8个灯,采用pwm的方式。 定时器 通道 <

解决虚拟机时间引起的奇怪问题

一直使用得好好的虚拟机最近出了一个奇怪问题在虚拟机装好的lamp在客户端访问phpmyadmin的时候,使用firefox登录没问题,但是使用IE不行总是停留在登录的界面,而且没有提供任何的出错信息,就连在apache的日志里面也看不到.注意到同样访问的时候,在IE上显示的转向的url是[url]htt…

TensorFlow 基本操作

Tensorflow基本概念 图(Graph):图描述了计算的过程&#xff0c;TensorFlow使用图来表示计算任务。张量(Tensor):TensorFlow使用tensor表示数据。每个Tensor是一个类型化的多维数组。操作(op):图中的节点被称为op(opearation的缩写)&#xff0c;一个op获得/输入0个或多个Tensor…

03_zookeeper伪集群安装

一句话说明白&#xff1a;在1台机器上模拟多台机器&#xff0c;对外提供服务 在理解zookeeper集群安装方法的基础上&#xff0c;本文描述如何将1个机器模拟为3个节点的zookeeper集群&#xff0c;建议先参考阅读本文的前一期 zookeeper伪集群安装总结 在本机上通过复制的方式&am…

python合成语音_MicroPython动手做(25)——语音合成与语音识别

6、AB按键切换语言合成项目[mw_shl_codepython,true]#MicroPython动手做(25)——语音合成与语音识别#AB按键切换语言合成项目from mpython import *import networkimport timeimport ntptimefrom xunfei import *import audiomy_wifi wifi()my_wifi.connectWiFi("zh"…

专访谷歌CEO:像对待家人一样对待员工

导语&#xff1a;《财富》近日公布了“2012年度美国100家最适宜工作的公司”榜单&#xff0c;谷歌当选冠军。即将于2月6日出版的美国《财富》杂志印刷版将刊登对谷歌CEO拉里佩奇(Larry Page)的专访&#xff0c;对谷歌的工作环境进行了介绍。 以下为采访概要&#xff1a; 问&a…

TensorFlow 分布式

一、简介 使用单台机器或者单个GPU/CPU来进行模型训练&#xff0c;训练速度会受资源的影响&#xff0c;因为毕竟单个的设备的计算能力和存储能力具有一定的上限的&#xff0c;针对这个问题&#xff0c;TensorFlow支持分布式模型运算&#xff0c;支持多机器、多GPU、多CPU各种模…

第五周测试

---恢复内容开始--- 一 视频知识 1 linux系统下如何区分内核态与用户态 在内核态&#xff1a;cs:eip可以是任意的地址&#xff0c;4G的内存地址空间 在用户态&#xff1a;cs:eip只能访问0x00000000—0xbfffffff的地址空间 2 系统调用的三层皮&#xff1a;xyz、system_call和sys…

网页制作小技巧:dl dt dd标签用法

< DOCTYPE html PUBLIC -WCDTD XHTML StrictEN httpwwwworgTRxhtmlDTDxhtml-strictdtd> 一般我们在做列表的时候通常只会用到ul和li,至于DL一般都很少用到&#xff0c;它也属于列表类的标签&#xff0c;下面说一下大概的用法&#xff1a; <dl>标记定义了一个定义列…

latex公式对齐_Word 写公式最方便的方法

自从用上了word 2016之后&#xff0c;发现他的公式编辑器真香!真香!!他有了latex的优雅&#xff0c;又有了Mathtype的可视化效果&#xff0c;甚至更好哈&#xff0c;当编辑大量公式时也不会因为插件问题卡掉当前的努力。学起来也不复杂&#xff0c;反正是word. 强烈推荐。我们最…

路要怎么走?关于程序员成长的一点思考

程序员的我们&#xff0c;是否想过今后的路该怎么走、如何发展、技术怎样提高?其实这也是我一直在思考的问题。下面就此问题&#xff0c;分享下我的看法。因为我阅历有限&#xff0c;有什么说的不对的&#xff0c;大家见谅&#xff0c;千万不要喷…… 一、程序员应该打好基础 …

TensorFlow 常见API

数据类型转换相关API Tensor Shape获取以及设置相关API Tensor合并、分割相关API Error相关类API 常量类型的Tensor对象相关API 序列和随机Tensor对象相关API Session相关API 逻辑运算符相关API 比较运算符相关API 调试相关API 图像处理-编码解码相关API 图像处理-调整大小相关…

python封装继承多态_浅谈JavaScript的面向对象和它的封装、继承、多态

写在前面既然是浅谈&#xff0c;就不会从原理上深度分析&#xff0c;只是帮助我们更好地理解...面向对象与面向过程面向对象和面向过程是两种不同的编程思想&#xff0c;刚开始接触编程的时候&#xff0c;我们大都是从面向过程起步的&#xff0c;毕竟像我一样&#xff0c;大家接…