goframe开发一个企业网站 rabbitmq队例15

RabbitMQ消息队列封装

在目录internal/pkg/rabbitmq/rabbitmq.go

# 消息队列配置
mq:# 消息队列类型: rocketmq 或 rabbitmqtype: "rabbitmq"# 是否启用消息队列enabled: truerocketmq:nameServer: "127.0.0.1:9876"producerGroup: "myProducerGroup"consumerGroup: "myConsumerGroup"brokerAddress: "127.0.0.1:10911"  # 添加 broker 地址rabbitmq:url: "amqp://wanghaibin:wanghaibin@127.0.0.1:5672/"exchange: "gf_exchange"dlx_exchange: "gf_dlx_exchange"    # 新增:死信交换机queue: "gf_queue"delay_queue: "gf_delay_queue"      # 新增:延迟队列routingKey: "gf_key"vhost: "/"
package rabbitmqimport ("context""fmt""time""github.com/gogf/gf/v2/frame/g"amqp "github.com/rabbitmq/amqp091-go"
)var (// conn RabbitMQ连接实例conn *amqp.Connection// channel RabbitMQ通道实例channel *amqp.Channel
)// Initialize 初始化 RabbitMQ 连接和通道
// 包括:建立连接、创建通道、声明交换机和队列、建立绑定关系
func Initialize() {var err errorctx := context.Background()// 从配置文件获取RabbitMQ连接URLurl := g.Cfg().MustGet(ctx, "rabbitmq.url").String()// 建立RabbitMQ连接conn, err = amqp.Dial(url)if err != nil {g.Log().Fatalf(ctx, "Failed to connect to RabbitMQ: %v", err)}// 创建通道channel, err = conn.Channel()if err != nil {g.Log().Fatalf(ctx, "Failed to open channel: %v", err)}// 1. 声明主交换机// 类型:direct,持久化:true,自动删除:false,内部的:false,非阻塞:falseerr = channel.ExchangeDeclare(g.Cfg().MustGet(ctx, "rabbitmq.exchange").String(),"direct", // 交换机类型true,     // 持久化false,    // 自动删除false,    // 内部的false,    // 非阻塞nil,      // 参数)if err != nil {g.Log().Fatalf(ctx, "Failed to declare main exchange: %v", err)}// 2. 声明死信交换机(DLX)// 用于处理无法被正常消费的消息err = channel.ExchangeDeclare(g.Cfg().MustGet(ctx, "rabbitmq.dlx_exchange").String(),"direct",true,false,false,false,nil,)if err != nil {g.Log().Fatalf(ctx, "Failed to declare DLX exchange: %v", err)}// 3. 声明主队列// 持久化:true,非自动删除,非排他,非阻塞_, err = channel.QueueDeclare(g.Cfg().MustGet(ctx, "rabbitmq.queue").String(),true,  // 持久化false, // 自动删除false, // 排他的false, // 非阻塞nil,   // 参数)if err != nil {g.Log().Fatalf(ctx, "Failed to declare main queue: %v", err)}// 4. 声明延迟队列// 配置死信交换机参数args := amqp.Table{"x-dead-letter-exchange":    g.Cfg().MustGet(ctx, "rabbitmq.dlx_exchange").String(),"x-dead-letter-routing-key": g.Cfg().MustGet(ctx, "rabbitmq.routingKey").String(),}_, err = channel.QueueDeclare(g.Cfg().MustGet(ctx, "rabbitmq.delay_queue").String(),true,false,false,false,args,)if err != nil {g.Log().Fatalf(ctx, "Failed to declare delay queue: %v", err)}// 5. 绑定主队列到主交换机err = channel.QueueBind(g.Cfg().MustGet(ctx, "rabbitmq.queue").String(),      // 队列名g.Cfg().MustGet(ctx, "rabbitmq.routingKey").String(), // 路由键g.Cfg().MustGet(ctx, "rabbitmq.exchange").String(),   // 交换机名false,nil,)if err != nil {g.Log().Fatalf(ctx, "Failed to bind main queue: %v", err)}// 6. 绑定主队列到死信交换机err = channel.QueueBind(g.Cfg().MustGet(ctx, "rabbitmq.queue").String(),g.Cfg().MustGet(ctx, "rabbitmq.routingKey").String(),g.Cfg().MustGet(ctx, "rabbitmq.dlx_exchange").String(),false,nil,)if err != nil {g.Log().Fatalf(ctx, "Failed to bind queue to DLX: %v", err)}g.Log().Info(ctx, "RabbitMQ initialized successfully")
}// PublishMessage 发布消息到RabbitMQ
// 参数:
//   - ctx: 上下文
//   - message: 要发送的消息内容
// 返回:
//   - error: 发送错误,如果成功则为nil
func PublishMessage(ctx context.Context, message string) error {// 创建带超时的上下文ctxTimeout, cancel := context.WithTimeout(ctx, 5*time.Second)defer cancel()// 发布消息到指定的交换机和路由err := channel.PublishWithContext(ctxTimeout,g.Cfg().MustGet(ctx, "rabbitmq.exchange").String(),g.Cfg().MustGet(ctx, "rabbitmq.routingKey").String(),false, // mandatoryfalse, // immediateamqp.Publishing{ContentType: "text/plain",Body:        []byte(message),},)if err != nil {return fmt.Errorf("failed to publish message: %v", err)}return nil
}// ConsumeMessages 消费队列中的消息
// 参数:
//   - ctx: 上下文
//   - handler: 消息处理函数
// 返回:
//   - error: 消费错误,如果成功则为nil
func ConsumeMessages(ctx context.Context, handler func(string) error) error {messages, err := channel.Consume(g.Cfg().MustGet(ctx, "rabbitmq.queue").String(),"",    // consumerfalse, // auto-ackfalse, // exclusivefalse, // no-localfalse, // no-waitnil,   // args)if err != nil {return fmt.Errorf("failed to register a consumer: %v", err)}// 启动goroutine处理消息go func() {for msg := range messages {err := handler(string(msg.Body))if err != nil {g.Log().Errorf(ctx, "Error handling message: %v", err)msg.Nack(false, true) // 处理失败,消息重新入队} else {msg.Ack(false) // 处理成功,确认消息}}}()return nil
}// Cleanup 清理RabbitMQ连接和通道
func Cleanup() {if channel != nil {channel.Close()}if conn != nil {conn.Close()}
}// GetChannel 获取RabbitMQ通道实例
func GetChannel() *amqp.Channel {return channel
}// PurgeQueue 清空指定队列中的所有消息
// 参数:
//   - ctx: 上下文
// 返回:
//   - error: 清空错误,如果成功则为nil
func PurgeQueue(ctx context.Context) error {_, err := channel.QueuePurge(g.Cfg().MustGet(ctx, "rabbitmq.queue").String(),false, // no-wait)return err
}// PublishDelayMessage 发送延迟消息
// 参数:
//   - ctx: 上下文
//   - message: 消息内容
//   - delaySeconds: 延迟秒数
// 返回:
//   - error: 发送错误,如果成功则为nil
func PublishDelayMessage(ctx context.Context, message string, delaySeconds int) error {return channel.PublishWithContext(ctx,"",                                                    // 默认交换机g.Cfg().MustGet(ctx, "rabbitmq.delay_queue").String(), // 延迟队列false,false,amqp.Publishing{ContentType: "text/plain",Body:        []byte(message),Expiration:  fmt.Sprintf("%d", delaySeconds*1000), // 转换为毫秒},)
}// GetQueueLength 获取队列中的消息数量
// 参数:
//   - ctx: 上下文
// 返回:
//   - int: 消息数量
//   - error: 获取错误,如果成功则为nil
func GetQueueLength(ctx context.Context) (int, error) {queue, err := channel.QueueInspect(g.Cfg().MustGet(ctx, "rabbitmq.queue").String(),)if err != nil {return 0, fmt.Errorf("failed to inspect queue: %v", err)}return queue.Messages, nil
}
logic逻辑的实现
package rabbitmqmsgimport ("context""fmt""gf_new_web/internal/pkg/rabbitmq""gf_new_web/internal/service""github.com/gogf/gf/v2/frame/g"
)// sRabbitmqMsg RabbitMQ消息服务结构体
type sRabbitmqMsg struct{}// New 创建新的RabbitMQ消息服务实例
func New() *sRabbitmqMsg {return &sRabbitmqMsg{}
}// init 初始化函数,在包加载时自动注册RabbitMQ消息服务
func init() {service.RegisterRabbitmqMsg(New())
}// SendMessage 发送普通消息到RabbitMQ
// 参数:
//   - ctx: 上下文信息
//   - message: 要发送的消息内容
// 返回:
//   - error: 发送错误,成功则为nil
func (s *sRabbitmqMsg) SendMessage(ctx context.Context, message string) error {return rabbitmq.PublishMessage(ctx, message)
}// SendDelayMessage 发送延迟消息到RabbitMQ
// 参数:
//   - ctx: 上下文信息
//   - message: 要发送的消息内容
//   - delaySeconds: 延迟时间(秒)
// 返回:
//   - error: 发送错误,成功则为nil
func (s *sRabbitmqMsg) SendDelayMessage(ctx context.Context, message string, delaySeconds int) error {return rabbitmq.PublishDelayMessage(ctx, message, delaySeconds)
}// SendBatchMessages 批量发送消息到RabbitMQ
// 参数:
//   - ctx: 上下文信息
//   - messages: 消息内容数组
// 返回:
//   - error: 发送错误,成功则为nil
// 注意:任一消息发送失败都会导致整个批次失败
func (s *sRabbitmqMsg) SendBatchMessages(ctx context.Context, messages []string) error {for _, msg := range messages {if err := rabbitmq.PublishMessage(ctx, msg); err != nil {return err}}return nil
}// GetQueueLength 获取队列当前的消息数量
// 参数:
//   - ctx: 上下文信息
// 返回:
//   - int: 队列中的消息数量
//   - error: 获取错误,成功则为nil
func (s *sRabbitmqMsg) GetQueueLength(ctx context.Context) (int, error) {queue, err := rabbitmq.GetChannel().QueueInspect(g.Cfg().MustGet(ctx, "rabbitmq.queue").String(),)if err != nil {return 0, fmt.Errorf("failed to inspect queue: %v", err)}return queue.Messages, nil
}// PurgeQueue 清空队列中的所有消息
// 参数:
//   - ctx: 上下文信息
// 返回:
//   - error: 清空错误,成功则为nil
func (s *sRabbitmqMsg) PurgeQueue(ctx context.Context) error {return rabbitmq.PurgeQueue(ctx)
}// handleMessage 处理接收到的单条消息
// 参数:
//   - message: 消息内容
// 返回:
//   - error: 处理错误,成功则为nil
// 注意:这是内部方法,实现具体的消息处理逻辑
func (s *sRabbitmqMsg) handleMessage(message string) error {// 记录接收到的消息g.Log().Info(context.Background(), "收到消息:", message)// TODO: 在这里添加实际的消息处理逻辑return nil
}// Initialize 初始化消息消费处理
// 参数:
//   - ctx: 上下文信息
// 返回:
//   - error: 初始化错误,成功则为nil
// 功能:启动消息消费者,并设置消息处理函数
func (s *sRabbitmqMsg) Initialize(ctx context.Context) error {return rabbitmq.ConsumeMessages(ctx, func(msg string) error {return s.handleMessage(msg)})
}

生成service ,不再写上

controller代码

package frontimport ("fmt""gf_new_web/internal/service""time""github.com/gogf/gf/v2/frame/g""github.com/gogf/gf/v2/net/ghttp"
)var (RabbitMsg = cRabbitMsg{}
)type cRabbitMsg struct{}// SendMessage 处理发送普通消息的HTTP请求
// 请求参数:
//   - message: 消息内容
// 响应格式:
//   成功:{"code": 0, "msg": "消息发送成功"}
//   失败:{"code": -1, "msg": "错误信息"}
func (c *cRabbitMsg) SendMessage(r *ghttp.Request) {message := r.Get("message").String()err := service.RabbitmqMsg().SendMessage(r.GetCtx(), message)if err != nil {g.Log().Error(r.GetCtx(), err)r.Response.WriteJson(g.Map{"code": -1,"msg":  err.Error(),})return}r.Response.WriteJson(g.Map{"code": 0,"msg":  "消息发送成功",})
}// SendDelayMessage 处理发送延迟消息的HTTP请求
// 请求参数:
//   - message: 消息内容
//   - delay: 延迟时间(秒)
// 响应格式:
//   成功:{"code": 0, "msg": "延迟消息发送成功"}
//   失败:{"code": -1, "msg": "错误信息"}
func (c *cRabbitMsg) SendDelayMessage(r *ghttp.Request) {message := r.Get("message").String()delaySeconds := r.Get("delay").Int()err := service.RabbitmqMsg().SendDelayMessage(r.GetCtx(), message, delaySeconds)if err != nil {g.Log().Error(r.GetCtx(), err)r.Response.WriteJson(g.Map{"code": -1,"msg":  err.Error(),})return}r.Response.WriteJson(g.Map{"code": 0,"msg":  "延迟消息发送成功",})
}// SendBatchMessages 处理批量发送消息的HTTP请求
// 请求参数:
//   - messages: 消息内容数组
// 响应格式:
//   成功:{"code": 0, "msg": "批量消息发送成功"}
//   失败:{"code": -1, "msg": "错误信息"}
func (c *cRabbitMsg) SendBatchMessages(r *ghttp.Request) {messages := r.Get("messages").Strings()err := service.RabbitmqMsg().SendBatchMessages(r.GetCtx(), messages)if err != nil {g.Log().Error(r.GetCtx(), err)r.Response.WriteJson(g.Map{"code": -1,"msg":  err.Error(),})return}r.Response.WriteJson(g.Map{"code": 0,"msg":  "批量消息发送成功",})
}// GetQueueLength 处理获取队列长度的HTTP请求
// 响应格式:
//   成功:{"code": 0, "msg": "获取队列长度成功", "data": 队列长度}
//   失败:{"code": -1, "msg": "错误信息"}
func (c *cRabbitMsg) GetQueueLength(r *ghttp.Request) {length, err := service.RabbitmqMsg().GetQueueLength(r.GetCtx())if err != nil {g.Log().Error(r.GetCtx(), err)r.Response.WriteJson(g.Map{"code": -1,"msg":  err.Error(),})return}r.Response.WriteJson(g.Map{"code": 0,"msg":  "获取队列长度成功","data": length,})
}// PurgeQueue 处理清空队列的HTTP请求
// 响应格式:
//   成功:{"code": 0, "msg": "清空队列成功"}
//   失败:{"code": -1, "msg": "错误信息"}
func (c *cRabbitMsg) PurgeQueue(r *ghttp.Request) {err := service.RabbitmqMsg().PurgeQueue(r.GetCtx())if err != nil {g.Log().Error(r.GetCtx(), err)r.Response.WriteJson(g.Map{"code": -1,"msg":  err.Error(),})return}r.Response.WriteJson(g.Map{"code": 0,"msg":  "清空队列成功",})
}// ConsumeMessages 处理消费消息的HTTP请求
// 特点:异步处理,非阻塞
// 响应格式:
//   成功:{"code": 0, "msg": "消息消费已开始,请查看服务器日志获取消费详情"}
//   失败:{"code": -1, "msg": "错误信息"}
func (c *cRabbitMsg) ConsumeMessages(r *ghttp.Request) {g.Log().Info(r.GetCtx(), "开始消费消息...")done := make(chan bool)go func() {err := service.RabbitmqMsg().Initialize(r.GetCtx())if err != nil {g.Log().Error(r.GetCtx(), "消费消息出错:", err)r.Response.WriteJson(g.Map{"code": -1,"msg":  fmt.Sprintf("消费消息失败: %v", err),})done <- truereturn}}()select {case <-done:returncase <-time.After(5 * time.Second):g.Log().Info(r.GetCtx(), "消息消费进行中...")r.Response.WriteJson(g.Map{"code": 0,"msg":  "消息消费已开始,请查看服务器日志获取消费详情",})}
}

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

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

相关文章

前端web

题目&#xff1a;制作带有下拉悬停菜单的导航栏 效果图 一、先制作标签 <body> <div id"menu"> <div id"container"> <div class"item">游戏1 <div class…

【大数据学习 | kafka高级部分】kafka中的选举机制

controller的选举 首先第一个选举就是借助于zookeeper的controller的选举 第一个就是controller的选举&#xff0c;这个选举是借助于zookeeper的独享锁实现的&#xff0c;先启动的broker会在zookeeper的/contoller节点上面增加一个broker信息&#xff0c;谁创建成功了谁就是主…

Android 开发指南:初学者入门

Android 是全球最受欢迎的移动操作系统之一&#xff0c;为开发者提供了丰富的工具和资源来创建各种类型的应用程序。本文将为你提供一个全面的入门指南&#xff0c;帮助你从零开始学习 Android 开发。 目录 1. 了解 Android 平台[1]2. 设置开发环境[2]3. 学习基础知识[3]4. 创…

漏洞挖掘 | 某医院小程序支付漏洞+越权

漏洞挖掘 | 某医院小程序支付漏洞越权 登陆后点击个人信息&#xff0c;抓包&#xff0c;放到repeter模块 修改strUserID参数可以越权查看别人信息 放intruder模块可以跑数据&#xff0c;这里有几万信息泄露 回到首页&#xff0c;点击医生咨询功能点 随便选一个需要付费的医…

MFC图形函数学习07——画扇形函数

绘制扇形函数是MFC中绘图的基本函数&#xff0c;它绘制的仍是由椭圆弧与椭圆中心连线构成的椭圆扇形&#xff0c;特例是由圆弧与圆心连线构成的圆扇形。 一、绘制扇形函数 原型&#xff1a;BOOL Pie(int x1,int y1,int x2,int y2,int x3,int y3,int x4,int y4); …

基于Python的膳食健康系统

作者&#xff1a;计算机学姐 开发技术&#xff1a;SpringBoot、SSM、Vue、MySQL、JSP、ElementUI、Python、小程序等&#xff0c;“文末源码”。 专栏推荐&#xff1a;前后端分离项目源码、SpringBoot项目源码、Vue项目源码、SSM项目源码、微信小程序源码 精品专栏&#xff1a;…

头歌网络安全(11.12)

头歌禁止复制解决 必须先下篡改猴&#xff01;&#xff01;&#xff01;&#xff01; 头歌复制助手 Educoder Copy Helperhttps://scriptcat.org/zh-CN/script-show-page/1860 Java生成验证码 第1关&#xff1a;使用Servlet生成验证码 任务描述 本关任务&#xff1a;使用se…

ROM修改进阶教程------安卓14 安卓15去除app签名验证的几种操作步骤 详细图文解析

在安卓14 安卓15的固件中。如果修改了系统级别的app。那么就会触发安卓14 15的应用签名验证。要么会导致修改的固件会进不去系统,或者进入系统有bug。博文将从几方面来解析去除安卓14 15应用签名验证的几种方法。 💝💝💝通过博文了解: 1💝💝💝-----安卓14去除…

第七部分:2. STM32之ADC实验--AD多通道(AD采集三路传感器模块实验:光敏传感器、热敏传感器、反射式传感器附赠温湿度传感器教程)

这个多通道采用非扫描模式--单次转换模式 1.代码配置链路图 2. ADC的输入通道 3.ADC的非扫描模式的转换模式&#xff08;单次和连续&#xff09; 4.ADC的扫描模式的转换模式&#xff08;单次和连续&#xff09; 5.采集校准 代码实验&#xff1a; 代码部分&#xff1a; #inclu…

crond 任务调度 (Linux相关指令:crontab)

相关视频链接 crontab 进行 定时任务 的设置 概述 任务调度&#xff1a;是指系统在某个时间执行的特定的命令或程序 任务调度的分类&#xff1a; 1.系统工作&#xff1a;有些重要的工作必须周而复始地执行。如病毒扫描等。 2.个别用户可能希望执行某些程序&#xff0c;比如…

基于python 的opencv 使用GrabCut算法分割图像代码

#利用grabcut算法分割图像import numpy as np import cv2 from matplotlib import pyplot as plt import warningswarnings.filterwarnings("ignore", module "matplotlib")imgpath E:/code/image_opencv_test/lena.jpg img cv2.imread(imgpath)Coords1x…

AndroidStudio-文本显示

一、设置文本的内容 1.方式&#xff1a; &#xff08;1&#xff09;在XML文件中通过属性&#xff1a;android:text设置文本 例如&#xff1a; <?xml version"1.0" encoding"utf-8"?> <LinearLayout xmlns:android"http://schemas.andr…

css:没错又是我

背景 给元素添加背景样式 还可以设置背景颜色、背景图片&#xff08;教练我要学这个&#xff09;、背景平铺、背景图片位置、背景图像固定 背景颜色 这个我们用过&#xff0c;就是&#xff1a; a {background-color: hotpink; } 一般默认值是transparent&#xff0c;也就…

使用Git工具在GitHub的仓库中上传文件夹(超详细)

如何使用Git工具在GitHub的仓库中上传文件夹&#xff1f; 如果觉得博主写的还可以&#xff0c;点赞收藏关注噢~ 第一步&#xff1a;拥有一个本地的仓库 可以fork别人的仓库或者自己新创建 fork别人的仓库 或者自己创建一个仓库 按照要求填写完成后&#xff0c;点击按钮创建…

uniapp的基本使用(easycom规范和条件编译)和uview组件的安装和使用

文章目录 1、uniapp1.uview组件安装2.uview-plus组件安装 2、条件编译3、easycom规范1.组件路径符合规范2.自定义easycom配置的示例 总结 1、uniapp UniApp的UI组件库&#xff0c;如TMUI、uViewUI、FirstUI、TuniaoUI、ThorUI等&#xff0c;这些组件库适用于Vue3和TypeScript&…

攻防世界37-unseping-CTFWeb

攻防世界37-unseping-CTFWeb <?php highlight_file(__FILE__);class ease{private $method;private $args;function __construct($method, $args) {$this->method $method;$this->args $args;}function __destruct(){if (in_array($this->method, array("…

【大数据学习 | HBASE高级】region split机制和策略

1. region split机制 ​ HRegionServer拆分region的步骤是&#xff0c;先将该region下线&#xff0c;然后拆分&#xff0c;将其子region加入到hbase:meta表中&#xff0c;再将他们加入到原本的HRegionServer中&#xff0c;最后汇报Master。 split前&#xff1a;hbase:meta表有…

FMC 扩展子卡6 路 422,8 组 LVDS,8 路 GPIO

FMC 扩展子卡6 路 422,8 组 LVDS,8 路 GPIO 卡是一款支持多路 LVCMOS 和 LVDS 信号互转的 FMC 扩展子板。它能支持 6 路 422 信号的输入 / 输出 ,8 组 LVDS 信号的输入 / 输出和 8 路 GPIO 信号的输入 / 输出。本产品基于一些逻辑转换芯片而设计&#xff0c;能实现差分信号转单…

old-cms(原生PHP开发的企业网站管理系统)

old-cms是一个使用原生PHP开发的实用的PHP企业网站管理系统&#xff0c;包括企业网站常用的功能板块&#xff0c;如&#xff1a;产品管理、新闻管理、栏目管理、模板标签管理、分类管理、诚聘英才、在线留言反馈、关于我们&#xff08;公司简介&#xff09;等等&#xff0c;也有…

IPv4与IPv6的优缺点

IPv4 和 IPv6 都是 TCP/IP 协议的版本。IP 是指互联网协议&#xff0c;是传输控制协议/互联网协议套件&#xff08;TCP/IP&#xff09;的主要部分。 TCP/IP 是一套标准和规则&#xff0c;用于规范不同网络上的设备之间打包数据&#xff08;数据报&#xff09;的传输和交换。互…