feat: add Go backend
This commit is contained in:
richjohnrambo
@@ -0,0 +1,75 @@
|
||||
// Package queue
|
||||
// @Link https://github.com/bufanyun/hotgo
|
||||
// @Copyright Copyright (c) 2023 HotGo CLI
|
||||
// @Author Ms <133814250@qq.com>
|
||||
// @License https://github.com/bufanyun/hotgo/blob/master/LICENSE
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// Consumer 消费者接口，实现该接口即可加入到消费队列中
|
||||
type Consumer interface {
|
||||
GetTopic() string // 获取消费主题
|
||||
Handle(ctx context.Context, mqMsg MqMsg) (err error) // 处理消息的方法
|
||||
}
|
||||
|
||||
// consumerManager 消费者管理
|
||||
type consumerManager struct {
|
||||
sync.Mutex
|
||||
list map[string]Consumer // 维护的消费者列表
|
||||
}
|
||||
|
||||
var consumers = &consumerManager{
|
||||
list: make(map[string]Consumer),
|
||||
}
|
||||
|
||||
// RegisterConsumer 注册任务到消费者队列
|
||||
func RegisterConsumer(cs Consumer) {
|
||||
consumers.Lock()
|
||||
defer consumers.Unlock()
|
||||
topic := cs.GetTopic()
|
||||
if _, ok := consumers.list[topic]; ok {
|
||||
Logger().Debugf(ctx, "queue.RegisterConsumer topic:%v duplicate registration.", topic)
|
||||
return
|
||||
}
|
||||
consumers.list[topic] = cs
|
||||
}
|
||||
|
||||
// StartConsumersListener 启动所有已注册的消费者监听
|
||||
func StartConsumersListener(ctx context.Context) {
|
||||
for _, c := range consumers.list {
|
||||
go func(c Consumer) {
|
||||
consumerListen(ctx, c)
|
||||
}(c)
|
||||
}
|
||||
}
|
||||
|
||||
// consumerListen 消费者监听
|
||||
func consumerListen(ctx context.Context, job Consumer) {
|
||||
var (
|
||||
topic = job.GetTopic()
|
||||
c, err = InstanceConsumer()
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
Logger().Fatalf(ctx, "InstanceConsumer %s err:%+v", topic, err)
|
||||
return
|
||||
}
|
||||
|
||||
if listenErr := c.ListenReceiveMsgDo(topic, func(mqMsg MqMsg) {
|
||||
err = job.Handle(ctx, mqMsg)
|
||||
|
||||
// if err != nil {
|
||||
// // 遇到错误，重新加入到队列
|
||||
// //queue.Push(topic, mqMsg.Body)
|
||||
// }
|
||||
|
||||
// 记录消费队列日志
|
||||
ConsumerLog(ctx, topic, mqMsg, err)
|
||||
}); listenErr != nil {
|
||||
Logger().Fatalf(ctx, "消费队列：%s 监听失败, err:%+v", topic, listenErr)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,148 @@
|
||||
// Package queue
|
||||
// @Link https://github.com/bufanyun/hotgo
|
||||
// @Copyright Copyright (c) 2023 HotGo CLI
|
||||
// @Author Ms <133814250@qq.com>
|
||||
// @License https://github.com/bufanyun/hotgo/blob/master/LICENSE
|
||||
package queue
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/gogf/gf/v2/errors/gerror"
|
||||
"github.com/gogf/gf/v2/os/gfile"
|
||||
"hotgo/internal/library/queue/disk"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Disk 磁盘队列
|
||||
|
||||
type DiskProducerMq struct {
|
||||
config *disk.Config
|
||||
producers map[string]*disk.Queue
|
||||
sync.Mutex
|
||||
}
|
||||
|
||||
type DiskConsumerMq struct {
|
||||
config *disk.Config
|
||||
}
|
||||
|
||||
func RegisterDiskMqConsumer(config *disk.Config) (client MqConsumer, err error) {
|
||||
return &DiskConsumerMq{
|
||||
config: config,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// ListenReceiveMsgDo 消费数据
|
||||
func (q *DiskConsumerMq) ListenReceiveMsgDo(topic string, receiveDo func(mqMsg MqMsg)) (err error) {
|
||||
if topic == "" {
|
||||
return gerror.New("disk.ListenReceiveMsgDo topic is empty")
|
||||
}
|
||||
|
||||
var (
|
||||
queue = NewDiskQueue(topic, q.config)
|
||||
sleep = time.Second
|
||||
)
|
||||
|
||||
go func() {
|
||||
for {
|
||||
if index, offset, data, err := queue.Read(); err == nil {
|
||||
var mqMsg MqMsg
|
||||
if err = json.Unmarshal(data, &mqMsg); err != nil {
|
||||
Logger().Warningf(ctx, "disk.ListenReceiveMsgDo Unmarshal err:%+v, topic：%v, data:%+v .", err, topic, string(data))
|
||||
continue
|
||||
}
|
||||
if mqMsg.MsgId != "" {
|
||||
receiveDo(mqMsg)
|
||||
queue.Commit(index, offset)
|
||||
sleep = time.Millisecond * 10
|
||||
}
|
||||
} else {
|
||||
sleep = time.Second
|
||||
}
|
||||
|
||||
time.Sleep(sleep)
|
||||
}
|
||||
}()
|
||||
|
||||
select {}
|
||||
}
|
||||
|
||||
func RegisterDiskMqProducer(config *disk.Config) (client MqProducer, err error) {
|
||||
return &DiskProducerMq{
|
||||
config: config,
|
||||
producers: make(map[string]*disk.Queue),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// SendMsg 按字符串类型生产数据
|
||||
func (d *DiskProducerMq) SendMsg(topic string, body string) (mqMsg MqMsg, err error) {
|
||||
return d.SendByteMsg(topic, []byte(body))
|
||||
}
|
||||
|
||||
// SendByteMsg 生产数据
|
||||
func (d *DiskProducerMq) SendByteMsg(topic string, body []byte) (mqMsg MqMsg, err error) {
|
||||
if topic == "" {
|
||||
return mqMsg, gerror.New("DiskMq topic is empty")
|
||||
}
|
||||
|
||||
mqMsg = MqMsg{
|
||||
RunType: SendMsg,
|
||||
Topic: topic,
|
||||
MsgId: getRandMsgId(),
|
||||
Body: body,
|
||||
Timestamp: time.Now(),
|
||||
}
|
||||
|
||||
mqMsgJson, err := json.Marshal(mqMsg)
|
||||
if err != nil {
|
||||
return mqMsg, gerror.New(fmt.Sprint("queue redis 生产者解析json消息失败:", err))
|
||||
}
|
||||
|
||||
queue := d.getProducer(topic)
|
||||
if err = queue.Write(mqMsgJson); err != nil {
|
||||
return mqMsg, gerror.New(fmt.Sprint("queue disk 生产者添加消息失败:", err))
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (d *DiskProducerMq) SendDelayMsg(topic string, body string, delaySecond int64) (mqMsg MqMsg, err error) {
|
||||
err = gerror.New("implement me")
|
||||
return
|
||||
}
|
||||
|
||||
func (d *DiskProducerMq) getProducer(topic string) *disk.Queue {
|
||||
queue, ok := d.producers[topic]
|
||||
if ok {
|
||||
return queue
|
||||
}
|
||||
queue = NewDiskQueue(topic, d.config)
|
||||
d.Lock()
|
||||
defer d.Unlock()
|
||||
d.producers[topic] = queue
|
||||
return queue
|
||||
}
|
||||
|
||||
func NewDiskQueue(topic string, config *disk.Config) *disk.Queue {
|
||||
conf := &disk.Config{
|
||||
Path: fmt.Sprintf(config.Path + "/" + config.GroupName + "/" + topic),
|
||||
BatchSize: config.BatchSize,
|
||||
BatchTime: config.BatchTime * time.Second,
|
||||
SegmentSize: config.SegmentSize,
|
||||
SegmentLimit: config.SegmentLimit,
|
||||
}
|
||||
|
||||
if !gfile.Exists(conf.Path) {
|
||||
if err := gfile.Mkdir(conf.Path); err != nil {
|
||||
Logger().Errorf(ctx, "NewDiskQueue Failed to create the cache directory. Procedure, err:%+v", err)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
queue, err := disk.New(conf)
|
||||
if err != nil {
|
||||
Logger().Errorf(ctx, "NewDiskQueue err:%v", err)
|
||||
return nil
|
||||
}
|
||||
return queue
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
// Package disk
|
||||
// @Link https://github.com/bufanyun/hotgo
|
||||
// @Copyright Copyright (c) 2023 HotGo CLI
|
||||
// @Author Ms <133814250@qq.com>
|
||||
// @License https://github.com/bufanyun/hotgo/blob/master/LICENSE
|
||||
package disk
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
filePerm = 0600 // 数据写入权限
|
||||
indexFile = ".index" // 消息索引文件
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
GroupName string // 组群名称
|
||||
Path string // 数据存放路径
|
||||
BatchSize int64 // 每N条消息同步一次，batchSize和batchTime满足其一就会同步一次
|
||||
BatchTime time.Duration // 每N秒消息同步一次
|
||||
SegmentSize int64 // 每个topic分片数据文件最大字节
|
||||
SegmentLimit int64 // 每个topic最大分片数据文件数量
|
||||
}
|
||||
|
||||
type Queue struct {
|
||||
sync.RWMutex
|
||||
close bool
|
||||
ticker *time.Ticker
|
||||
wg *sync.WaitGroup
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
writer *writer
|
||||
reader *reader
|
||||
}
|
||||
|
||||
func New(config *Config) (queue *Queue, err error) {
|
||||
if _, err = os.Stat(config.Path); err != nil {
|
||||
return
|
||||
}
|
||||
queue = &Queue{close: false, wg: &sync.WaitGroup{}, writer: &writer{config: config}, reader: &reader{config: config}}
|
||||
queue.ticker = time.NewTicker(config.BatchTime)
|
||||
queue.ctx, queue.cancel = context.WithCancel(context.TODO())
|
||||
err = queue.reader.restore()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
go queue.sync()
|
||||
return
|
||||
}
|
||||
|
||||
// Write data
|
||||
func (q *Queue) Write(data []byte) error {
|
||||
if q.close {
|
||||
return errors.New("closed")
|
||||
}
|
||||
|
||||
q.Lock()
|
||||
defer q.Unlock()
|
||||
|
||||
return q.writer.write(data)
|
||||
}
|
||||
|
||||
// Read data
|
||||
func (q *Queue) Read() (int64, int64, []byte, error) {
|
||||
if q.close {
|
||||
return 0, 0, nil, errors.New("closed")
|
||||
}
|
||||
|
||||
q.RLock()
|
||||
defer q.RUnlock()
|
||||
|
||||
index, offset, data, err := q.reader.read()
|
||||
if err == io.EOF && (q.writer.file == nil || q.reader.file.Name() != q.writer.file.Name()) {
|
||||
_ = q.reader.safeRotate()
|
||||
}
|
||||
return index, offset, data, err
|
||||
}
|
||||
|
||||
// Commit index and offset
|
||||
func (q *Queue) Commit(index int64, offset int64) {
|
||||
if q.close {
|
||||
return
|
||||
}
|
||||
|
||||
ck := &q.reader.checkpoint
|
||||
ck.Index, ck.Offset = index, offset
|
||||
q.reader.sync()
|
||||
}
|
||||
|
||||
// Close Queue
|
||||
func (q *Queue) Close() {
|
||||
if q.close {
|
||||
return
|
||||
}
|
||||
|
||||
q.close = true
|
||||
q.cancel()
|
||||
q.wg.Wait()
|
||||
q.writer.close()
|
||||
q.reader.close()
|
||||
}
|
||||
|
||||
// sync data
|
||||
func (q *Queue) sync() {
|
||||
q.wg.Add(1)
|
||||
defer q.wg.Done()
|
||||
for {
|
||||
select {
|
||||
case <-q.ticker.C:
|
||||
q.Lock()
|
||||
q.writer.sync()
|
||||
q.Unlock()
|
||||
case <-q.ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,181 @@
|
||||
// Package disk
|
||||
// @Link https://github.com/bufanyun/hotgo
|
||||
// @Copyright Copyright (c) 2023 HotGo CLI
|
||||
// @Author Ms <133814250@qq.com>
|
||||
// @License https://github.com/bufanyun/hotgo/blob/master/LICENSE
|
||||
package disk
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
var (
|
||||
errorQueueEmpty = errors.New("queue is empty")
|
||||
)
|
||||
|
||||
type reader struct {
|
||||
file *os.File
|
||||
index int64
|
||||
offset int64
|
||||
reader *bufio.Reader
|
||||
checkpoint checkpoint
|
||||
config *Config
|
||||
}
|
||||
|
||||
type checkpoint struct {
|
||||
Index int64 `json:"index"`
|
||||
Offset int64 `json:"offset"`
|
||||
}
|
||||
|
||||
// read data
|
||||
func (r *reader) read() (int64, int64, []byte, error) {
|
||||
if err := r.check(); err != nil {
|
||||
return r.index, r.offset, nil, err
|
||||
}
|
||||
|
||||
// read a line
|
||||
data, err := r.reader.ReadBytes('\n')
|
||||
if err != nil {
|
||||
return r.index, r.offset, nil, err
|
||||
}
|
||||
data = bytes.TrimRight(data, "\n")
|
||||
|
||||
r.offset += int64(len(data)) + 1
|
||||
return r.index, r.offset, data, err
|
||||
}
|
||||
|
||||
// check a new segment
|
||||
func (r *reader) check() error {
|
||||
if r.file != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
file, err := r.next()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return r.open(file)
|
||||
}
|
||||
|
||||
func (r *reader) open(file string) (err error) {
|
||||
if r.file, err = os.OpenFile(file, os.O_RDONLY, filePerm); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// get file index
|
||||
r.index = r.getIndex(file)
|
||||
|
||||
// seek read offset
|
||||
if _, err = r.file.Seek(r.offset, 0); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
r.reader = bufio.NewReader(r.file)
|
||||
return nil
|
||||
}
|
||||
|
||||
// safeRotate to next segment
|
||||
func (r *reader) safeRotate() error {
|
||||
// if there is no next file, it is not cleared
|
||||
if _, err := r.next(); err == errorQueueEmpty {
|
||||
return nil
|
||||
}
|
||||
|
||||
return r.rotate()
|
||||
}
|
||||
|
||||
// rotate to next segment
|
||||
func (r *reader) rotate() error {
|
||||
if r.file == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// close segment
|
||||
_ = r.file.Close()
|
||||
r.file, r.offset, r.reader = nil, 0, nil
|
||||
return nil
|
||||
}
|
||||
|
||||
// close reader
|
||||
func (r *reader) close() {
|
||||
if r.file == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if err := r.file.Close(); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
r.file, r.reader, r.index, r.offset = nil, nil, 0, 0
|
||||
}
|
||||
|
||||
// sync index and offset
|
||||
func (r *reader) sync() {
|
||||
name := path.Join(r.config.Path, indexFile)
|
||||
if data, err := json.Marshal(&r.checkpoint); err == nil {
|
||||
_ = os.WriteFile(name, data, filePerm)
|
||||
}
|
||||
}
|
||||
|
||||
// restore index and offset
|
||||
func (r *reader) restore() (err error) {
|
||||
name := path.Join(r.config.Path, indexFile)
|
||||
|
||||
// uninitialized
|
||||
if _, err1 := os.Stat(name); err1 != nil {
|
||||
r.sync()
|
||||
}
|
||||
|
||||
data, _ := os.ReadFile(name)
|
||||
|
||||
_ = json.Unmarshal(data, &r.checkpoint)
|
||||
r.index, r.offset = r.checkpoint.Index, r.checkpoint.Offset
|
||||
if r.index == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
if err = r.open(fmt.Sprintf("%s/%d.data", r.config.Path, r.index)); err != nil {
|
||||
r.offset = 0
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// next segment
|
||||
func (r *reader) next() (string, error) {
|
||||
files, err := filepath.Glob(filepath.Join(r.config.Path, "*.data"))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
sort.Strings(files)
|
||||
|
||||
for _, file := range files {
|
||||
index := r.getIndex(file)
|
||||
if index < r.checkpoint.Index {
|
||||
_ = os.Remove(file) // remove expired segment
|
||||
}
|
||||
|
||||
if index > r.index {
|
||||
return file, nil
|
||||
}
|
||||
}
|
||||
|
||||
return "", errorQueueEmpty
|
||||
}
|
||||
|
||||
// get segment index
|
||||
func (r *reader) getIndex(filename string) int64 {
|
||||
base := filepath.Base(filename)
|
||||
name := base[0 : len(base)-len(path.Ext(filename))]
|
||||
index, _ := strconv.ParseInt(name, 10, 64)
|
||||
return index
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
// Package disk
|
||||
// @Link https://github.com/bufanyun/hotgo
|
||||
// @Copyright Copyright (c) 2023 HotGo CLI
|
||||
// @Author Ms <133814250@qq.com>
|
||||
// @License https://github.com/bufanyun/hotgo/blob/master/LICENSE
|
||||
package disk
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path"
|
||||
"path/filepath"
|
||||
"time"
|
||||
)
|
||||
|
||||
type writer struct {
|
||||
file *os.File
|
||||
size int64
|
||||
count int64
|
||||
writer *bufio.Writer
|
||||
config *Config
|
||||
}
|
||||
|
||||
// write data
|
||||
func (w *writer) write(data []byte) error {
|
||||
// append newline
|
||||
data = append(data, "\n"...)
|
||||
size := int64(len(data))
|
||||
|
||||
// close current segment for rotate
|
||||
if w.size+size > w.config.SegmentSize {
|
||||
w.close()
|
||||
}
|
||||
|
||||
// create a new segment
|
||||
if w.file == nil {
|
||||
if err := w.open(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// write to buffer
|
||||
if _, err := w.writer.Write(data); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
w.size += size
|
||||
|
||||
// sync data to disk
|
||||
w.count++
|
||||
if w.count >= w.config.BatchSize {
|
||||
w.sync()
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// create a new segment
|
||||
func (w *writer) open() (err error) {
|
||||
if w.segmentNum() >= w.config.SegmentLimit {
|
||||
return errors.New("segment num exceeds the limit")
|
||||
}
|
||||
|
||||
name := path.Join(w.config.Path, fmt.Sprintf("%013d.data", time.Now().UnixNano()/1e6))
|
||||
if w.file, err = os.OpenFile(name, os.O_CREATE|os.O_WRONLY, filePerm); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
w.size = 0
|
||||
// disable auto flush
|
||||
w.writer = bufio.NewWriterSize(w.file, int(w.config.SegmentSize))
|
||||
w.writer.Reset(w.file)
|
||||
return err
|
||||
}
|
||||
|
||||
// sync data to disk
|
||||
func (w *writer) sync() {
|
||||
if w.writer == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if err := w.writer.Flush(); err == nil {
|
||||
w.count = 0
|
||||
}
|
||||
}
|
||||
|
||||
// close segment
|
||||
func (w *writer) close() {
|
||||
if w.file == nil {
|
||||
return
|
||||
}
|
||||
|
||||
w.sync()
|
||||
if err := w.file.Close(); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
w.size, w.file, w.writer = 0, nil, nil
|
||||
}
|
||||
|
||||
// segment num
|
||||
func (w *writer) segmentNum() int64 {
|
||||
segments, _ := filepath.Glob(path.Join(w.config.Path, "*.data"))
|
||||
return int64(len(segments))
|
||||
}
|
||||
@@ -0,0 +1,258 @@
|
||||
// Package queue
|
||||
// @Link https://github.com/bufanyun/hotgo
|
||||
// @Copyright Copyright (c) 2023 HotGo CLI
|
||||
// @Author Ms <133814250@qq.com>
|
||||
// @License https://github.com/bufanyun/hotgo/blob/master/LICENSE
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/IBM/sarama"
|
||||
"github.com/gogf/gf/v2/errors/gerror"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"hotgo/internal/consts"
|
||||
"hotgo/utility/simple"
|
||||
"time"
|
||||
)
|
||||
|
||||
type KafkaMq struct {
|
||||
Partitions int32
|
||||
producerIns sarama.AsyncProducer
|
||||
consumerIns sarama.ConsumerGroup
|
||||
}
|
||||
|
||||
type KafkaConfig struct {
|
||||
ClientId string
|
||||
Brokers []string
|
||||
GroupID string
|
||||
Partitions int32
|
||||
Replication int16
|
||||
Version string
|
||||
UserName string
|
||||
Password string
|
||||
}
|
||||
|
||||
// SendMsg 按字符串类型生产数据
|
||||
func (r *KafkaMq) SendMsg(topic string, body string) (mqMsg MqMsg, err error) {
|
||||
return r.SendByteMsg(topic, []byte(body))
|
||||
}
|
||||
|
||||
// SendByteMsg 生产数据
|
||||
func (r *KafkaMq) SendByteMsg(topic string, body []byte) (mqMsg MqMsg, err error) {
|
||||
msg := &sarama.ProducerMessage{
|
||||
Topic: topic,
|
||||
Value: sarama.ByteEncoder(body),
|
||||
Timestamp: time.Now(),
|
||||
}
|
||||
|
||||
if r.producerIns == nil {
|
||||
err = gerror.New("queue kafka producerIns is nil")
|
||||
return
|
||||
}
|
||||
|
||||
r.producerIns.Input() <- msg
|
||||
sendCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
select {
|
||||
case info := <-r.producerIns.Successes():
|
||||
return MqMsg{
|
||||
RunType: SendMsg,
|
||||
Topic: info.Topic,
|
||||
Offset: info.Offset,
|
||||
Partition: info.Partition,
|
||||
Timestamp: info.Timestamp,
|
||||
}, nil
|
||||
case fail := <-r.producerIns.Errors():
|
||||
if nil != fail {
|
||||
return mqMsg, fail.Err
|
||||
}
|
||||
case <-sendCtx.Done():
|
||||
return mqMsg, gerror.New("send mqMst timeout")
|
||||
}
|
||||
return mqMsg, nil
|
||||
}
|
||||
|
||||
func (r *KafkaMq) SendDelayMsg(topic string, body string, delaySecond int64) (mqMsg MqMsg, err error) {
|
||||
err = gerror.New("implement me")
|
||||
return
|
||||
}
|
||||
|
||||
// ListenReceiveMsgDo 消费数据
|
||||
func (r *KafkaMq) ListenReceiveMsgDo(topic string, receiveDo func(mqMsg MqMsg)) (err error) {
|
||||
if r.consumerIns == nil {
|
||||
return gerror.New("queue kafka consumer not register")
|
||||
}
|
||||
|
||||
consumer := KaConsumer{
|
||||
ready: make(chan bool),
|
||||
receiveDoFun: receiveDo,
|
||||
}
|
||||
|
||||
consumerCtx, cancel := context.WithCancel(context.Background())
|
||||
go func(consumerCtx context.Context) {
|
||||
for {
|
||||
if err = r.consumerIns.Consume(consumerCtx, []string{topic}, &consumer); err != nil {
|
||||
Logger().Fatalf(ctx, "kafka Error from consumer, err%+v", err)
|
||||
}
|
||||
|
||||
if consumerCtx.Err() != nil {
|
||||
Logger().Debugf(ctx, fmt.Sprintf("kafka consoumer stop : %v", consumerCtx.Err()))
|
||||
return
|
||||
}
|
||||
consumer.ready = make(chan bool)
|
||||
}
|
||||
}(consumerCtx)
|
||||
|
||||
// await till the consumer has been set up
|
||||
<-consumer.ready
|
||||
Logger().Debug(ctx, "kafka consumer up and running!...")
|
||||
|
||||
simple.Event().Register(consts.EventServerClose, func(ctx context.Context, args ...interface{}) {
|
||||
Logger().Debug(ctx, "kafka consumer close...")
|
||||
cancel()
|
||||
if err = r.consumerIns.Close(); err != nil {
|
||||
Logger().Fatalf(ctx, "kafka Error closing client, err:%+v", err)
|
||||
}
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
// RegisterKafkaMqConsumer 注册消费者
|
||||
func RegisterKafkaMqConsumer(connOpt KafkaConfig) (client MqConsumer, err error) {
|
||||
mqIns := &KafkaMq{}
|
||||
kfkVersion, err := sarama.ParseKafkaVersion(connOpt.Version)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if !validateVersion(kfkVersion) {
|
||||
kfkVersion = sarama.V2_4_0_0
|
||||
}
|
||||
|
||||
brokers := connOpt.Brokers
|
||||
conf := sarama.NewConfig()
|
||||
conf.Consumer.Return.Errors = true
|
||||
conf.Version = kfkVersion
|
||||
if connOpt.UserName != "" {
|
||||
conf.Net.SASL.Enable = true
|
||||
conf.Net.SASL.User = connOpt.UserName
|
||||
conf.Net.SASL.Password = connOpt.Password
|
||||
}
|
||||
|
||||
// 默认按随机方式消费
|
||||
conf.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
|
||||
conf.Consumer.Offsets.Initial = sarama.OffsetNewest
|
||||
conf.Consumer.Offsets.AutoCommit.Interval = 10 * time.Millisecond
|
||||
conf.ClientID = connOpt.ClientId
|
||||
|
||||
consumerClient, err := sarama.NewConsumerGroup(brokers, connOpt.GroupID, conf)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
mqIns.consumerIns = consumerClient
|
||||
return mqIns, err
|
||||
}
|
||||
|
||||
// RegisterKafkaProducer 注册并启动生产者接口实现
|
||||
func RegisterKafkaProducer(connOpt KafkaConfig) (client MqProducer, err error) {
|
||||
mqIns := &KafkaMq{}
|
||||
connOpt.ClientId = "HOTGO-Producer"
|
||||
|
||||
// 这里如果使用go程需要处理chan同步问题
|
||||
if err = doRegisterKafkaProducer(connOpt, mqIns); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return mqIns, nil
|
||||
}
|
||||
|
||||
// doRegisterKafkaProducer 注册同步类型实例
|
||||
func doRegisterKafkaProducer(connOpt KafkaConfig, mqIns *KafkaMq) (err error) {
|
||||
kfkVersion, err := sarama.ParseKafkaVersion(connOpt.Version)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if !validateVersion(kfkVersion) {
|
||||
kfkVersion = sarama.V2_4_0_0
|
||||
}
|
||||
|
||||
brokers := connOpt.Brokers
|
||||
conf := sarama.NewConfig()
|
||||
// 等待服务器所有副本都保存成功后的响应
|
||||
conf.Producer.RequiredAcks = sarama.WaitForAll
|
||||
// 随机向partition发送消息
|
||||
conf.Producer.Partitioner = sarama.NewRandomPartitioner
|
||||
// 是否等待成功和失败后的响应,只有上面的RequireAcks设置不是NoReponse这里才有用.
|
||||
conf.Producer.Return.Successes = true
|
||||
|
||||
conf.Producer.Return.Errors = true
|
||||
conf.Producer.Compression = sarama.CompressionNone
|
||||
conf.ClientID = connOpt.ClientId
|
||||
|
||||
conf.Version = kfkVersion
|
||||
if connOpt.UserName != "" {
|
||||
conf.Net.SASL.Enable = true
|
||||
conf.Net.SASL.User = connOpt.UserName
|
||||
conf.Net.SASL.Password = connOpt.Password
|
||||
}
|
||||
|
||||
mqIns.producerIns, err = sarama.NewAsyncProducer(brokers, conf)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
simple.Event().Register(consts.EventServerClose, func(ctx context.Context, args ...interface{}) {
|
||||
g.Log().Debug(ctx, "kafka producer AsyncClose...")
|
||||
mqIns.producerIns.AsyncClose()
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
// validateVersion 验证版本是否有效
|
||||
func validateVersion(version sarama.KafkaVersion) bool {
|
||||
for _, item := range sarama.SupportedVersions {
|
||||
if version.String() == item.String() {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
type KaConsumer struct {
|
||||
ready chan bool
|
||||
receiveDoFun func(mqMsg MqMsg)
|
||||
}
|
||||
|
||||
// Setup is run at the beginning of a new session, before ConsumeClaim
|
||||
func (consumer *KaConsumer) Setup(sarama.ConsumerGroupSession) error {
|
||||
// Mark the consumer as ready
|
||||
close(consumer.ready)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
|
||||
func (consumer *KaConsumer) Cleanup(sarama.ConsumerGroupSession) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
|
||||
func (consumer *KaConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
|
||||
// NOTE:
|
||||
// Do not move the code below to a goroutine.
|
||||
// The `ConsumeClaim` itself is called within a goroutine, see:
|
||||
// https://github.com/Shopify/sarama/blob/master/consumer_group.go#L27-L29
|
||||
// `ConsumeClaim` 方法已经是 goroutine 调用 不要在该方法内进行 goroutine
|
||||
for message := range claim.Messages() {
|
||||
consumer.receiveDoFun(MqMsg{
|
||||
RunType: ReceiveMsg,
|
||||
Topic: message.Topic,
|
||||
Body: message.Value,
|
||||
Offset: message.Offset,
|
||||
Timestamp: message.Timestamp,
|
||||
Partition: message.Partition,
|
||||
})
|
||||
session.MarkMessage(message, "")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
// Package queue
|
||||
// @Link https://github.com/bufanyun/hotgo
|
||||
// @Copyright Copyright (c) 2023 HotGo CLI
|
||||
// @Author Ms <133814250@qq.com>
|
||||
// @License https://github.com/bufanyun/hotgo/blob/master/LICENSE
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/os/glog"
|
||||
)
|
||||
|
||||
const (
|
||||
ConsumerLogErrFormat = "消费 [%s] 失败, body:%+v, err:%+v"
|
||||
ProducerLogErrFormat = "生产 [%s] 失败, body:%+v, err:%+v"
|
||||
)
|
||||
|
||||
func Logger() *glog.Logger {
|
||||
return g.Log("queue")
|
||||
}
|
||||
|
||||
// ConsumerLog 消费日志
|
||||
func ConsumerLog(ctx context.Context, topic string, mqMsg MqMsg, err error) {
|
||||
if err != nil {
|
||||
Logger().Errorf(ctx, ConsumerLogErrFormat, topic, string(mqMsg.Body), err)
|
||||
}
|
||||
}
|
||||
|
||||
// ProducerLog 生产日志
|
||||
func ProducerLog(ctx context.Context, topic string, mqMsg MqMsg, err error) {
|
||||
if err != nil {
|
||||
Logger().Infof(ctx, ProducerLogErrFormat, topic, string(mqMsg.Body), err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
// Package queue
|
||||
// @Link https://github.com/bufanyun/hotgo
|
||||
// @Copyright Copyright (c) 2023 HotGo CLI
|
||||
// @Author Ms <133814250@qq.com>
|
||||
// @License https://github.com/bufanyun/hotgo/blob/master/LICENSE
|
||||
package queue
|
||||
|
||||
import (
|
||||
"github.com/gogf/gf/v2/util/gconv"
|
||||
)
|
||||
|
||||
// Push 推送队列
|
||||
func Push(topic string, data interface{}) (err error) {
|
||||
q, err := InstanceProducer()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
mqMsg, err := q.SendMsg(topic, gconv.String(data))
|
||||
ProducerLog(ctx, topic, mqMsg, err)
|
||||
return
|
||||
}
|
||||
|
||||
// DelayPush 推送延迟队列
|
||||
// redis delay 传入 秒。如：10代表延迟10秒
|
||||
// rocketmq delay 传入 延迟级别。如：2代表延迟5秒
|
||||
// rocketmq reference delay level definition: 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
|
||||
// rocketmq delay level starts from 1. for example, if we set param level=1, then the delay time is 1s.
|
||||
func DelayPush(topic string, data interface{}, delay int64) (err error) {
|
||||
q, err := InstanceProducer()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
mqMsg, err := q.SendDelayMsg(topic, gconv.String(data), delay)
|
||||
ProducerLog(ctx, topic, mqMsg, err)
|
||||
return
|
||||
}
|
||||
@@ -0,0 +1,219 @@
|
||||
// Package queue
|
||||
// @Link https://github.com/bufanyun/hotgo
|
||||
// @Copyright Copyright (c) 2023 HotGo CLI
|
||||
// @Author Ms <133814250@qq.com>
|
||||
// @License https://github.com/bufanyun/hotgo/blob/master/LICENSE
|
||||
package queue
|
||||
|
||||
import (
|
||||
"github.com/gogf/gf/v2/errors/gerror"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/os/gctx"
|
||||
"hotgo/internal/library/queue/disk"
|
||||
"hotgo/utility/charset"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type MqProducer interface {
|
||||
SendMsg(topic string, body string) (mqMsg MqMsg, err error)
|
||||
SendByteMsg(topic string, body []byte) (mqMsg MqMsg, err error)
|
||||
SendDelayMsg(topic string, body string, delay int64) (mqMsg MqMsg, err error)
|
||||
}
|
||||
|
||||
type MqConsumer interface {
|
||||
ListenReceiveMsgDo(topic string, receiveDo func(mqMsg MqMsg)) (err error)
|
||||
}
|
||||
|
||||
const (
|
||||
_ = iota
|
||||
SendMsg
|
||||
ReceiveMsg
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
Switch bool `json:"switch"`
|
||||
Driver string `json:"driver"`
|
||||
GroupName string `json:"groupName"`
|
||||
Redis RedisConf
|
||||
Rocketmq RocketmqConf
|
||||
Kafka KafkaConf
|
||||
Disk *disk.Config
|
||||
}
|
||||
|
||||
type RedisConf struct {
|
||||
Timeout int64 `json:"timeout"`
|
||||
}
|
||||
|
||||
type RocketmqConf struct {
|
||||
NameSrvAdders []string `json:"nameSrvAdders"`
|
||||
AccessKey string `json:"accessKey"`
|
||||
SecretKey string `json:"secretKey"`
|
||||
BrokerAddr string `json:"brokerAddr"`
|
||||
Retry int `json:"retry"`
|
||||
LogLevel string `json:"logLevel"`
|
||||
}
|
||||
|
||||
type KafkaConf struct {
|
||||
Address []string `json:"address"`
|
||||
Version string `json:"version"`
|
||||
RandClient bool `json:"randClient"`
|
||||
MultiConsumer bool `json:"multiConsumer"`
|
||||
}
|
||||
|
||||
type MqMsg struct {
|
||||
RunType int `json:"run_type"`
|
||||
Topic string `json:"topic"`
|
||||
MsgId string `json:"msg_id"`
|
||||
Offset int64 `json:"offset"`
|
||||
Partition int32 `json:"partition"`
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
Body []byte `json:"body"`
|
||||
}
|
||||
|
||||
var (
|
||||
ctx = gctx.GetInitCtx()
|
||||
mqProducerInstanceMap map[string]MqProducer
|
||||
mqConsumerInstanceMap map[string]MqConsumer
|
||||
mutex sync.Mutex
|
||||
config Config
|
||||
)
|
||||
|
||||
func init() {
|
||||
mqProducerInstanceMap = make(map[string]MqProducer)
|
||||
mqConsumerInstanceMap = make(map[string]MqConsumer)
|
||||
if err := g.Cfg().MustGet(ctx, "queue").Scan(&config); err != nil {
|
||||
Logger().Warningf(ctx, "queue init err:%+v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// InstanceConsumer 实例化消费者
|
||||
func InstanceConsumer() (mqClient MqConsumer, err error) {
|
||||
return NewConsumer(config.GroupName)
|
||||
}
|
||||
|
||||
// InstanceProducer 实例化生产者
|
||||
func InstanceProducer() (mqClient MqProducer, err error) {
|
||||
return NewProducer(config.GroupName)
|
||||
}
|
||||
|
||||
// NewProducer 初始化生产者实例
|
||||
func NewProducer(groupName string) (mqClient MqProducer, err error) {
|
||||
if item, ok := mqProducerInstanceMap[groupName]; ok {
|
||||
return item, nil
|
||||
}
|
||||
|
||||
if groupName == "" {
|
||||
err = gerror.New("mq groupName is empty.")
|
||||
return
|
||||
}
|
||||
|
||||
switch config.Driver {
|
||||
case "rocketmq":
|
||||
if len(config.Rocketmq.NameSrvAdders) == 0 {
|
||||
err = gerror.New("queue.rocketmq.nameSrvAdders is empty.")
|
||||
return
|
||||
}
|
||||
mqClient, err = RegisterRocketProducer()
|
||||
case "kafka":
|
||||
if len(config.Kafka.Address) == 0 {
|
||||
err = gerror.New("queue kafka address is not support")
|
||||
return
|
||||
}
|
||||
mqClient, err = RegisterKafkaProducer(KafkaConfig{
|
||||
Brokers: config.Kafka.Address,
|
||||
GroupID: groupName,
|
||||
Version: config.Kafka.Version,
|
||||
})
|
||||
case "redis":
|
||||
if _, err = g.Redis().Do(ctx, "ping"); err == nil {
|
||||
mqClient = RegisterRedisMqProducer(RedisOption{
|
||||
Timeout: config.Redis.Timeout,
|
||||
}, groupName)
|
||||
}
|
||||
|
||||
case "disk":
|
||||
config.Disk.GroupName = groupName
|
||||
mqClient, err = RegisterDiskMqProducer(config.Disk)
|
||||
default:
|
||||
err = gerror.New("queue driver is not support")
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
mutex.Lock()
|
||||
defer mutex.Unlock()
|
||||
mqProducerInstanceMap[groupName] = mqClient
|
||||
return
|
||||
}
|
||||
|
||||
// NewConsumer 初始化消费者实例
|
||||
func NewConsumer(groupName string) (mqClient MqConsumer, err error) {
|
||||
if groupName == "" {
|
||||
err = gerror.New("mq groupName is empty.")
|
||||
return
|
||||
}
|
||||
|
||||
switch config.Driver {
|
||||
case "rocketmq":
|
||||
if len(config.Rocketmq.NameSrvAdders) == 0 {
|
||||
err = gerror.New("queue.rocketmq.nameSrvAdders is empty.")
|
||||
return
|
||||
}
|
||||
mqClient, err = RegisterRocketConsumer()
|
||||
case "kafka":
|
||||
if len(config.Kafka.Address) == 0 {
|
||||
err = gerror.New("queue kafka address is not support")
|
||||
return
|
||||
}
|
||||
|
||||
randTag := string(charset.RandomCreateBytes(6))
|
||||
// 是否支持创建多个消费者
|
||||
if !config.Kafka.MultiConsumer {
|
||||
randTag = "001"
|
||||
}
|
||||
|
||||
if item, ok := mqConsumerInstanceMap[groupName+"-"+randTag]; ok {
|
||||
return item, nil
|
||||
}
|
||||
|
||||
clientId := "hotgo-consumer-" + groupName
|
||||
if config.Kafka.RandClient {
|
||||
clientId += "-" + randTag
|
||||
}
|
||||
|
||||
mqClient, err = RegisterKafkaMqConsumer(KafkaConfig{
|
||||
Brokers: config.Kafka.Address,
|
||||
GroupID: groupName,
|
||||
Version: config.Kafka.Version,
|
||||
ClientId: clientId,
|
||||
})
|
||||
case "redis":
|
||||
if _, err = g.Redis().Do(ctx, "ping"); err == nil {
|
||||
mqClient = RegisterRedisMqConsumer(RedisOption{
|
||||
Timeout: config.Redis.Timeout,
|
||||
}, groupName)
|
||||
}
|
||||
case "disk":
|
||||
config.Disk.GroupName = groupName
|
||||
mqClient, err = RegisterDiskMqConsumer(config.Disk)
|
||||
default:
|
||||
err = gerror.New("queue driver is not support")
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
mutex.Lock()
|
||||
defer mutex.Unlock()
|
||||
mqConsumerInstanceMap[groupName] = mqClient
|
||||
return
|
||||
}
|
||||
|
||||
// BodyString 返回消息体
|
||||
func (m *MqMsg) BodyString() string {
|
||||
return string(m.Body)
|
||||
}
|
||||
@@ -0,0 +1,277 @@
|
||||
// Package queue
|
||||
// @Link https://github.com/bufanyun/hotgo
|
||||
// @Copyright Copyright (c) 2023 HotGo CLI
|
||||
// @Author Ms <133814250@qq.com>
|
||||
// @License https://github.com/bufanyun/hotgo/blob/master/LICENSE
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/gogf/gf/v2/database/gredis"
|
||||
"github.com/gogf/gf/v2/errors/gerror"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"hotgo/utility/encrypt"
|
||||
"math/rand"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
type RedisMq struct {
|
||||
poolName string
|
||||
groupName string
|
||||
timeout int64
|
||||
}
|
||||
|
||||
type RedisOption struct {
|
||||
Timeout int64
|
||||
}
|
||||
|
||||
// SendMsg 按字符串类型生产数据
|
||||
func (r *RedisMq) SendMsg(topic string, body string) (mqMsg MqMsg, err error) {
|
||||
return r.SendByteMsg(topic, []byte(body))
|
||||
}
|
||||
|
||||
// SendByteMsg 生产数据
|
||||
func (r *RedisMq) SendByteMsg(topic string, body []byte) (mqMsg MqMsg, err error) {
|
||||
if r.poolName == "" {
|
||||
return mqMsg, gerror.New("RedisMq producer not register")
|
||||
}
|
||||
|
||||
if topic == "" {
|
||||
return mqMsg, gerror.New("RedisMq topic is empty")
|
||||
}
|
||||
|
||||
mqMsg = MqMsg{
|
||||
RunType: SendMsg,
|
||||
Topic: topic,
|
||||
MsgId: getRandMsgId(),
|
||||
Body: body,
|
||||
Timestamp: time.Now(),
|
||||
}
|
||||
|
||||
data, err := json.Marshal(mqMsg)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
key := r.genKey(r.groupName, topic)
|
||||
if _, err = g.Redis().Do(ctx, "LPUSH", key, data); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if r.timeout > 0 {
|
||||
if _, err = g.Redis().Do(ctx, "EXPIRE", key, r.timeout); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (r *RedisMq) SendDelayMsg(topic string, body string, delaySecond int64) (mqMsg MqMsg, err error) {
|
||||
if delaySecond < 1 {
|
||||
return r.SendMsg(topic, body)
|
||||
}
|
||||
|
||||
if r.poolName == "" {
|
||||
err = gerror.New("SendDelayMsg RedisMq not register")
|
||||
return
|
||||
}
|
||||
|
||||
if topic == "" {
|
||||
err = gerror.New("SendDelayMsg RedisMq topic is empty")
|
||||
return
|
||||
}
|
||||
|
||||
mqMsg = MqMsg{
|
||||
RunType: SendMsg,
|
||||
Topic: topic,
|
||||
MsgId: getRandMsgId(),
|
||||
Body: []byte(body),
|
||||
Timestamp: time.Now(),
|
||||
}
|
||||
|
||||
data, err := json.Marshal(mqMsg)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
var (
|
||||
conn = g.Redis()
|
||||
key = r.genKey(r.groupName, "delay:"+topic)
|
||||
expireSecond = time.Now().Unix() + delaySecond
|
||||
timePiece = fmt.Sprintf("%s:%d", key, expireSecond)
|
||||
z = gredis.ZAddMember{Score: float64(expireSecond), Member: timePiece}
|
||||
)
|
||||
|
||||
if _, err = conn.ZAdd(ctx, key, &gredis.ZAddOption{}, z); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if _, err = conn.RPush(ctx, timePiece, data); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// consumer will also delete the item
|
||||
if r.timeout > 0 {
|
||||
_, _ = conn.Expire(ctx, timePiece, r.timeout+delaySecond)
|
||||
_, _ = conn.Expire(ctx, key, r.timeout)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// ListenReceiveMsgDo 消费数据
|
||||
func (r *RedisMq) ListenReceiveMsgDo(topic string, receiveDo func(mqMsg MqMsg)) (err error) {
|
||||
if r.poolName == "" {
|
||||
return gerror.New("RedisMq producer not register")
|
||||
}
|
||||
if topic == "" {
|
||||
return gerror.New("RedisMq topic is empty")
|
||||
}
|
||||
|
||||
var (
|
||||
key = r.genKey(r.groupName, topic)
|
||||
key2 = r.genKey(r.groupName, "delay:"+topic)
|
||||
)
|
||||
|
||||
go func() {
|
||||
for range time.Tick(300 * time.Millisecond) {
|
||||
mqMsgList := r.loopReadQueue(key)
|
||||
for _, mqMsg := range mqMsgList {
|
||||
receiveDo(mqMsg)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
go func() {
|
||||
mqMsgCh, errCh := r.loopReadDelayQueue(key2)
|
||||
for mqMsg := range mqMsgCh {
|
||||
receiveDo(mqMsg)
|
||||
}
|
||||
for err = range errCh {
|
||||
if err != nil && err != context.Canceled && err != context.DeadlineExceeded {
|
||||
Logger().Infof(ctx, "ListenReceiveMsgDo Delay topic:%v, err:%+v", topic, err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
select {}
|
||||
}
|
||||
|
||||
// 生成队列key
|
||||
func (r *RedisMq) genKey(groupName string, topic string) string {
|
||||
return fmt.Sprintf("queue:%s_%s", groupName, topic)
|
||||
}
|
||||
|
||||
func (r *RedisMq) loopReadQueue(key string) (mqMsgList []MqMsg) {
|
||||
conn := g.Redis()
|
||||
for {
|
||||
data, err := conn.Do(ctx, "RPOP", key)
|
||||
if err != nil {
|
||||
Logger().Warningf(ctx, "loopReadQueue redis RPOP err:%+v", err)
|
||||
break
|
||||
}
|
||||
|
||||
if data.IsEmpty() {
|
||||
break
|
||||
}
|
||||
|
||||
var mqMsg MqMsg
|
||||
if err = data.Scan(&mqMsg); err != nil {
|
||||
Logger().Warningf(ctx, "loopReadQueue Scan err:%+v", err)
|
||||
break
|
||||
}
|
||||
|
||||
if mqMsg.MsgId != "" {
|
||||
mqMsgList = append(mqMsgList, mqMsg)
|
||||
}
|
||||
}
|
||||
return mqMsgList
|
||||
}
|
||||
|
||||
func RegisterRedisMqProducer(connOpt RedisOption, groupName string) (client MqProducer) {
|
||||
return RegisterRedisMq(connOpt, groupName)
|
||||
}
|
||||
|
||||
// RegisterRedisMqConsumer 注册消费者
|
||||
func RegisterRedisMqConsumer(connOpt RedisOption, groupName string) (client MqConsumer) {
|
||||
return RegisterRedisMq(connOpt, groupName)
|
||||
}
|
||||
|
||||
// RegisterRedisMq 注册redis实例
|
||||
func RegisterRedisMq(connOpt RedisOption, groupName string) *RedisMq {
|
||||
return &RedisMq{
|
||||
poolName: encrypt.Md5ToString(fmt.Sprintf("%s-%d", groupName, time.Now().UnixNano())),
|
||||
groupName: groupName,
|
||||
timeout: connOpt.Timeout,
|
||||
}
|
||||
}
|
||||
|
||||
func getRandMsgId() string {
|
||||
rand.NewSource(time.Now().UnixNano())
|
||||
radium := rand.Intn(999) + 1
|
||||
timeCode := time.Now().UnixNano()
|
||||
return fmt.Sprintf("%d%.4d", timeCode, radium)
|
||||
}
|
||||
|
||||
func (r *RedisMq) loopReadDelayQueue(key string) (resCh chan MqMsg, errCh chan error) {
|
||||
resCh = make(chan MqMsg)
|
||||
errCh = make(chan error, 1)
|
||||
|
||||
go func() {
|
||||
defer close(resCh)
|
||||
defer close(errCh)
|
||||
|
||||
conn := g.Redis()
|
||||
for {
|
||||
now := time.Now().Unix()
|
||||
do, err := conn.Do(ctx, "zrangebyscore", key, "0", strconv.FormatInt(now, 10), "limit", 0, 1)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
val := do.Strings()
|
||||
if len(val) == 0 {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
errCh <- ctx.Err()
|
||||
return
|
||||
case <-time.After(time.Second):
|
||||
continue
|
||||
}
|
||||
}
|
||||
for _, listK := range val {
|
||||
for {
|
||||
pop, err := conn.LPop(ctx, listK)
|
||||
if err != nil {
|
||||
errCh <- err
|
||||
return
|
||||
} else if pop.IsEmpty() {
|
||||
_, _ = conn.ZRem(ctx, key, listK)
|
||||
_, _ = conn.Del(ctx, listK)
|
||||
break
|
||||
} else {
|
||||
var mqMsg MqMsg
|
||||
if err = pop.Scan(&mqMsg); err != nil {
|
||||
g.Log().Warningf(ctx, "loopReadDelayQueue Scan err:%+v", err)
|
||||
break
|
||||
}
|
||||
|
||||
if mqMsg.MsgId == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
select {
|
||||
case resCh <- mqMsg:
|
||||
case <-ctx.Done():
|
||||
errCh <- ctx.Err()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
return resCh, errCh
|
||||
}
|
||||
@@ -0,0 +1,311 @@
|
||||
// Package queue
|
||||
// @Link https://github.com/bufanyun/hotgo
|
||||
// @Copyright Copyright (c) 2023 HotGo CLI
|
||||
// @Author Ms <133814250@qq.com>
|
||||
// @License https://github.com/bufanyun/hotgo/blob/master/LICENSE
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"hotgo/internal/consts"
|
||||
"hotgo/utility/simple"
|
||||
"hotgo/utility/validate"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/apache/rocketmq-client-go/v2"
|
||||
"github.com/apache/rocketmq-client-go/v2/admin"
|
||||
"github.com/apache/rocketmq-client-go/v2/consumer"
|
||||
"github.com/apache/rocketmq-client-go/v2/primitive"
|
||||
"github.com/apache/rocketmq-client-go/v2/producer"
|
||||
"github.com/apache/rocketmq-client-go/v2/rlog"
|
||||
"github.com/gogf/gf/v2/errors/gerror"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/os/grpool"
|
||||
)
|
||||
|
||||
type RocketMq struct {
|
||||
producerIns rocketmq.Producer
|
||||
consumerIns rocketmq.PushConsumer
|
||||
}
|
||||
|
||||
type RocketManager struct {
|
||||
Producer *RocketMq
|
||||
Consumer *RocketMq
|
||||
pMutex sync.Mutex
|
||||
cMutex sync.Mutex
|
||||
goPool *grpool.Pool
|
||||
}
|
||||
|
||||
var rocketManager = &RocketManager{}
|
||||
|
||||
func init() {
|
||||
setRocketCloseEvent()
|
||||
}
|
||||
|
||||
func setRocketCloseEvent() {
|
||||
simple.Event().Register(consts.EventServerClose, func(ctx context.Context, args ...interface{}) {
|
||||
if rocketManager == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if rocketManager.Producer != nil {
|
||||
err := rocketManager.Producer.producerIns.Shutdown()
|
||||
if err != nil {
|
||||
Logger().Warningf(ctx, "rocketmq producer close err:%v", err)
|
||||
return
|
||||
}
|
||||
Logger().Debug(ctx, "rocketmq producer close...")
|
||||
}
|
||||
|
||||
if rocketManager.Consumer != nil {
|
||||
err := rocketManager.Consumer.consumerIns.Shutdown()
|
||||
if err != nil {
|
||||
Logger().Warningf(ctx, "rocketmq consumer close err:%v", err)
|
||||
return
|
||||
}
|
||||
Logger().Debug(ctx, "rocketmq consumer close...")
|
||||
}
|
||||
|
||||
for rocketManager.goPool != nil && rocketManager.goPool.Size() != 0 {
|
||||
g.Log().Debugf(ctx, "waiting for eocketmq consumer to complete execution[%v][%v]...", rocketManager.goPool.Size(), rocketManager.goPool.Jobs())
|
||||
time.Sleep(time.Second)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func GetRocketManager() *RocketManager {
|
||||
return rocketManager
|
||||
}
|
||||
|
||||
// RegisterRocketProducer 注册并启动生产者接口实现
|
||||
func RegisterRocketProducer() (client MqProducer, err error) {
|
||||
return RegisterRocketMqProducer()
|
||||
}
|
||||
|
||||
// RegisterRocketConsumer 注册消费者
|
||||
func RegisterRocketConsumer() (client MqConsumer, err error) {
|
||||
return RegisterRocketMqConsumer()
|
||||
}
|
||||
|
||||
// createTopicIfNotExists 主题不存在就自动创建
|
||||
func (r *RocketMq) createTopicIfNotExists(topic string) (err error) {
|
||||
if len(config.Rocketmq.BrokerAddr) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
client, err := admin.NewAdmin(
|
||||
admin.WithResolver(primitive.NewPassthroughResolver(config.Rocketmq.NameSrvAdders)),
|
||||
admin.WithCredentials(primitive.Credentials{
|
||||
AccessKey: config.Rocketmq.AccessKey,
|
||||
SecretKey: config.Rocketmq.SecretKey,
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
result, err := client.FetchAllTopicList(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if validate.InSlice(result.TopicList, topic) {
|
||||
return
|
||||
}
|
||||
|
||||
Logger().Debugf(ctx, "create topic:%v", topic)
|
||||
err = client.CreateTopic(ctx, admin.WithTopicCreate(topic), admin.WithBrokerAddrCreate(config.Rocketmq.BrokerAddr))
|
||||
return
|
||||
}
|
||||
|
||||
// SendMsg 按字符串类型生产数据
|
||||
func (r *RocketMq) SendMsg(topic string, body string) (mqMsg MqMsg, err error) {
|
||||
return r.SendByteMsg(topic, []byte(body))
|
||||
}
|
||||
|
||||
// SendByteMsg 生产数据
|
||||
func (r *RocketMq) SendByteMsg(topic string, body []byte) (mqMsg MqMsg, err error) {
|
||||
if r.producerIns == nil {
|
||||
return mqMsg, gerror.New("rocketMq producer not register")
|
||||
}
|
||||
|
||||
result, err := r.producerIns.SendSync(ctx, &primitive.Message{
|
||||
Topic: topic,
|
||||
Body: body,
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if result.Status != primitive.SendOK {
|
||||
return mqMsg, gerror.Newf("rocketMq producer send msg error status:%v", result.Status)
|
||||
}
|
||||
|
||||
mqMsg = MqMsg{
|
||||
RunType: SendMsg,
|
||||
Topic: topic,
|
||||
MsgId: result.MsgID,
|
||||
Body: body,
|
||||
}
|
||||
return mqMsg, nil
|
||||
}
|
||||
|
||||
func (r *RocketMq) SendDelayMsg(topic string, body string, delayTimeLevel int64) (mqMsg MqMsg, err error) {
|
||||
if r.producerIns == nil {
|
||||
return mqMsg, gerror.New("rocketMq producer not register")
|
||||
}
|
||||
|
||||
msg := primitive.NewMessage(topic, []byte(body))
|
||||
msg.WithDelayTimeLevel(int(delayTimeLevel))
|
||||
|
||||
result, err := r.producerIns.SendSync(ctx, msg)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if result.Status != primitive.SendOK {
|
||||
return mqMsg, gerror.Newf("rocketMq producer send msg error status:%v", result.Status)
|
||||
}
|
||||
|
||||
mqMsg = MqMsg{
|
||||
RunType: SendMsg,
|
||||
Topic: topic,
|
||||
MsgId: result.MsgID,
|
||||
Body: []byte(body),
|
||||
}
|
||||
return mqMsg, nil
|
||||
}
|
||||
|
||||
// ListenReceiveMsgDo 消费数据
|
||||
func (r *RocketMq) ListenReceiveMsgDo(topic string, receiveDo func(mqMsg MqMsg)) (err error) {
|
||||
if r.consumerIns == nil {
|
||||
return gerror.New("rocketMq consumer not register")
|
||||
}
|
||||
|
||||
rocketManager.cMutex.Lock()
|
||||
defer rocketManager.cMutex.Unlock()
|
||||
|
||||
if err = r.createTopicIfNotExists(topic); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = r.consumerIns.Subscribe(topic, consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
|
||||
for _, item := range msgs {
|
||||
_ = rocketManager.goPool.Add(ctx, func(ctx context.Context) {
|
||||
receiveDo(MqMsg{
|
||||
RunType: ReceiveMsg,
|
||||
Topic: item.Topic,
|
||||
MsgId: item.MsgId,
|
||||
Body: item.Body,
|
||||
})
|
||||
})
|
||||
}
|
||||
return consumer.ConsumeSuccess, nil
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if err = r.consumerIns.Start(); err != nil {
|
||||
_ = r.consumerIns.Unsubscribe(topic)
|
||||
return
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// RegisterRocketMqProducer 注册rocketmq生产者
|
||||
func RegisterRocketMqProducer() (mqIns *RocketMq, err error) {
|
||||
if rocketManager.Producer != nil {
|
||||
return rocketManager.Producer, nil
|
||||
}
|
||||
|
||||
rocketManager.pMutex.Lock()
|
||||
defer rocketManager.pMutex.Unlock()
|
||||
|
||||
if rocketManager.Producer != nil {
|
||||
return rocketManager.Producer, nil
|
||||
}
|
||||
|
||||
mqIns = new(RocketMq)
|
||||
retry := config.Rocketmq.Retry
|
||||
if retry <= 0 {
|
||||
retry = 0
|
||||
}
|
||||
|
||||
mqIns.producerIns, err = rocketmq.NewProducer(
|
||||
producer.WithNsResolver(primitive.NewPassthroughResolver(config.Rocketmq.NameSrvAdders)),
|
||||
producer.WithRetry(retry),
|
||||
producer.WithGroupName(config.GroupName),
|
||||
producer.WithCredentials(primitive.Credentials{
|
||||
AccessKey: config.Rocketmq.AccessKey,
|
||||
SecretKey: config.Rocketmq.SecretKey,
|
||||
}),
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err = mqIns.producerIns.Start(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
_, err = mqIns.producerIns.SendSync(ctx, primitive.NewMessage("hotgo-ping", []byte("1")))
|
||||
if err != nil {
|
||||
err = gerror.Newf("连通性测试不通过，请检查`queue.rocketmq.nameSrvAdders`或权限配置是否有误。err:%+v", err.Error())
|
||||
return nil, err
|
||||
}
|
||||
|
||||
SetRLogLevel()
|
||||
rocketManager.Producer = mqIns
|
||||
return rocketManager.Producer, nil
|
||||
}
|
||||
|
||||
// RegisterRocketMqConsumer 注册rocketmq消费者
|
||||
func RegisterRocketMqConsumer() (mqIns *RocketMq, err error) {
|
||||
if rocketManager.Consumer != nil {
|
||||
return rocketManager.Consumer, nil
|
||||
}
|
||||
|
||||
rocketManager.cMutex.Lock()
|
||||
defer rocketManager.cMutex.Unlock()
|
||||
|
||||
if rocketManager.Consumer != nil {
|
||||
return rocketManager.Consumer, nil
|
||||
}
|
||||
|
||||
// 利用生产者检查一下连通性
|
||||
if _, err = RegisterRocketMqProducer(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
mqIns = new(RocketMq)
|
||||
mqIns.consumerIns, err = rocketmq.NewPushConsumer(
|
||||
consumer.WithConsumerModel(consumer.Clustering),
|
||||
consumer.WithNsResolver(primitive.NewPassthroughResolver(config.Rocketmq.NameSrvAdders)),
|
||||
consumer.WithGroupName(config.GroupName),
|
||||
consumer.WithCredentials(primitive.Credentials{
|
||||
AccessKey: config.Rocketmq.AccessKey,
|
||||
SecretKey: config.Rocketmq.SecretKey,
|
||||
}),
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// 开多携程处理消费任务，可以根据业务实际情况调整该配置
|
||||
rocketManager.goPool = grpool.New(5)
|
||||
|
||||
SetRLogLevel()
|
||||
rocketManager.Consumer = mqIns
|
||||
return rocketManager.Consumer, nil
|
||||
}
|
||||
|
||||
// SetRLogLevel 设置rocketmq日志输出等级
|
||||
func SetRLogLevel() {
|
||||
level := g.Cfg().MustGet(ctx, "queue.rocketmq.logLevel", "all").String()
|
||||
rlog.SetLogLevel(level)
|
||||
}
|
||||
Reference in New Issue
Block a user