基于golang多消息队列中间件的封装nsq,rabbitmq,kafka

基于golang多消息队列中间件的封装nsq,rabbitmq,kafka

场景

在创建个人的公共方法库中有这样一个需求,就是不同的项目会用到不同的消息队列中间件,我的思路把所有的消息队列中间件进行封装一个消息队列接口(MQer)有两个方法一个生产一个消费,那么在实例化对象的时候根据配置文件指定当前项目使用的那个消息队列中间件;

接口模型

这个模型的核心思想是消息队列的核心功能生产者生产消息方法和消费者消费消息,任何消息队列都必须有这两个功能;根据如下代码消息队列中间件是可扩展的,只需在实例化消息队列对象那里添加新消息队列的实现;

// MQer 消息队列接口
type MQer interface {Producer(topic string, data []byte)Consumer(topic, channel string, ch chan []byte, f func(b []byte))
}// NewMQ 实例化消息队列对象
func NewMQ() MQer {switch conf.Conf.Default.Mq { // mq 设置的类型case "nsq":return new(MQNsqService)case "rabbit":return new(MQRabbitService)case "kafka":return new(MQKafkaService)default:return new(MQNsqService)}
}/*
配置文件结构设计mqType: "" # nsq, rabbit, kafka  这三个值然当然了是可扩展的nsq:producer: ""consumer: ""rabbit:addr: ""user: ""password: ""kafka:addr: ""
*/

各个消息队列的实现

1. 依赖库
  • nsq : github.com/nsqio/go-nsq
  • rabbitmq : github.com/streadway/amqp
  • kafka : github.com/Shopify/sarama
2. nsq

nsq结构体

// MQNsqService NSQ消息队列
type MQNsqService struct {
}

生产者

// Producer 生产者
func (m *MQNsqService) Producer(topic string, data []byte) {nsqConf := &nsq.Config{}client, err := nsq.NewProducer(nsqServer, nsqConf)if err != nil {log.Error("[nsq]无法连接到队列")return}log.DebugF(fmt.Sprintf("[生产消息] topic : %s -->  %s", topic, string(data)))err = client.Publish(topic, data)if err != nil {log.Error("[生产消息] 失败 : " + err.Error())}
}

消费者

var (nsqServer   = conf.Conf.Default.Nsq.Producer // nsqServer
)// Consumer 消费者
func (m *MQNsqService) Consumer(topic, channel string, ch chan []byte, f func(b []byte)) {mh, err := NewMessageHandler(nsqServer, channel)if err != nil {log.Error(err)return}go func() {mh.SetMaxInFlight(1000)mh.Registry(topic, ch)}()go func() {for {select {case s := <-ch:f(s)}}}()log.DebugF("[NSQ] ServerID:%v => %v started", channel, topic)
}// MessageHandler MessageHandler
type MessageHandler struct {msgChan     chan *goNsq.Messagestop        boolnsqServer   stringChannel     stringmaxInFlight int
}// NewMessageHandler return new MessageHandler
func NewMessageHandler(nsqServer string, channel string) (mh *MessageHandler, err error) {if nsqServer == "" {err = fmt.Errorf("[NSQ] need nsq server")return}mh = &MessageHandler{msgChan:   make(chan *goNsq.Message, 1024),stop:      false,nsqServer: nsqServer,Channel:   channel,}return
}// Registry register nsq topic
func (m *MessageHandler) Registry(topic string, ch chan []byte) {config := goNsq.NewConfig()if m.maxInFlight > 0 {config.MaxInFlight = m.maxInFlight}consumer, err := goNsq.NewConsumer(topic, m.Channel, config)if err != nil {panic(err)}consumer.SetLogger(nil, 0)consumer.AddHandler(goNsq.HandlerFunc(m.handlerMessage))err = consumer.ConnectToNSQLookupd(m.nsqServer)if err != nil {panic(err)}m.process(ch)
}
  1. rabbitmq
    结构体
// MQRabbitService Rabbit消息队列
type MQRabbitService struct {
}

生产者

// Producer 生产者
func (m *MQRabbitService) Producer(topic string, data []byte) {mq, err := NewRabbitMQPubSub(topic)if err != nil {log.Error("[rabbit]无法连接到队列")return}//defer mq.Destroy()log.DebugF(fmt.Sprintf("[生产消息] topic : %s -->  %s", topic, string(data)))err = mq.PublishPub(data)if err != nil {log.Error("[生产消息] 失败 : " + err.Error())}
}// NewRabbitMQPubSub 订阅模式创建 rabbitMq实例  (目前用的fanout模式)
func NewRabbitMQPubSub(exchangeName string) (*RabbitMQ, error) {mq, err := NewRabbitMQ("", exchangeName, "", "")if mq == nil || err != nil {return nil, err}//获取connectionmq.conn, err = amqp.Dial(mq.MqUrl)mq.failOnErr(err, "failed to connect mq!")if mq.conn == nil || err != nil {return nil, err}//获取channelmq.channel, err = mq.conn.Channel()mq.failOnErr(err, "failed to open a channel!")return mq, err
}...其余代码见源码: https://github.com/mangenotwork/common/tree/main/mq 

消费者

// Consumer 消费者
func (m *MQRabbitService) Consumer(topic, serverId string, ch chan []byte, f func(b []byte)) {mh, err := NewRabbitMQPubSub(topic)if err != nil {log.Error("[rabbit]无法连接到队列")return}msg := mh.RegistryReceiveSub()go func(m <-chan amqp.Delivery) {for {select {case s := <-m:f(s.Body)}}}(msg)log.DebugF("[Rabbit] ServerID:%v => %v started", serverId, topic)
}// NewRabbitMQPubSub 订阅模式创建 rabbitMq实例  (目前用的fanout模式)
func NewRabbitMQPubSub(exchangeName string) (*RabbitMQ, error) {mq, err := NewRabbitMQ("", exchangeName, "", "")if mq == nil || err != nil {return nil, err}//获取connectionmq.conn, err = amqp.Dial(mq.MqUrl)mq.failOnErr(err, "failed to connect mq!")if mq.conn == nil || err != nil {return nil, err}//获取channelmq.channel, err = mq.conn.Channel()mq.failOnErr(err, "failed to open a channel!")return mq, err
}... 其余代码见源码: https://github.com/mangenotwork/common/tree/main/mq 
  1. kafka
    结构体
// MQKafkaService Kafka消息队列
type MQKafkaService struct {
}

生产者

func (m *MQKafkaService) Producer(topic string, data []byte) {config := sarama.NewConfig()config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follower都确认config.Producer.Partitioner = sarama.NewRandomPartitioner //写到随机分区中,我们默认设置32个分区config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回// 构造一个消息msg := &sarama.ProducerMessage{}msg.Topic = topicmsg.Value = sarama.ByteEncoder(data)// 连接kafkaclient, err := sarama.NewSyncProducer(kafkaServer, config)if err != nil {log.Error("Producer closed, err:", err)return}defer client.Close()// 发送消息pid, offset, err := client.SendMessage(msg)if err != nil {log.Error("send msg failed, err:", err)return}log.InfoF("pid:%v offset:%v\n", pid, offset)
}

消费者

// Consumer 消费者
func (m *MQKafkaService) Consumer(topic, serverId string, ch chan []byte, f func(b []byte)) {var wg sync.WaitGroupconsumer, err := sarama.NewConsumer(kafkaServer, nil)if err != nil {log.ErrorF("Failed to start consumer: %s", err)return}partitionList, err := consumer.Partitions("task-status-data") // 通过topic获取到所有的分区if err != nil {log.Error("Failed to get the list of partition: ", err)return}log.Info(partitionList)for partition := range partitionList { // 遍历所有的分区pc, err := consumer.ConsumePartition(topic, int32(partition), sarama.OffsetNewest) // 针对每个分区创建一个分区消费者if err != nil {log.ErrorF("Failed to start consumer for partition %d: %s\n", partition, err)}wg.Add(1)go func(sarama.PartitionConsumer) { // 为每个分区开一个go协程取值for msg := range pc.Messages() { // 阻塞直到有值发送过来,然后再继续等待log.DebugF("Partition:%d, Offset:%d, key:%s, value:%s\n", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))f(msg.Value)}defer pc.AsyncClose()wg.Done()}(pc)}wg.Wait()consumer.Close()
}

总结

golang的接口是一种抽象类型,是对其他类型行为的概括与抽象,从语法角度来看,接口是一组方法定义的集合,文本的封装使用了golang接口这一特性,把所有的消息队列中间件抽象为一个MQer拥有生产和消费两个方法,具体的各个消息队列中间件去实现这两个方法即可,最明显的优点在于扩展性,解耦性,选择性,维护性这几个表象上。

完整代码

https://github.com/mangenotwork/common/tree/main/mq

你的星星是我分享的最大动力 : )

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

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

相关文章

跨境多商户中日韩英多语言商城搭建(PC+小程序+H5),搭建方案

随着全球化的推进&#xff0c;跨境电商正变得越来越普遍。在本文中&#xff0c;我们将介绍跨境电商系统开发中多语言商城独立站的部署搭建方案。 准备工作 在开始部署搭建之前&#xff0c;需要准备以下环境&#xff1a; 服务器&#xff0c;确保服务器具备足够的性能和稳定性。 …

Windows网络自学的第一天:创建线程

CreateThread函数&#xff1a; 该函数用于创建一个新的线程并在其上运行指定的函数&#xff0c;原型如下&#xff1a; HANDLE CreateThread(LPSECURITY_ATTRIBUTES lpThreadAttributes,SIZE_T dwStackSize,LPTHREAD_START_ROUTINE lpStartAddress,LPVOID …

RF手机天线仿真介绍(一):金属边框天线和LDS天线

目录 简介LDS天线LDS天线仿真 金属边框天线金属边框天线仿真 简介 最早的手机是外置式天线&#xff0c;从NOKIA开始采用内置式天线&#xff0c;开始采用内置金属片&#xff08;一般是0.1MM厚的不锈钢片冲压而成&#xff09;&#xff0c;随后为降低成本&#xff0c;后来改用FPC…

powershell脚本写一个托盘图标

1、准备ico格式图标 star_bethlehem_icon 文件名改为star.ico 2、安装VSCode 如何下载安装VSCode 扩展&#xff1a;PowerShell扩展 3、创建项目 1、运行PowerShell命令 mkdir trayicon_ps1;cd trayicon_ps1;New-Item trayicon.ps1;code .2、将star.ico放入trayicon_ps1文…

DDS中间件设计

OpenDDS、FastDDS数据分发服务中间件设计 软件架构 应用层DDS层RTPS层传输层 软件层次 FastDDS整体架构如下&#xff0c;这里可以看到DDS和RTPS的关系。另外缺少一部分IDL&#xff08;统一描述语言&#xff09;&#xff0c;其应该是Pub、Sub的反序列化、序列化工具。 在RT…

【深度学习】各个开源库总结及实战-总目录

前言 此专栏主要是用MMCV和PaddlePaddle,它们都是优秀的开源库,用于计算机视觉和深度学习任务。MMCV提供了丰富的计算机视觉工具和算法,基于PyTorch框架,适合对PyTorch熟悉的用户。PaddlePaddle是百度开发的深度学习平台,提供易用且高性能的深度学习框架。 此专栏主要包括…

Ubuntu安装harbor(http模式)并随便上传一个

Ubuntu安装harbor&#xff08;http模式&#xff09; docker和harbor的介绍就免了&#xff0c;都不知道啥东西&#xff0c;还安装搞毛 先安装docker环境 不要问&#xff0c;软件源之类的配置&#xff0c;挨个梭就行 sudo apt update sudo apt install apt-transport-https ca…

Python3 高级教程 | Python3 CGI编程(二)

目录 一、什么是CGI 二、网页浏览 三、CGI架构图 四、Web服务器支持及配置 五、第一个CGI程序 六、HTTP头部 七、CGI环境变量 八、GET和POST方法 &#xff08;一&#xff09;使用GET方法传输数据 &#xff08;二&#xff09;简单的url实例&#xff1a;GET方法 &#x…

什么是redux?如何在react 项目中使用redux?

redux 概念 redux是一种用于管理JavaScript应用程序的状态管理库。它可以与React、Augular、Vue等前端框架结合使用&#xff0c;但也可以纯在JavaScript应用程序中独立使用。redux遵循单项数据流的原则&#xff0c;通过一个全局的状态树来管理应用程序的状态&#xff0c;从而使…

快速排序——“数据结构与算法”

各位CSDN的uu们好呀&#xff0c;今天又是小雅兰的数据结构与算法专栏啦&#xff0c;下面&#xff0c;就让我们进入快速排序的世界吧&#xff01;&#xff01;&#xff01; 快速排序 快速排序是Hoare于1962年提出的一种二叉树结构的交换排序方法&#xff0c;其基本思想为&…

【Docker】Docker容器与虚拟机的对比、三要素、架构和镜像加速器的详细讲解

&#x1f680;欢迎来到本文&#x1f680; &#x1f349;个人简介&#xff1a;陈童学哦&#xff0c;目前学习C/C、算法、Python、Java等方向&#xff0c;一个正在慢慢前行的普通人。 &#x1f3c0;系列专栏&#xff1a;陈童学的日记 &#x1f4a1;其他专栏&#xff1a;CSTL&…

什么?你还没有用过JPA Buddy,那么你工作肯定没5年

1. 概述 JPA Buddy是一个广泛使用的IntelliJ IDEA插件&#xff0c;面向使用JPA数据模型和相关技术&#xff08;如Spring DataJPA&#xff0c;DB版本控制工具&#xff08;Flyway&#xff0c;Liquibase&#xff09;&#xff0c;MapStruct等&#xff09;的新手和有经验的开发人员…

局域网ssh登录windows自带Linux系统(WSL)踩坑记录

局域网ssh登录windows自带Linux系统&#xff08;WSL&#xff09;踩坑记录 系统和工具安装查看IPLinux用户和端口设置Windows ssh 登录Mac ssh 登录 系统和工具安装 先按照网上的教程安装好Linux系统&#xff08;一般都是安装Ubuntu&#xff09;&#xff0c;安装好ssh等工具&am…

《Java-SE-第二十七章》之常见的锁策略

前言 在你立足处深挖下去,就会有泉水涌出!别管蒙昧者们叫嚷:“下边永远是地狱!” 博客主页&#xff1a;KC老衲爱尼姑的博客主页 博主的github&#xff0c;平常所写代码皆在于此 共勉&#xff1a;talk is cheap, show me the code 作者是爪哇岛的新手&#xff0c;水平很有限&…

RD算法(四)登堂入室 —— 成像完成

SAR成像专栏目录_lightninghenry的博客-CSDN博客https://lightning.blog.csdn.net/article/details/122393577?spm=1001.2014.3001.5502先放RD算法最终的成像结果: 经简单的地距投影后为(地距投影的内容在后面的几何校正章节中讲解): 温哥华这地形还真像是一张怪兽的巨嘴呀…

方差分析||判断数据是否符合正态分布

方差分析练习题 练习学习笔记&#xff1a; &#xff08;1&#xff09; 标准差和标准偏差、均方差是一个东西。标准误差和标准误是一个东西。这两个东西有区别。 &#xff08;2&#xff09;单因素方差分析&#xff08;MATLAB求解&#xff09; &#xff08;3&#xff09;使用an…

【雕爷学编程】MicroPython动手做(28)——物联网之Yeelight

知识点&#xff1a;什么是掌控板&#xff1f; 掌控板是一块普及STEAM创客教育、人工智能教育、机器人编程教育的开源智能硬件。它集成ESP-32高性能双核芯片&#xff0c;支持WiFi和蓝牙双模通信&#xff0c;可作为物联网节点&#xff0c;实现物联网应用。同时掌控板上集成了OLED…

同比环比使用方法

一、解释&#xff1a; 1.同比&#xff1a;本期与去年同期相比 &#xff0c;如2023年8月 比 2022年8月 2.环比&#xff1a;本期与上期相比 &#xff0c;如2023年8月 比 2023年7月 二、应用&#xff1a; 1.场景&#xff1a;统计日报、周报、月报、年报下进店客流的同比和环…

TypeScript使用技巧

文章目录 使用技巧TypeScript内置的工具类型keyofextends 限定泛型interface 与 type 区别 TypeScript作为JavaScript的超集,通过提供静态类型系统和对ES6新特性的支持,使JavaScript开发变得更加高效和可维护。掌握TypeScript的使用技巧,可以帮助我们更好地开发和组织JavaScrip…

银河麒麟V10 飞腾 Qt环境搭建

采用在线安装方式&#xff1a; 1、在线安装qt组件 sudo apt-get install qt5-* 2、在线安装qt creator sudo apt-get install qtcreator 以上简单两步安装完成后&#xff0c;新建项目已经可以编译过&#xff0c;但ClangCodeModel会报错如下图 the code model could not parse …