分布式消息队列:Rabbitmq(2)

目录

一:交换机

1:Direct交换机

1.1生产者端代码:

 1.2:消费者端代码:

2:Topic主题交换机 

2.1:生产者代码: 

2.2:消费者代码:

 二:核心特性

2.1:消息过期机制

2.1.1:给队列中的全部消息指定过期时间

2.1.2:给某条消息指定过期时间 

2.2:死信队列

一:交换机

1:Direct交换机

绑定:让交换机和队列进行关联，可以指定让交换机把什么样的消息发送给队列。

rountingkey:路由键，控制消息要发送哪个队列。

特点:根据路由键指定要转发到指定的队列

场景:特定的消息指定给特定的队列

1.1生产者端代码:

我们规定，通过控制台输入消息和路由，来指定谁完成该任务。

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.util.Scanner;public class DirectProducer {private static final String EXCHANGE_NAME = "2";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {//创建交换机的名称channel.exchangeDeclare(EXCHANGE_NAME, "direct");Scanner scanner=new Scanner(System.in);while(scanner.hasNext()){String userInput=scanner.nextLine();String[] s = userInput.split(" ");if(s.length<1){continue;}//指定路由keyString message=s[0];String routingKey=s[1];//发布消息/*第一个参数:发布到哪个交换机第二个参数:路由键*/channel.basicPublish(EXCHANGE_NAME,routingKey,null,message.getBytes("UTF-8"));System.out.println("[x] Sent"+message+"with rounting"+routingKey+" ");}}}//..}

 1.2:消费者端代码:

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.util.Scanner;public class DirectProducer {private static final String EXCHANGE_NAME = "2";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {//创建交换机的名称channel.exchangeDeclare(EXCHANGE_NAME, "direct");Scanner scanner=new Scanner(System.in);while(scanner.hasNext()){String userInput=scanner.nextLine();String[] s = userInput.split(" ");if(s.length<1){continue;}//指定路由keyString message=s[0];String routingKey=s[1];//发布消息/*第一个参数:发布到哪个交换机第二个参数:路由键*/channel.basicPublish(EXCHANGE_NAME,routingKey,null,message.getBytes("UTF-8"));System.out.println("[x] Sent"+message+"with rounting"+routingKey+" ");}}}//..}

运行结果:

2:Topic主题交换机 

特点:消息会根据一个模糊的路由键转发到指定的队列中。

场景:特定的一类消息只交给特定的一类系统(程序来处理)。

绑定关系:模糊匹配消息队列  *:匹配一个单词       #:匹配0个或多个单词

2.1:生产者代码: 

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.util.Scanner;public class TopicProducer {private static final String EXCHANGE_NAME = "3";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.exchangeDeclare(EXCHANGE_NAME, "topic");Scanner scanner=new Scanner(System.in);while(scanner.hasNext()){String userInput=scanner.nextLine();String[] s = userInput.split(" ");if(s.length<1){continue;}//指定路由keyString message=s[0];String routingKey=s[1];//发布消息/*第一个参数:发布到哪个交换机第二个参数:路由键*/channel.basicPublish(EXCHANGE_NAME,routingKey,null,message.getBytes("UTF-8"));System.out.println("[x] Sent"+message+"with rounting"+routingKey+" ");}}}
}

2.2:消费者代码:

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;public class TopicConsumer {private static final String EXCHANGE_NAME = "3";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, "topic");//创建消息队列String queueName="fronted_queue";channel.queueDeclare(queueName,true,false,false,null);channel.queueBind(queueName,EXCHANGE_NAME,"#.前端.#");String queueName2="backed-_queue";channel.queueDeclare(queueName2,true,false,false,null);channel.queueBind(queueName2,EXCHANGE_NAME,"#.后端.#");String queueName3="product_queue";channel.queueDeclare(queueName3,true,false,false,null);channel.queueBind(queueName3,EXCHANGE_NAME,"#.产品.#");System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [前端] Received '" +delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [后端] Received '" +delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};DeliverCallback deliverCallback3 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [产品] Received '" +delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};channel.basicConsume(queueName, true, deliverCallback1, consumerTag -> { });channel.basicConsume(queueName2, true, deliverCallback2, consumerTag -> { });channel.basicConsume(queueName3, true, deliverCallback3, consumerTag -> { });}}

运行结果:

 二:核心特性

2.1:消息过期机制

特点:给每条消息指定一个有效期，一段时间内未被消费，就过期了。

2.1.1:给队列中的全部消息指定过期时间

在消费者中对于队列的全部消息指定过期时间，如果在过期时间内，还没有消费者取消息，消息才会过期，如果消息已经接收到，但是没确认，是不会过期的。

public class TTLConsumer {private final static String QUEUE_NAME = "ttl_queue";public static void main(String[] argv) throws Exception {//创建连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");//创建频道，提供通信Connection connection = factory.newConnection();Channel channel = connection.createChannel();//指定消息队列的过期时间Map<String ,Object> args=new HashMap<>();args.put("x-message-ttl",5000);//args:指定参数channel.queueDeclare(QUEUE_NAME, false, false,false, args);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");//如何处理消息DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");};channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });}
}

2.1.2:给某条消息指定过期时间 

//在发送者这边设置过期时间
public class TTLProducer {private final static String QUEUE_NAME = "ttl_queue";public static void main(String[] argv) throws Exception {//创建连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();//频道相当于客户端(jdbcClient,redisClient),提供了和消队列server建立通信，程序通过channel进行发送消息Channel channel = connection.createChannel()) {//创建消息队列，第二个参数(durable):是否开启持久化,第三个参数exclusiove:是否允许当前这个创建消息队列的//连接操作消息队列 第四个参数:没有人使用队列，是否需要删除String message = "Hello World!";//给消息指定过期时间AMQP.BasicProperties properties=new AMQP.BasicProperties.Builder().expiration("1000").build();channel.basicPublish("", QUEUE_NAME, properties, message.getBytes(StandardCharsets.UTF_8));System.out.println(" [x] Sent '" + message + "'");}}

2.2:死信队列

为了保证消息的可靠性，比如每条消息都成功消费，需要提供一个容错机制，即失败的消息怎么处理，相当于死信。

死信:过期的消息，拒收的消息，处理失败的消息，消息队列满了统称为死信。

死信队列:处理死信的队列。

死信交换机:给死信队列发送消息的交换机，也存在路由绑定。

a:创建死信交换机和死信队列

   //声明死信交换机channel.exchangeDeclare(WORK_NAME,"direct");//声明死信队列String queueName="boss_queue";channel.queueDeclare(queueName,true,false,false,null);channel.queueBind(queueName,EXCHANGE_Name,"boss");String queueName2="waibao_queue";channel.queueDeclare(queueName2, false, false, false, null);channel.queueBind(queueName2,EXCHANGE_Name,"waibao");

b:给失败后的需要容错的队列绑定死信交换机

  //声明交换机channel.exchangeDeclare(WORK_NAME, "direct");Map<String,Object> map=new HashMap<>();//声明要绑定的死信交换机map.put("x-dead-letter-exchange",DEAD_EXCHANGE_NAME);//声明要绑定的死信队列map.put("x-dead-letter-routing-key","waibao_queue");//创建消息队列String queueName="xiaodog_queue";channel.queueDeclare(queueName,true,false,false,map);channel.queueBind(queueName,WORK_NAME,"xiaodog");Map<String,Object> map2=new HashMap<>();//声明要绑定的死信交换机map2.put("x-dead-letter-exchange",DEAD_EXCHANGE_NAME);map2.put("x-dead-letter-routing-key","boss_queue");String queueName2="xiaocat_queue";channel.queueDeclare(queueName2,true,false,false,map2);channel.queueBind(queueName2,WORK_NAME,"xiaocat");

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

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

相关文章

elasticsearch-5.6.15集群部署，如何部署x-pack并添加安全认证

elasticsearch-5.6.15集群部署，如何部署x-pack并添加安全认证

目录 一、环境 1、JDK、映射、域名、三墙 2、三台服务器创建用户、并为用户授权 二、配置elasticsearch-5.6.15实例 1、官网获取elasticsearch-5.6.15.tar.gz&#xff0c;拉取到三台服务器 2、elas环境准备 3、修改elasticsearch.yml配置 4、修改软、硬件线程数 5、修改…
阅读更多...
GAMP源码阅读（中）伪距单点定位 SPP

GAMP源码阅读（中）伪距单点定位 SPP

原始 Markdown文档、Visio流程图、XMind思维导图见&#xff1a;https://github.com/LiZhengXiao99/Navigation-Learning 文章目录 一、SPP 解算1、spp()&#xff1a;单点定位主入口函数2、estpos()3、estpose_()4、valsol()&#xff1a;GDOP和卡方检验结果有效性 二、卫星位置钟…
阅读更多...
基于SSM的n省出口基地公共信息服务平台设计与实现

基于SSM的n省出口基地公共信息服务平台设计与实现

末尾获取源码 开发语言&#xff1a;Java Java开发工具&#xff1a;JDK1.8 后端框架&#xff1a;SSM 前端&#xff1a;采用JSP技术开发 数据库&#xff1a;MySQL5.7和Navicat管理工具结合 服务器&#xff1a;Tomcat8.5 开发软件&#xff1a;IDEA / Eclipse 是否Maven项目&#x…
阅读更多...
vue3-vite-ts-pinia

vue3-vite-ts-pinia

Vue3 vite Ts pinia 实战 源码 electron 仓库地址&#xff1a;https://gitee.com/szxio/vue3-vite-ts-pinia 视频地址&#xff1a;小满Vue3&#xff08;课程导读&#xff09;_哔哩哔哩_bilibili 课件地址&#xff1a;Vue3_小满zs的博客-CSDN博客 初始化Vue3项目 方式一 …
阅读更多...
【计算机网络笔记】DNS报文格式

【计算机网络笔记】DNS报文格式

DNS 提供域名到主机IP地址的映射  域名服务的三大要素&#xff1a;  域&#xff08;Domain&#xff09;和域名(Domain name)&#xff1a; 域指由地 理位置或业务类型而联系在一起的一组计算机构 成。  主机&#xff1a;由域名来标识。域名是由字符和&#xff08;或&a…
阅读更多...
【多线程面试题十】、说一说notify()、notifyAll()的区别

【多线程面试题十】、说一说notify()、notifyAll()的区别

文章底部有个人公众号&#xff1a;热爱技术的小郑。主要分享开发知识、学习资料、毕业设计指导等。有兴趣的可以关注一下。为何分享&#xff1f; 踩过的坑没必要让别人在再踩&#xff0c;自己复盘也能加深记忆。利己利人、所谓双赢。 面试官&#xff1a;说一说notify()、notify…
阅读更多...
pdf转jpg的方法【ps和工具方法】

pdf转jpg的方法【ps和工具方法】

pdf转jpg的方法&#xff1a; 1.photoshop办法&#xff1a; pdf直接拖入ps中&#xff0c;另存为*.Jpg文件即可 另外注意的时候&#xff0c;有时候别人给你pdf文件中包含你需要的jpg文件&#xff0c;千万不要截图进入ps中&#xff0c;直接把文件拖入ps中&#xff0c;这样的文件…
阅读更多...
皮卡丘RCE靶场通关攻略

皮卡丘RCE靶场通关攻略

皮卡丘RCE靶场通关攻略 文章目录 皮卡丘RCE靶场通关攻略RCE(remote command/code execute)概述远程系统命令执行启动环境漏洞练习第一关exec "ping"第二关 exec "eval" RCE(remote command/code execute)概述 RCE漏洞&#xff0c;可以让攻击者直接向后台服…
阅读更多...
el -table 多层级嵌套

el -table 多层级嵌套

只要你后端可以查到数据这个层级可以无限嵌套 这里用了懒加载&#xff0c;每次点击的时候将当前点击的父级id作为查询条件&#xff0c;向后端发送请求&#xff0c;来获取他子级的数据&#xff0c;并不是将所有数据查出来拼接返回的。 前端代码 <el-table:data"dataLis…
阅读更多...
基于Ubuntu20.04安装ROS系统

基于Ubuntu20.04安装ROS系统

文章目录 一、ROS简介二、ROS安装三、ROS安装测试四、安装问题解决1. sudo rosdepc init&#xff1a;找不到命令2. ERROR: cannot download default sources list from...3. Command roscore not found...4. Resource not found: roslaunch... 一、ROS简介 ROS是用于编写机器人…
阅读更多...
O(1) 时间插入、删除和获取随机元素

O(1) 时间插入、删除和获取随机元素

实现RandomizedSet 类&#xff1a; RandomizedSet() 初始化 RandomizedSet 对象 bool insert(int val) 当元素 val 不存在时&#xff0c;向集合中插入该项&#xff0c;并返回 true &#xff1b;否则&#xff0c;返回 false 。 bool remove(int val) 当元素 val 存在时&#xf…
阅读更多...
C# 递归算法使用简介_常用整理

C# 递归算法使用简介_常用整理

一、递归简介 递归算法是一种直接或者间接调用自身函数或者方法的算法。 递归算法的实质是把问题分解成规模缩小的同类问题的子问题&#xff0c;然后递归调用方法来表示问题的解。递归算法对解决一大类问题很有效&#xff0c;它可以使算法简洁和易于理解。 递归本质是循环&a…
阅读更多...
mysql如何查询数据出现的次数

mysql如何查询数据出现的次数

在mysql中&#xff0c;可以利用select语句配合group by和count查询数据出现的次数&#xff0c;count能够返回检索数据的数目&#xff0c;语法为“select 列名,count(*) as count from 表名 group by 列名”。 count函数是用来统计表中或数组中记录的一个函数&#xff0c;count…
阅读更多...
自定义控件的子控件布局（onLayout()方法）

自定义控件的子控件布局（onLayout()方法）

onLayout()方法用于指定布局中子控件的位置&#xff0c;该方法通常在自定义的ViewGroup容器中重写。 重写onLayout()方法中的常用方法&#xff1a; getChildCount() 获取子控件数量 getChildAt( int index ) 获取指定index的子控件&#xff0c;返回View view.getVisibilit…
阅读更多...
大语言模型系列

大语言模型系列

国产大模型开源一哥再登场&#xff0c;最强双语LLM「全家桶」级开源&#xff01;340亿参数超越Llama2-70B 为什么说大模型训练很难&#xff1f; - 知乎 GitHub - jeinlee1991/chinese-llm-benchmark: 中文大模型能力评测榜单&#xff1a;覆盖百度文心一言、chatgpt、阿里通义千…
阅读更多...
C#学习系列之继承

C#学习系列之继承

C#学习系列之继承 啰嗦继承使用特殊基类隐藏方法实际使用总结 啰嗦 基础学习。 继承 一个类派生于另一个基类型&#xff0c;它拥有该基础类型的所有成员字段和函数。A派生于B&#xff0c;继承A的所有东西&#xff0c;同时可以增加自己的东西。 使用 public class parent {p…
阅读更多...
bootstrap.yml文件未加载

bootstrap.yml文件未加载

springcloud2020.X.X版本官方重构了bootstrap引导配置的加载方式 需要添加以下依赖才能提高bootstrap.yml文件加载的优先级 <dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-bootstrap</artifactId&g…
阅读更多...
Spring-声明式事务

Spring-声明式事务

声明式事务 一、简介1、准备工作2、测试 二、声明式事务概念1、编程式事务2、声明式事务3、基于注解的声明式事务1.测试无事务情况2.加入事务①Transactional注解标识的位置②事务属性&#xff1a;只读③事务属性&#xff1a;超时④事务属性&#xff1a;回滚策略⑤事务属性&…
阅读更多...
《Effective C++》知识点（4）--设计与声明

《Effective C++》知识点（4）--设计与声明

18. 让接口容易被正确使用&#xff0c;不易被误用 18.1 好的接口很容易被正确使用&#xff0c;不容易被误用。你应该在你的所有接口中努力达成这些性质。任何接口如果要求客户必须记得做某些事情&#xff0c;就有着"不正确使用"的倾向。 18.2 "促进正确使用&quo…
阅读更多...
文心一言 VS 讯飞星火 VS chatgpt （123）-- 算法导论10.4 4题

文心一言 VS 讯飞星火 VS chatgpt （123）-- 算法导论10.4 4题

四、用go语言&#xff0c;对于一个含 n个结点的任意有根树&#xff0c;写出一个 O(n)时间的过程&#xff0c;输出其所有关键字该树以左孩子右兄弟表示法存储。 文心一言&#xff1a; 在计算机科学中&#xff0c;左孩子右兄弟表示法是一种用于表示树状结构的方法&#xff0c;其…
阅读更多...
最新文章

Copyright @ 2022~2023