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

news2024/9/23 13:27:19

目录

一:交换机

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");

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

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

相关文章

Gloss优化

Gloss优化

Gloss优化&#xff0c;Route – Gloss – Parameters .清除不必要的线和过孔&#xff0c;圆滑线&#xff0c;焊盘中间的线&#xff0c;把转角变成圆弧&#xff0c;自动布线总会产生一些布线效果不好、多余过孔等问题。此时可以利用allegro提供的Gloss命令对设计进行优化和调整&…
阅读更多...
Facebook广告效果数据获取

Facebook广告效果数据获取

一、背景 公司每年在Facebook和Google上投放了大量的广告&#xff0c;我总不能让老板登录Facebook广告投放平台上去看广告效果&#xff0c;其实老板只关注每天花了多少钱引来了多少客户&#xff0c;每个客户平均花费多少钱&#xff0c;其它的他才不关心&#xff0c;有Facebook…
阅读更多...
工作小计-GPU硬编以及依赖库 nvcuvidnvidia-encode

工作小计-GPU硬编以及依赖库 nvcuvidnvidia-encode

工作小计-GPU编码以及依赖库 已经是第三篇关于编解码的记录了。项目中用到GPU编码很久了&#xff0c;因为yuv太大&#xff0c;所以编码显得很重要。这次遇到的问题是环境的搭建问题。需要把开发机上的环境放到docker中&#xff0c;以保证docker中同样可以进行GPU的编码。 1 定…
阅读更多...
SpringCloud 微服务全栈体系（五）

SpringCloud 微服务全栈体系（五）

第七章 Feign 远程调用 先来看我们以前利用 RestTemplate 发起远程调用的代码&#xff1a; 存在下面的问题&#xff1a; 代码可读性差&#xff0c;编程体验不统一 参数复杂 URL 难以维护 Feign 是一个声明式的 http 客户端&#xff0c;官方地址&#xff1a;https://github.…
阅读更多...
国家数据局正式揭牌，数据专业融合型人才迎来发展良机

国家数据局正式揭牌，数据专业融合型人才迎来发展良机

文章目录 每日一句正能量摘要《数据要素安全流通》《Python数据挖掘&#xff1a;入门、进阶与实用案例分析》《数据保护&#xff1a;工作负载的可恢复性 》《Data Mesh权威指南》《分布式统一大数据虚拟文件系统 Alluxio原理、技术与实践》《云原生数据中台&#xff1a;架构、方…
阅读更多...
【安装tensorflow-CPU版本】

【安装tensorflow-CPU版本】

一、安装目的二、安装过程三、总结 一、安装目的 使自己的jupyter能用tensorflow 二、安装过程 首先打开anaconda prompt 接着输入conda list 查看自己是否安装了tensorflow 在 Python 中使用 pip 工具来升级 pip 自身并指定了使用清华大学的镜像源进行安装 python -m pip …
阅读更多...
防雷接地检测方法应用方案

防雷接地检测方法应用方案

防雷接地是指为了保护建筑物、设备和人员免受雷电危害而采取的将雷电电流导入地面的措施。 地凯科技防雷接地的测试方案一般包括以下几个方面&#xff1a; 测试目的&#xff1a;明确测试的对象、范围、要求和依据。 测试内容&#xff1a;根据不同类型的防雷接地装置&#xf…
阅读更多...
计算机操作系统重点概念整理-第二章 进程管理【期末复习|考研复习】

计算机操作系统重点概念整理-第二章 进程管理【期末复习|考研复习】

第二章 进程管理 【期末复习|考研复习】 系列文章传送门&#xff1a; 第一章 计算机系统概述 第二章 进程管理 第三章 进程同步 第四章 内存管理 第五章 文件管理 第六章 输出输出I/O管理 文章目录 第二章 进程管理 【期末复习|考研复习】前言二、进程管理2.1进程的定义&#…
阅读更多...
【随机过程】布朗运动

【随机过程】布朗运动

这里写目录标题 Brownian motion Brownian motion The brownian motion 1D and brownian motion 2D functions, written with the cumsum command and without for loops, are used to generate a one-dimensional and two-dimensional Brownian motion, respectively. 使用cu…
阅读更多...
JVM虚拟机：对象在内存中的存储布局

JVM虚拟机：对象在内存中的存储布局

本文重点 在前面的过程中，我们学习了对象创建过程，那么一个对象在内存中的布局是什么样的呢？ 对象在内存中的存储布局 普通对象 当我们创建一个对象的时候，它由三部分组成，分别为对象头（MarkWord+class指针（指向class对象）），实例数据（对象的成员变量），填充。如果…
阅读更多...
C++——C++入门（二）

C++——C++入门（二）

C 前言一、引用引用概念引用特性常引用使用场景传值、传引用效率比较值和引用的作为返回值类型的性能比较 引用和指针的区别 二、内联函数概念特性知识点提升 三、auto关键字类型别名思考auto简介auto的使用细则auto不能推导的场景 四、基于范围的for循环范围for的语法范围for的…
阅读更多...
Linux操作系统的基础IO

Linux操作系统的基础IO

目录 系统文件IOopen函数0 & 1 & 2文件描述符的分配规则重定向输入重定向输出重定向追加重定向dup2 FILE 文件系统inode 软硬链接软链接硬链接 动态库和静态库动静态库的命名方式静态库制作一个库使用库 动态库制作一个库使用库 系统文件IO open函数 int open(const …
阅读更多...
tftp服务的搭建

tftp服务的搭建

TFTP服务的搭建 1 先更新一下apt包 sudo apt-get update2 服务器端(虚拟机上)安装 TFTP相关软件 sudo apt-get install xinetd tftp tftpd -y3 创建TFTP共享目录 mkdir tftp_sharetftp_shaer的路径是/home/cwz/tftp_share 3.1 修改共享目录的权限 sudo chmod -R 777 tftp…
阅读更多...
网络基础-2

网络基础-2

IEEE制定了一个名为GARP的协议框架&#xff0c;该框架协议包含了两个具体协议&#xff0c;GMRP和GVRP。GVRP可以大大降低VLAN配置过程中的手工的工作量。 IP本身是一个协议文件的名称&#xff0c;该协议主要定义阐释了IP报文的格式。 类型网络号位数网络号个数主机号位数每个…
阅读更多...
element-ui vue2 iframe 嵌入外链新解

element-ui vue2 iframe 嵌入外链新解

效果如图 实现原理 在路由中通过 props 传值 {path: /iframe,component: Layout,meta: { title: 小助手, icon: example },children: [{path: chatglm,name: chatglm,props: { name: chatglm,url: https://chatglm.cn },component: () > import(/views/iframe/common),me…
阅读更多...
【代码思路】2023mathorcup 大数据数学建模B题 电商零售商家需求预测及库存优化问题

【代码思路】2023mathorcup 大数据数学建模B题 电商零售商家需求预测及库存优化问题

各位同学们好&#xff0c;我们之前已经发布了第一问的思路视频&#xff0c;然后我们现在会详细的进行代码和结果的一个讲解&#xff0c;然后同时我们之后还会录制其他小问更详细的思路以及代码的手把手教学。 大家我们先看一下代码这一部分&#xff0c;我们采用的软件是Jupyte…
阅读更多...
DBA笔记（1）

DBA笔记（1）

目录 1、rpm yum 命令的使用&#xff0c;参数的含义 rpm命令&#xff1a; yum命令&#xff1a; 2、上传镜像至虚拟机搭建本地yum源 3、chown chomd 命令每一个参数的含义 chown命令&#xff1a; chmod命令&#xff1a; 4、fdisk partd 硬盘分区命令用法 fdisk命令&am…
阅读更多...
Pytest单元测试框架生成HTML测试报告及优化的步骤

Pytest单元测试框架生成HTML测试报告及优化的步骤

本文主要介绍了Pytest单元测试框架生成HTML测试报告及优化的步骤&#xff0c;文中通过示例代码介绍的非常详细&#xff0c;具有一定的参考价值&#xff0c;感兴趣的小伙伴们可以参考一下 一、安装插件 要生成html类型的报告&#xff0c;需要使用pytest-html插件&#xff0c;可…
阅读更多...
古剑奇谭木语人氪金最强阵容，土豪配置

古剑奇谭木语人氪金最强阵容，土豪配置

古剑奇谭木语人是一款3D回合制RPG手游&#xff0c;以其精湛的古风画质、跌宕起伏的剧情和丰富多样的玩法而闻名。游戏中拥有许多强大的角色&#xff0c;每个角色都拥有独特的技能和机制。为了发挥出最大的实力&#xff0c;我们需要将角色搭配成一支强大的阵容。以下是当前版本中…
阅读更多...
Beego之Beego简介和安装

Beego之Beego简介和安装

1、beego简介 1.1 Beego简介 Beego是一个快速开发 Go 应用的 HTTP 框架&#xff0c;他可以用来快速开发 API、Web 及后端服务等各种应用&#xff0c;是一个 RESTful 的框架&#xff0c;主要设计灵感来源于tornado、sinatra和 flask 这三个框架&#xff0c;但是结合了 Go 本身…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023