深圳网站维护,自媒体135官网,做网站商业欺骗赔多少,珠海网站建设公司基于golang多消息队列中间件的封装nsq,rabbitmq,kafka
场景
在创建个人的公共方法库中有这样一个需求#xff0c;就是不同的项目会用到不同的消息队列中间件#xff0c;我的思路把所有的消息队列中间件进行封装一个消息队列接口#xff08;MQer#xff09;有两个方法一个…基于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-nsqrabbitmq : github.com/streadway/amqpkafka : 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)
}
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 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
你的星星是我分享的最大动力 : )
