Golang使用消息队列(RabbitMQ)

最近在使用Golang做了一个网盘项目(类似百度网盘),这个网盘项目有一个功能描述如下:用户会删除一个文件到垃圾回收站,回收站的文件有一个时间期限,比如24h,24h后数据库中记录和oss中文件会被删除,在之前的版本中,可以使用定时任务来检查数据库记录中删除时间来判断是否删除,但是这不是最佳的,因此考虑如何基于RabbitMQ来实现这个功能。

使用RabbitMQ的架构

在这里插入图片描述

代码

因为前端有点麻烦,这里全部使用Golang后端来模拟实现整个架构,包括生产端和消费端。这里有一些细节

  • 注意交换机和队列的绑定,一定要细心
  • 交换机一旦声明了就不能更改,如果要发生一些属性的更改,就要删除原来的内容,重新生成
  • 下列的内容不包含RabbitMQ持久化的内容
package mainimport ("fmt""github.com/streadway/amqp""log""strings"
)func InitRabbitMQ() *amqp.Connection {mq := "amqp"host := "127.0.0.1"port := "5672"user := "root"pwd := "root"dns := strings.Join([]string{mq, "://", user, ":", pwd,"@", host, ":", port, "/"}, "")conn, err := amqp.Dial(dns)if err != nil {log.Fatalf("Failed to connect to RabbitMQ: %v", err)}return conn
}func InitMainExchangeAndQueue(ch *amqp.Channel, userID string) *amqp.Channel {// 队列信息exchangeName := "main_exchange"queueName := fmt.Sprintf("user_queue_%s", userID)messageTTL := int32(300000)// 声明主交换机err := ch.ExchangeDeclare(exchangeName, // 交换机名"direct",     // Exchange typefalse,        // Durablefalse,        // Auto-deletedfalse,        // Internalfalse,        // No-waitnil,          // Arguments)if err != nil {log.Fatalf("Failed to declare an main exchange: %v", err)}// 声明用户队列_, err = ch.QueueDeclare(queueName, // 队列名false,     // Durablefalse,     // Delete when unusedfalse,     // Exclusivefalse,     // No-waitamqp.Table{"x-dead-letter-routing-key": "dead",          // routing-key"x-dead-letter-exchange":    "dead_exchange", // 死信交换机"x-message-ttl":             messageTTL,      // TTL},)if err != nil {log.Fatalf("Failed to declare a queue: %v", err)}// 绑定err = ch.QueueBind(queueName, userID, "main_exchange", false, nil)if err != nil {log.Fatalf("Failed to bind queue to exchange: %v", err)}return ch
}func InitDeadExchangeAndQueue(ch *amqp.Channel) {// 声明死信交换机err := ch.ExchangeDeclare("dead_exchange",amqp.ExchangeDirect,true,false,false,false,nil,)if err != nil {log.Fatalf("Failed to declare an dead exchange: %v", err)}// 声明一个死信队列_, err = ch.QueueDeclare("dead_queue",true,false,false,false,nil)if err != nil {log.Fatalf("Failed to declare a queue: %v", err)}// 绑定err = ch.QueueBind("dead_queue", "dead", "dead_exchange", false, nil)if err != nil {log.Fatalf("Failed to bind queue to exchange: %v", err)}
}func PublishMessage(ch *amqp.Channel, userID, fileID string) {// 用户信息message := fmt.Sprintf("%s|%s", userID, fileID)exchangeName := "main_exchange"// 发布用户消息err := ch.Publish(exchangeName, // ExchangeuserID,       // Routing keyfalse,        // Mandatoryfalse,        // Immediateamqp.Publishing{ContentType: "text/plain",Body:        []byte(message),})if err != nil {log.Fatalf("Failed to publish a message: %v", err)}log.Printf("Message sent to user %s: %s", userID, message)
}func ConsumeTTL(ch *amqp.Channel) {// 声明死信交换机err := ch.ExchangeDeclare("dead_exchange", // 交换机名"direct",        // Exchange typetrue,            // Durablefalse,           // Auto-deletedfalse,           // Internalfalse,           // No-waitnil,             // Arguments)if err != nil {log.Fatalf("Failed to declare a dead letter exchange: %v", err)}// 创建消费者并阻塞等待消费死信队列中的消息megs, err := ch.Consume("dead_queue", // Queue"",           // Consumerfalse,        // Auto-acknowledgefalse,        // Exclusivefalse,        // No-localfalse,        // No-waitnil,          // Args)if err != nil {log.Fatalf("Failed to register a consumer for dead letter queue: %v", err)}// 使用无限循环一直监听fmt.Println("Waiting for message from dead_queue......")for d := range megs {// 实际中,处理消息的逻辑,例如删除文件或其他操作fmt.Println(string(d.Body))// 消费完成后手动确认消息err = d.Ack(false)if err != nil {log.Fatalf("Failed to ack message: %v", err)}}
}func Consume(ch *amqp.Channel, userID string) {// 下面的信息可以通过前后端进行传递queueName := fmt.Sprintf("user_queue_%s", userID)// 消费消息megs, err := ch.Consume(queueName, // Queue"",        // Consumertrue,      // Auto-acknowledgefalse,     // Exclusivefalse,     // No-localfalse,     // No-waitnil,       // Args)if err != nil {log.Fatalf("Failed to register a consumer: %v", err)}// 这里直接是由前端发送过来的API进行触发,所以不用一直阻塞监听d, ok := <-megsif !ok {log.Fatalf("Failed to get message: %v", err)}fmt.Println(string(d.Body))// 消息完成后确认消息err = d.Ack(true)if err != nil {log.Fatalf("Failed to ack message: %v", err)}
}func main() {// 获取客户端client := InitRabbitMQ()defer client.Close()ch, err := client.Channel()if err != nil {log.Fatalf("Failed to open a channel: %v", err)}defer ch.Close()//ConsumeTTL(ch)// 构造dead_exchange及dead_queue// InitDeadExchangeAndQueue(ch)// 假设这是web请求信息//var userID1 = "test-id1"//var fileID1 = "file1"// 构造main_exchange及user_queue//ch = InitMainExchangeAndQueue(ch, userID1)// 针对用户1:假设还消息没有过期时候就被recovery,即在user_queue中就被消费,实际中发布消息的这部分逻辑应当放在前端中//PublishMessage(ch, userID1, fileID1)//time.Sleep(20 * time.Second)// 模拟后端消费消息//Consume(ch, userID1)// 针对用户2:模拟其不被后端消费,过期到死信队列中var userID2 = "test-id2"var fileID2 = "file2"ch = InitMainExchangeAndQueue(ch, userID2)PublishMessage(ch, userID2, fileID2)// 注意这个消息没有被消费,理论上应当被死信队列消费
}

从dead_exchange中消费:
在这里插入图片描述

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

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

相关文章

C语言之扫雷游戏实现篇

目录 主函数test.c 菜单函数 选择循环 扫雷游戏实现分析 整体思路 问题1 问题2 问题3 问题4 游戏函数&#xff08;函数调用&#xff09; 创建游戏盘数组mine 创建游戏盘数组show 初始化游戏盘数组InitBoard 展示游戏盘DisplayBoard 游戏盘置雷SetMine 游戏…

设计模式之创建者模式

文章目录 一、介绍二、应用三、案例1. 麦当劳11随心配2. 代码演示3. 演示结果 四、优缺点五、送给读者 一、介绍 建造者模式(Builder Pattern)属于创建型设计模式&#xff0c;很多博客文章的对它的作用解释为用于将复杂对象的创建过程与其细节表示分离。但对于初学者来说&…

VS2022 C++ 批量追加hosts文件IP地址与域名脚本V1.0

这是自己使用VS2022 C++开发的批量追加"C:\Windows\System32\drivers\etc\hosts"文件IP地址与域名的脚本。它可以把cfg.txt文件中例如: 192.168.1.9 test.email 192.168.123.9 doutia.com批量追加到hosts文件 详细源代码如下: AddToHostHosts.cpp //#includ…

Pyqt5打开电脑摄像头进行拍照

目录 1、设计UI界面 2、设计逻辑代码&#xff0c;建立连接显示窗口 3、结果 1、设计UI界面 将ui界面转为py文件后获得的逻辑代码为&#xff1a;&#xff08;文件名为 Camera.py&#xff09; # -*- coding: utf-8 -*-# Form implementation generated from reading ui file …

【ES6】—【必备知识】—深拷贝与浅拷贝

一、如何把一个对象复制给另一个对象 1.Object.assign() let target {} let source {a: 1,b: 2 } Object.assign(target,source) console.log(target) // {a: 1, b: 2}2.Object.assign() 只能实现浅拷贝 let target {a: {b:1,c: 8} } let source {a: {b:2} } Object.ass…

无涯教程-PHP - File 函数

文件系统功能用于访问和操纵文件系统&#xff0c;PHP为您提供了操纵文件的所有功能。 运行时配置 这些功能的行为受php.ini中的设置影响。 NameDefaultChangeableChangelogallow_url_fopen"1"PHP_INI_ALLPHP_INI_ALL in PHP < 4.3.4. PHP_INI_SYSTEM in PHP &l…

Linux:如何创建用户

“useradd” 命令用于在 Linux 或类 Unix 系统上创建新用户账户。它是一个需要超级用户权限的命令&#xff0c;因此你需要以 root 或具有 sudo 权限的用户身份运行它。下面是使用 “useradd” 命令创建新用户的基本语法&#xff1a; useradd [选项] 用户名 下面是一些常用选项…

windows上ffmpeg如何录制双屏幕中的一个屏幕上的视频

首先&#xff0c;如何在window上安装ffmpeg自己查找scoop安装ffmpeg. 如题&#xff1a; 如果你有两个屏幕&#xff0c;如何让ffmpeg来录制其中的一个屏幕的视频呢。 很简单&#xff0c;首先你要查看另外一个屏幕的分辨率&#xff1a; 第一步&#xff1a;进入系统中 第二步&am…

硬件产品的量产问题------硬件工程师在产线关注什么

前言&#xff1a; 产品开发测试无误&#xff0c;但量产缺遇到很多不良甚至DOA问题。 硬件开发过程中如何确保产线的治具、生产及硬件工程师在产线需要关注一些什么。 坚信&#xff1a;好的产品是要可以做出来的。 1、禁忌&#xff1a; 禁忌热插拔&#xff1b;禁忌测试不防呆…

Python Opencv实践 - 图像直方图均衡化

import cv2 as cv import numpy as np import matplotlib.pyplot as pltimg cv.imread("../SampleImages/pomeranian.png", cv.IMREAD_COLOR) print(img.shape)#图像直方图计算 #cv.calcHist(images, channels, mask, histSize, ranges, hist, accumulate) #images&…

link 和 @import 有什么区别

页面导入样式时&#xff0c;使用 link 和 import 有什么区别 HTML中使用 link 标签 <link rel"stylesheet" href"style.css" />CSS 中 import import "style.css"; import url("style.css"); /* 两种方式无差别 */两者的区别…

数学建模知识之小白入门篇

数学建模知识--小白入门篇 一、数学模型的定义二、建立数学模型的方法和步骤1. 模型准备2. 模型假设3. 模型构成4. 模型求解5. 模型分析 三、数模竞赛出题的指导思想四、竞赛中的常见题型1. 实际问题背景2&#xff0e;若干假设条件3&#xff0e;要求回答的问题 五、提交一篇论文…

ModStartBlog v8.0.0 博客归档页面,部分组件升级

ModStart 是一个基于 Laravel 模块化极速开发框架。模块市场拥有丰富的功能应用&#xff0c;支持后台一键快速安装&#xff0c;让开发者能快的实现业务功能开发。 系统完全开源&#xff0c;基于 Apache 2.0 开源协议。 功能特性 丰富的模块市场&#xff0c;后台一键快速安装会…

C语言入门 Day_9 条件判断

目录 前言&#xff1a; 1.if判断 2.else判断 3.易错点 4.思维导图 前言&#xff1a; 我们知道比较运算和逻辑运算都会得到一个布尔型的数据&#xff0c;要么为真&#xff08;true&#xff09;&#xff0c;要么为假&#xff08;false&#xff09;。 今天我们来学习真和假在…

面试热题(复原ip地址)

有效 IP 地址 正好由四个整数&#xff08;每个整数位于 0 到 255 之间组成&#xff0c;且不能含有前导 0&#xff09;&#xff0c;整数之间用 . 分隔。 例如&#xff1a;"0.1.2.201" 和 "192.168.1.1" 是 有效 IP 地址&#xff0c;但是 "0.011.255.24…

pyqt5-自定义停靠栏头部

import sys from PyQt5.QtWidgets import * from PyQt5.QtCore import * from PyQt5.QtGui import *class CustomDock(QDockWidget):def __init__(self, title: str, parentNone):super().__init__(title, parent)"""停靠栏的头部"""h_layout Q…

行式存储与列式存储

1.概述 数据处理大致可分为两大类&#xff0c;联机事务处理OLTP(on-line transaction processing) 和联机分析处理OLAP(on-line analytical processing)。 OLTP是传统关系型数据库的主要应用&#xff0c;用来执行一些基本的、日常的事务处理&#xff0c;比如数据库记录的增、删…

android NullPointerException externalCacheDir

先看代码&#xff1a; fun Context.getMyCacheDir(): String {return externalCacheDir!!.absolutePath "/my_cache" }如上代码&#xff0c;在某些手机可能会出现crash。 原因详细阅读api&#xff0c;注意他有一个大大的注解Nullable&#xff1a; Nullablepublic a…

uniapp微信小程序点击右上角菜单分享功能权限配置

个人项目地址&#xff1a; SubTopH前端开发个人站 &#xff08;自己开发的前端功能和UI组件&#xff0c;一些有趣的小功能&#xff0c;感兴趣的伙伴可以访问&#xff0c;欢迎提出更好的想法&#xff0c;私信沟通&#xff0c;网站属于静态页面&#xff09; SubTopH前端开发个人站…

solidity0.8.0的应用案例7:线性释放合约

在传统金融领域,一些公司会向员工和管理层提供股权。但大量股权同时释放会在短期产生抛售压力,拖累股价。因此,公司通常会引入一个归属期来延迟承诺资产的所有权。 同样的,在区块链领域,Web3初创公司会给团队分配代币,同时也会将代币低价出售给风投和私募。如果他们把这…