分布式消息队列: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.rhkb.cn/news/176998.html

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

相关文章

macOS 创建Flutter项目

macOS 创建Flutter项目

参考在 macOS 上安装和配置 Flutter 开发环境 - Flutter 中文文档 - Flutter 中文开发者网站 - Flutter 这个文档&#xff0c;配置好flutter的环境 编辑器可以选择vscode或者IDEA。 我这里以IDEA为例 打开 IDE 并选中 New Flutter Project。 选择 Flutter&#xff0c;验证 F…
阅读更多...
云游长江大桥，3DCAT实时云渲染助力打造沉浸化数字文旅平台

云游长江大桥，3DCAT实时云渲染助力打造沉浸化数字文旅平台

南京长江大桥是中国第一座自主设计建造的双层公路铁路桥&#xff0c;也是世界上最早的双层公路铁路桥之一。它不仅是一座桥梁&#xff0c;更是一座历史文化的见证者和传承者。它见证了中国人民的智慧和奋斗&#xff0c;承载了中国社会的变迁和发展。 如何让这座不可移动的文物…
阅读更多...
FreeRTOS_信号量之互斥信号量

FreeRTOS_信号量之互斥信号量

目录 1. 互斥信号量 1.1 互斥信号量简介 1.2 创建互斥信号量 1.2.1 函数 xSemaphoreCreateMutex() 1.2.2 函数 xSemaphoreCreateMutexStatic() 1.2.3 互斥信号量创建过程分析 1.2.4 释放互斥信号量 1.2.5 获取互斥信号量 2. 互斥信号量操作实验 2.1 实验程序 2.1.1 …
阅读更多...
Linux：文件操作

Linux：文件操作

目录 一、关于文件 1、文件类的系统接口 2、文件的含义 二、文件操作 1、C语言文件相关接口 2、系统接口 open close write read 三、文件描述符 关于fd fd的分配规则 输出重定向示例 输入重定向示例 追加重定向示例 dup2函数 缓冲区 stdout与stderr perror…
阅读更多...
Webpack常见的插件和模式

Webpack常见的插件和模式

文章目录 一、认识插件Plugin1.认识Plugin 二、CleanWebpackPlugin三、HtmlWebpackPlugin1.生成index.html分析2.自定义HTML模板3.自定义模板数据填充 四、DefinePlugin1.DefinePlugin的介绍2.DefinePlugin的使用 五、Mode配置 一、认识插件Plugin 1.认识Plugin Webpack的另一…
阅读更多...
数据结构：算法（特性，时间复杂度，空间复杂度）

数据结构：算法（特性，时间复杂度，空间复杂度）

目录 1.算法的概念2.算法的特性1.有穷性2.确定性3.可行性4.输入5.输出 3.好算法的特质1.正确性2.可读性3.健壮性4.高效率与低存储需求 4.算法的时间复杂度1.事后统计的问题2.复杂度表示的计算1.加法规则2.乘法规则3.常见函数数量级比较 5.算法的空间复杂度1.程序的内存需求2.例…
阅读更多...
5.3 用户定义的完整性

5.3 用户定义的完整性

思维导图&#xff1a; 5.3 用户定义的完整性 用户定义的完整性是确保数据库中的数据满足特定应用的语义要求。这是通过关系数据库管理系统(RDBMS)中的内置机制来完成的&#xff0c;不需要依赖应用程序来执行。 5.3.1 属性上的约束条件 1. 定义属性上的约束条件 当在CREATE T…
阅读更多...
centos7安装oxidized备份软件

centos7安装oxidized备份软件

首先需要提前下载ruby&#xff0c;因为默认yum安装的版本太低 https://cache.ruby-lang.org/pub/ruby/3.1/ruby-3.1.0.tar.gz 1、yum remove ruby ruby-devel&#xff08;有就卸载&#xff0c;没有则忽略&#xff09; 2、将下载好的ruby包解压到/opt下 [rootoxidized ruby-…
阅读更多...
Python---字符串输入和输出---input()、格式化输出：%，f形式，format形式

Python---字符串输入和输出---input()、格式化输出：%，f形式，format形式

字符串输入: 在Python代码中&#xff0c;我们可以使用input()方法来接收用户的输入信息。记住&#xff1a;在Python中&#xff0c;input()方法返回的结果是一个字符串类型的数据。 如果之后使用输入的数据&#xff0c;一定要记得利用数据类型转换。 相关链接&#xff1a;Pyt…
阅读更多...
nodejs升级或降级

nodejs升级或降级

node有一个模块叫n&#xff0c;是专门用来管理node.js的版本。 升级或降级步骤 1 、安装n模块 npm install -g n 2、 升级node.js到最新稳定版 n stable Ps: n后面也可以跟随版本号&#xff08;用于升级或降级&#xff09;比如&#xff1a; n v16.12.0
阅读更多...
蚂蚁SOFA Stack融合大模型发布升级版 将为机构产研效能提升30%

蚂蚁SOFA Stack融合大模型发布升级版 将为机构产研效能提升30%

11月1日&#xff0c;在云栖大会上&#xff0c;蚂蚁集团正式发布CodeFuse全面加持的SOFAStack5.0升级版本&#xff0c;向企业提供全方位研发运维智能助手相关能力。这是继蚂蚁集团在外滩大会发布代码大模型CodeFuse之后&#xff0c;首次公布面向行业的商业化产品进展。 “大模型…
阅读更多...
Controllable Guide-Space for Generalizable Face Forgery Detection

Controllable Guide-Space for Generalizable Face Forgery Detection

一、研究背景 以往工作专注于提取伪造特征的共同特性和真假域鉴别性信息&#xff0c;以提升特征泛化性。 但在训练过程中&#xff0c;这些方法只区分真假域&#xff0c;并将不同的伪造域看作一类而不加以区分。 这会导致伪造样本进一步以伪造不相关特征&#xff08;如&#xff…
阅读更多...
ACID模型

ACID模型

ACID 是数据库管理系统&#xff08;DBMS&#xff09;中用来确保事务处理正确性和可靠性的四个特性的首字母缩写。ACID 是指原子性&#xff08;Atomicity&#xff09;、一致性&#xff08;Consistency&#xff09;、隔离性&#xff08;Isolation&#xff09;和持久性&#xff08…
阅读更多...
如何通过 NAT 模式连接VMware虚拟机以及存在和不存在ens33文件的解决方案

如何通过 NAT 模式连接VMware虚拟机以及存在和不存在ens33文件的解决方案

文章目录 前言1 VMware配置1.1 打开vmvare虚拟网络编辑器1.2 取消使用本地DHCP1.3 NAT设置 2 虚拟机的配置2.1 存在ens332.2.1 修改ifcfg-ens33文件2.2.1.1 为什么设置BOOTPROTOstatic&#xff1f;2.2.1.2 如何选择使用static还是dhcp&#xff1f; 2.2.2 关闭防火墙 2.2 不存在…
阅读更多...
VScode clangd 插件浏览 linux 源码

VScode clangd 插件浏览 linux 源码

文章目录 VScode clangd 插件浏览 linux 源码clangd 安装与配置VScode 插件安装clangd 安装方法一方法二 clangd 配置 cmake 生成bear 生成 compile_commands.json触发 clangd linux 内核脚本生成 compile_commands.json 文件三种方式对比 VScode clangd 插件浏览 linux 源码 …
阅读更多...
抖音协议算法最新版

抖音协议算法最新版

抖音的协议算法是指用于推荐内容和个性化用户体验的算法系统。这些算法根据用户的兴趣、行为和偏好来推荐适合他们的视频内容&#xff0c;以提供更好的用户体验。 抖音的协议算法使用了大量的数据和机器学习技术来实现个性化推荐。以下是一些可能应用于抖音协议算法的技术和方法…
阅读更多...
使用Objective-C和ASIHTTPRequest库进行Douban电影分析

使用Objective-C和ASIHTTPRequest库进行Douban电影分析

概述 Douban是一个提供图书、音乐、电影等文化内容的社交网站&#xff0c;它的电影频道包含了大量的电影信息和用户评价。本文将介绍如何使用Objective-C语言和ASIHTTPRequest库进行Douban电影分析&#xff0c;包括如何获取电影数据、如何解析JSON格式的数据、如何使用代理IP技…
阅读更多...
机器学习-特征工程

机器学习-特征工程

一、特征工程介绍 1.1 什么是特征 数值特征&#xff08;连续特征&#xff09;、文本特征&#xff08;离散特征&#xff09; 1.2 特征的种类 1.3 特征工程 特征是机器学习可疑直接使用的&#xff0c;模型和特征之间是一个循环过程&#xff1b; 实际上特征工程就是将原始数据…
阅读更多...
3 Tensorflow构建模型详解

3 Tensorflow构建模型详解

上一篇&#xff1a;2 用TensorFlow构建一个简单的神经网络-CSDN博客 本篇目标是介绍如何构建一个简单的线性回归模型&#xff0c;要点如下&#xff1a; 了解神经网络原理构建模型的一般步骤模型重要参数介绍 1、神经网络概念 接上一篇&#xff0c;用tensorflow写了一个猜测西…
阅读更多...
微信小程序：自定义组件传值——获取手机验证码

微信小程序：自定义组件传值——获取手机验证码

一&#xff1a;遇到的问题 通过自己自定义的组件编写的表单&#xff0c;发现传值不了&#xff0c;点击后收到的值为空。 二&#xff1a;创建组件 先在根目录创建components文件夹&#xff0c;创建img-verify文件夹&#xff08;这个是我取的组件名字&#xff09;&#xff0c;在…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023