分布式消息队列: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;
                    }
                    //指定路由key
                    String 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;
                    }
                    //指定路由key
                    String 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;
                }
                //指定路由key
                String 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");

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

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

零信任安全模型和多因素身份验证：提升网络安全的关键一步

零信任安全模型和多因素身份验证：提升网络安全的关键一步

近年来&#xff0c;随着疫情的蔓延和科技的飞速发展&#xff0c;数据和工作的数字化程度前所未有。这虽然为机会创造提供了更多空间&#xff0c;但也为潜在威胁行为者提供了新的入侵途径。因此&#xff0c;数据泄露的防范已经成为每个组织IT基础设施中不可或缺的一部分。 数据泄…
阅读更多...
远程IO在激光行业：实现高效、精准控制的解决方案

远程IO在激光行业：实现高效、精准控制的解决方案

激光机简介 激光机是激光雕刻机、激光切割机和激光打标机的总称。激光机利用其高温的工作原理作用于被加工材料表面&#xff0c;同时根据输入到机器内部的图形&#xff0c;绘制出客户要求的图案、文字等。激光机根据用途可分为激光切割机和激光雕刻机。其中&#xff0c;激光切割…
阅读更多...
liunx练习题之在同一主机提供多个的web服务

liunx练习题之在同一主机提供多个的web服务

虚拟web主机类型 一、基于端口 1.vim /etc/httpd/conf.d/vhost2.conf ---- — 改变http服务默认访问路径 <directory /testweb1>allowoverride none 表示不允许覆盖其他配置require all granted 表示允许所有请求 </directory> <virtualhost 0.0.0.0:…
阅读更多...
NVME CMB原理和常规使用方案详解

NVME CMB原理和常规使用方案详解

什么是CMB 在NVMe Express 1.2 Spec中开始支持一个特性&#xff0c;那就是CMB&#xff08;Controller Memory Buffer&#xff09;&#xff0c;是指SSD控制器内部的读写存储缓冲区&#xff0c;与HMB&#xff08;Host Memory Buffer&#xff09;的不同处在于所使用的内存地址位于…
阅读更多...
回归预测 | Matlab实现RIME-CNN-SVM霜冰优化算法优化卷积神经网络-支持向量机的多变量回归预测

回归预测 | Matlab实现RIME-CNN-SVM霜冰优化算法优化卷积神经网络-支持向量机的多变量回归预测

回归预测 | Matlab实现RIME-CNN-SVM霜冰优化算法优化卷积神经网络-支持向量机的多变量回归预测 目录 回归预测 | Matlab实现RIME-CNN-SVM霜冰优化算法优化卷积神经网络-支持向量机的多变量回归预测效果一览基本介绍程序设计参考资料 效果一览 基本介绍 1.RIME-CNN-SVM霜冰优化算…
阅读更多...
【Linux】深入理解系统文件操作（1w字超详解）

【Linux】深入理解系统文件操作（1w字超详解）

1.系统下的文件操作&#xff1a; ❓是不是只有C\C有文件操作呢&#xff1f;&#x1f4a1;Python、Java、PHP、go也有&#xff0c;他们的文件操作的方法是不一样的啊 1.1对于文件操作的思考&#xff1a; 我们之前就说过了&#xff1a;文件内容属性 针对文件的操作就变成了对…
阅读更多...
轧钢测径仪在螺纹钢负公差轧制中的四大作用！

轧钢测径仪在螺纹钢负公差轧制中的四大作用！

螺纹钢为什么要进行负公差轧制&#xff1f; 在标准允许范围内&#xff0c;越接近负公差&#xff0c;那么在合格规范内&#xff0c;所损耗的原材料越少&#xff0c;而螺纹钢轧制速度快&#xff0c;更是以吨的量进行成交&#xff0c;因此控制的原材料积少成多&#xff0c;对其成本…
阅读更多...
因存在色情内容，夸克被罚50万元

因存在色情内容，夸克被罚50万元

媒体经济的繁荣、自媒体、直播等各种形式的信息传播疯狂发展&#xff0c;但是各种形式的信息资源大规模生产时&#xff0c;“色情”&#xff0c;“暴力”的图像和视频不可控的滋生&#xff0c;特别是某些 APP 或浏览器。一旦打开&#xff0c;满屏都是“哥哥&#xff0c;快来啊”…
阅读更多...
java判断上午下午的方法

java判断上午下午的方法

一、代码 import java.text.SimpleDateFormat; import java.util.Date; import java.util.Locale;public class Main1 {public static void main(String[] args) {Date nowDate new Date();SimpleDateFormat sdf new SimpleDateFormat("aa", Locale.ENGLISH);Strin…
阅读更多...
AI大模型在短视频处理和剪辑中的应用，文末送书

AI大模型在短视频处理和剪辑中的应用，文末送书

&#x1f3c6;作者简介&#xff0c;黑夜开发者&#xff0c;CSDN领军人物&#xff0c;全栈领域优质创作者✌&#xff0c;CSDN博客专家&#xff0c;阿里云社区专家博主&#xff0c;2023年6月CSDN上海赛道top4。 &#x1f3c6;数年电商行业从业经验&#xff0c;历任核心研发工程师…
阅读更多...
矢量图形编辑软件illustrator 2023 mac中文软件特点

矢量图形编辑软件illustrator 2023 mac中文软件特点

illustrator 2023 mac是一款矢量图形编辑软件&#xff0c;用于创建和编辑排版、图标、标志、插图和其他类型的矢量图形。 illustrator 2023 mac软件特点 矢量图形&#xff1a;illustrator创建的图形是矢量图形&#xff0c;可以无限放大而不失真&#xff0c;这与像素图形编辑软…
阅读更多...
【Jenkins】新建任务FAQ

【Jenkins】新建任务FAQ

问题1. 源码管理处填入Repository URL&#xff0c;报错&#xff1a;无法连接仓库&#xff1a;Error performing git command: ls-remote -h https://github.com/txy2023/GolangLearning.git HEAD 原因&#xff1a; jenkins全局工具配置里默认没有添加git的路径&#xff0c;如果…
阅读更多...
全方位 Linux 性能调优经验总结

全方位 Linux 性能调优经验总结

Part1Linux性能优化 1性能优化 性能指标 高并发和响应快对应着性能优化的两个核心指标&#xff1a;吞吐和延时 图片来自: www.ctq6.cn 应用负载角度&#xff1a;直接影响了产品终端的用户体验系统资源角度&#xff1a;资源使用率、饱和度等 性能问题的本质就是系统资源已经…
阅读更多...
【排序算法】 归并排序详解！深入理解！思想+实现！

【排序算法】 归并排序详解！深入理解！思想+实现！

&#x1f3a5; 屿小夏 &#xff1a; 个人主页 &#x1f525;个人专栏 &#xff1a; 算法—排序篇 &#x1f304; 莫道桑榆晚&#xff0c;为霞尚满天&#xff01; 文章目录 &#x1f4d1;前言&#x1f324;️归并排序的思想☁️基本思想☁️归并的思想实现☁️分治法 &#x1f3…
阅读更多...
java 版本企业招标投标管理系统源码+多个行业+tbms+及时准确+全程电子化

java 版本企业招标投标管理系统源码+多个行业+tbms+及时准确+全程电子化

项目说明 随着公司的快速发展&#xff0c;企业人员和经营规模不断壮大&#xff0c;公司对内部招采管理的提升提出了更高的要求。在企业里建立一个公平、公开、公正的采购环境&#xff0c;最大限度控制采购成本至关重要。符合国家电子招投标法律法规及相关规范&#xff0c;以及审…
阅读更多...
一文彻底理解python浅拷贝和深拷贝

一文彻底理解python浅拷贝和深拷贝

目录 一、必备知识二、基本概念三、列表&#xff0c;元组&#xff0c;集合&#xff0c;字符串&#xff0c;字典浅拷贝3.1 列表3.2 元组3.3 集合3.4 字符串3.5 字典3.6 特别注意可视化展示浅拷贝总结 四、列表&#xff0c;元组&#xff0c;集合&#xff0c;字符串&#xff0c;字…
阅读更多...
Python+Robotframework实现http接口自动化测试详解

Python+Robotframework实现http接口自动化测试详解

前言 下周即将展开一个http接口测试的需求&#xff0c;刚刚完成的java类接口测试工作中&#xff0c;由于之前犯懒&#xff0c;没有提前搭建好自动化回归测试框架&#xff0c;以至于后期rd每修改一个bug&#xff0c;经常导致之前没有问题的case又产生了bug&#xff0c;所以需要…
阅读更多...
LED数码管的静态显示与动态显示（Keil+Proteus）

LED数码管的静态显示与动态显示（Keil+Proteus）

前言 就是今天看了一下书上的单片机实验&#xff0c;发现很多的器件在Proteus中都不知道怎么去查找&#xff0c;然后想做一下这个实验&#xff0c;尝试能不能实现&#xff0c;LED数码管的两个还可以实现&#xff0c;但是用LED点阵显示器的时候他那个网络标号不知道是什么情况&…
阅读更多...
openGauss学习笔记-111 openGauss 数据库管理-管理用户及权限-用户权限设置

openGauss学习笔记-111 openGauss 数据库管理-管理用户及权限-用户权限设置

文章目录 openGauss学习笔记-111 openGauss 数据库管理-管理用户及权限-用户权限设置111.1 给用户直接授予某对象的权限111.2 给用户指定角色111.3 回收用户权限 openGauss学习笔记-111 openGauss 数据库管理-管理用户及权限-用户权限设置 111.1 给用户直接授予某对象的权限 …
阅读更多...
【uniapp】uniapp实现input输入显示数字键盘：

【uniapp】uniapp实现input输入显示数字键盘：

文章目录 一、官网文档&#xff1a;二、文档&#xff1a;三、效果&#xff1a;【1】number&#xff1a;iPhone原生数字键盘不显示小数点【2】digit&#xff1a;【3】digit和inputmode&#xff1a; 一、官网文档&#xff1a; input | uni-app官网 二、文档&#xff1a; 三、效果…
阅读更多...
最新文章

Copyright @ 2022~2023