chore: 并入 iotlcm-admin-svr 历史到 admin-svr/（原提交 49818c6）
This commit is contained in:
MiFaZhan
@@ -0,0 +1,127 @@
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gogf/gf/v2/container/gqueue"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/net/gtrace"
|
||||
"github.com/gogf/gf/v2/os/gctx"
|
||||
"github.com/gogf/gf/v2/os/grpool"
|
||||
)
|
||||
|
||||
type tMemoryQueue struct {
|
||||
q *gqueue.Queue
|
||||
p Process
|
||||
grpool *grpool.Pool
|
||||
}
|
||||
|
||||
var (
|
||||
memRwLock sync.RWMutex
|
||||
memQueues = make(map[string]*tMemoryQueue)
|
||||
)
|
||||
|
||||
func RegisterMemoryQueue(ctx context.Context, topic string, p Process, config *Config) *tMemoryQueue {
|
||||
memRwLock.Lock()
|
||||
defer memRwLock.Unlock()
|
||||
q, ok := memQueues[topic]
|
||||
if !ok {
|
||||
q = &tMemoryQueue{
|
||||
q: gqueue.New(),
|
||||
p: p,
|
||||
}
|
||||
memQueues[topic] = q
|
||||
}
|
||||
workers := 1
|
||||
if config != nil && config.Workers > 0 {
|
||||
workers = config.Workers
|
||||
}
|
||||
if p != nil {
|
||||
q.p = p
|
||||
q.grpool = grpool.New(workers)
|
||||
go q.Handle(ctx)
|
||||
}
|
||||
g.Log().Infof(ctx, "RegisterMemoryQueue %s success, workers: %d, process: %v", topic, workers, p)
|
||||
return q
|
||||
}
|
||||
|
||||
func UnregisterMemoryQueue(ctx context.Context, topic string) {
|
||||
memRwLock.Lock()
|
||||
defer memRwLock.Unlock()
|
||||
q, ok := memQueues[topic]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
q.q.Close()
|
||||
delete(memQueues, topic)
|
||||
}
|
||||
|
||||
func (q *tMemoryQueue) GetType() QueueType {
|
||||
return QueueTypeMemory
|
||||
}
|
||||
|
||||
func (q *tMemoryQueue) GetTopic() string {
|
||||
return q.p.GetTopic()
|
||||
}
|
||||
|
||||
func (q *tMemoryQueue) Push(ctx context.Context, topic string, data interface{}, expire int64) (err error) {
|
||||
jsonData, err := json.Marshal(data)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
payload := Payload{
|
||||
Ts: time.Now().UnixMilli(),
|
||||
TraceId: gctx.CtxId(ctx),
|
||||
Topic: topic,
|
||||
Data: json.RawMessage(jsonData),
|
||||
Expire: expire,
|
||||
}
|
||||
q.q.Push(payload)
|
||||
return
|
||||
}
|
||||
|
||||
func (q *tMemoryQueue) Handle(ctx context.Context) (err error) {
|
||||
if q.grpool == nil {
|
||||
g.Log().Errorf(ctx, "grpool is nil")
|
||||
return errors.New("grpool is nil")
|
||||
}
|
||||
p := q.p
|
||||
if p == nil {
|
||||
g.Log().Errorf(ctx, "process is nil")
|
||||
return errors.New("process is nil")
|
||||
}
|
||||
|
||||
for {
|
||||
|
||||
data := q.q.Pop()
|
||||
if data == nil {
|
||||
break
|
||||
}
|
||||
payload, ok := data.(Payload)
|
||||
if !ok {
|
||||
g.Log().Errorf(ctx, "memqueue process unmarshal error: %v", data)
|
||||
continue
|
||||
}
|
||||
subCtx := gctx.New()
|
||||
gtrace.WithTraceID(subCtx, payload.TraceId)
|
||||
q.grpool.Add(ctx, func(ctx context.Context) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
g.Log().Errorf(ctx, "memqueue process panic: %v, topic: %s, trace_id: %s, hex(data): %v",
|
||||
r, q.p.GetTopic(), payload.TraceId, hex.EncodeToString(payload.Data))
|
||||
}
|
||||
}()
|
||||
err := p.Handle(subCtx, payload)
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "memqueue process error: %v, topic: %s, trace_id: %s, hex(data): %v",
|
||||
err, q.p.GetTopic(), payload.TraceId, hex.EncodeToString(payload.Data))
|
||||
}
|
||||
})
|
||||
}
|
||||
return
|
||||
}
|
||||
@@ -0,0 +1,177 @@
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"sync"
|
||||
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
)
|
||||
|
||||
var ErrNotFoundTopic = errors.New("not found topic")
|
||||
|
||||
type Payload struct {
|
||||
Ts int64 `json:"ts"` // 毫秒时间戳
|
||||
TraceId string `json:"trace_id"` // 链路ID
|
||||
Topic string `json:"topic"` // 消费主题 队列名
|
||||
Data json.RawMessage `json:"data"` // 数据
|
||||
Expire int64 `json:"expire"` // 过期时长毫秒 0:不过期 业务内自行处理
|
||||
ClientId string `json:"client_id"` // 客户端ID
|
||||
}
|
||||
|
||||
type Config struct {
|
||||
RedisName string `json:"redis_name"` // redis名称
|
||||
Workers int `json:"workers"` // 工作线程数
|
||||
ClientId string `json:"client_id"` // 客户端ID
|
||||
}
|
||||
|
||||
type Process interface {
|
||||
GetTopic() string // 获取消费主题
|
||||
GetType() QueueType // 获取类型
|
||||
Handle(ctx context.Context, p Payload) (err error) // 处理过程的方法
|
||||
}
|
||||
|
||||
type QueueType string
|
||||
|
||||
const (
|
||||
QueueTypeMemory QueueType = "memory_queue"
|
||||
QueueTypeRedis QueueType = "redis_queue"
|
||||
QueueTypeRedisBroadcast QueueType = "redis_broadcast"
|
||||
)
|
||||
|
||||
type sProducer interface {
|
||||
GetType() QueueType // 获取类型
|
||||
GetTopic() string // 获取主题
|
||||
Push(ctx context.Context, topic string, data interface{}, expire int64) (err error) // 推送消息
|
||||
}
|
||||
|
||||
type sConsumer interface {
|
||||
GetType() QueueType
|
||||
GetTopic() string // 获取消费主题
|
||||
Handle(ctx context.Context) (err error) // 处理过程的方法
|
||||
}
|
||||
|
||||
var (
|
||||
rwLock sync.RWMutex
|
||||
producer = make(map[string]sProducer)
|
||||
consumer = make(map[string]sConsumer)
|
||||
)
|
||||
|
||||
func NewProducer(ctx context.Context, t QueueType, topic string, config *Config) error {
|
||||
if config == nil {
|
||||
config = &Config{}
|
||||
}
|
||||
if _, ok := producer[topic]; ok {
|
||||
return errors.New("producer already exists")
|
||||
}
|
||||
var p sProducer
|
||||
switch t {
|
||||
case QueueTypeMemory:
|
||||
p = RegisterMemoryQueue(ctx, topic, nil, config)
|
||||
case QueueTypeRedis:
|
||||
p = RegisterRedisQueue(ctx, topic, nil, config)
|
||||
case QueueTypeRedisBroadcast:
|
||||
p = RegisterRedisBroadcast(ctx, topic, nil, config)
|
||||
}
|
||||
rwLock.Lock()
|
||||
producer[topic] = p
|
||||
rwLock.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func NewConsumer(ctx context.Context, p Process, config *Config) error {
|
||||
if config == nil {
|
||||
config = &Config{}
|
||||
}
|
||||
if _, ok := consumer[p.GetTopic()+"__"+config.ClientId]; ok {
|
||||
return errors.New("consumer already exists")
|
||||
}
|
||||
var c sConsumer
|
||||
switch p.GetType() {
|
||||
case QueueTypeMemory:
|
||||
c = RegisterMemoryQueue(ctx, p.GetTopic(), p, config)
|
||||
case QueueTypeRedis:
|
||||
c = RegisterRedisQueue(ctx, p.GetTopic(), p, config)
|
||||
case QueueTypeRedisBroadcast:
|
||||
c = RegisterRedisBroadcast(ctx, p.GetTopic(), p, config)
|
||||
}
|
||||
rwLock.Lock()
|
||||
consumer[p.GetTopic()+"__"+config.ClientId] = c
|
||||
rwLock.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func UnregisterProducer(ctx context.Context, topic string) error {
|
||||
rwLock.Lock()
|
||||
defer rwLock.Unlock()
|
||||
p, ok := producer[topic]
|
||||
if !ok {
|
||||
return errors.New("producer not found")
|
||||
}
|
||||
switch p.GetType() {
|
||||
case QueueTypeMemory:
|
||||
UnregisterMemoryQueue(ctx, topic)
|
||||
case QueueTypeRedis:
|
||||
UnregisterRedisQueue(ctx, topic)
|
||||
case QueueTypeRedisBroadcast:
|
||||
UnregisterRedisBroadcast(ctx, topic, "")
|
||||
}
|
||||
delete(producer, topic)
|
||||
return nil
|
||||
}
|
||||
|
||||
func UnregisterConsumer(ctx context.Context, topic string, groupId string) error {
|
||||
rwLock.Lock()
|
||||
defer rwLock.Unlock()
|
||||
c, ok := consumer[topic+"__"+groupId]
|
||||
if !ok {
|
||||
return ErrNotFoundTopic
|
||||
}
|
||||
switch c.GetType() {
|
||||
case QueueTypeMemory:
|
||||
UnregisterMemoryQueue(ctx, topic)
|
||||
case QueueTypeRedis:
|
||||
UnregisterRedisQueue(ctx, topic)
|
||||
case QueueTypeRedisBroadcast:
|
||||
UnregisterRedisBroadcast(ctx, topic, groupId)
|
||||
}
|
||||
delete(consumer, topic+"__"+groupId)
|
||||
return nil
|
||||
}
|
||||
|
||||
func ProducerTopicExists(topic string) bool {
|
||||
rwLock.RLock()
|
||||
_, ok := producer[topic]
|
||||
rwLock.RUnlock()
|
||||
return ok
|
||||
}
|
||||
|
||||
func ConsumerTopicExists(topic string) bool {
|
||||
rwLock.RLock()
|
||||
_, ok := consumer[topic]
|
||||
rwLock.RUnlock()
|
||||
return ok
|
||||
}
|
||||
|
||||
func Push(ctx context.Context, topic string, data interface{}) (err error) {
|
||||
rwLock.RLock()
|
||||
p, ok := producer[topic]
|
||||
rwLock.RUnlock()
|
||||
if !ok {
|
||||
g.Log().Errorf(ctx, "topic not found. topic: %s", topic)
|
||||
return ErrNotFoundTopic
|
||||
}
|
||||
return p.Push(ctx, topic, data, 0)
|
||||
}
|
||||
|
||||
func PushWithExpire(ctx context.Context, topic string, data interface{}, expire int64) (err error) {
|
||||
rwLock.RLock()
|
||||
p, ok := producer[topic]
|
||||
rwLock.RUnlock()
|
||||
if !ok {
|
||||
g.Log().Errorf(ctx, "topic not found. topic: %s", topic)
|
||||
return ErrNotFoundTopic
|
||||
}
|
||||
return p.Push(ctx, topic, data, expire)
|
||||
}
|
||||
@@ -0,0 +1,100 @@
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
_ "github.com/gogf/gf/contrib/nosql/redis/v2"
|
||||
"github.com/gogf/gf/v2/os/gctx"
|
||||
)
|
||||
|
||||
type testMemoryConsumer struct {
|
||||
}
|
||||
|
||||
func (c *testMemoryConsumer) GetTopic() string {
|
||||
return "test"
|
||||
}
|
||||
|
||||
func (c *testMemoryConsumer) GetType() QueueType {
|
||||
return QueueTypeMemory
|
||||
}
|
||||
|
||||
func (c *testMemoryConsumer) Handle(ctx context.Context, p Payload) error {
|
||||
fmt.Println("testMemoryConsumer", p)
|
||||
return nil
|
||||
}
|
||||
|
||||
type testRedisConsumer struct {
|
||||
QueueType QueueType
|
||||
}
|
||||
|
||||
func (c *testRedisConsumer) GetTopic() string {
|
||||
return "test"
|
||||
}
|
||||
|
||||
func (c *testRedisConsumer) GetType() QueueType {
|
||||
return c.QueueType
|
||||
}
|
||||
|
||||
func (c *testRedisConsumer) Handle(ctx context.Context, p Payload) error {
|
||||
fmt.Printf("testRedisConsumer %+v \n", p)
|
||||
var data string
|
||||
json.Unmarshal(p.Data, &data)
|
||||
fmt.Printf("testRedisConsumer %+v \n", data)
|
||||
|
||||
time.Sleep(1 * time.Second)
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestQueue(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
NewProducer(ctx, QueueTypeMemory, "test", nil)
|
||||
NewConsumer(ctx, &testMemoryConsumer{}, nil)
|
||||
Push(ctx, "test", "content1")
|
||||
Push(ctx, "test", "content2")
|
||||
Push(ctx, "test", "content3")
|
||||
time.Sleep(1 * time.Second)
|
||||
}
|
||||
|
||||
func TestRedisQueue(t *testing.T) {
|
||||
ctx := gctx.New()
|
||||
|
||||
NewProducer(ctx, QueueTypeRedis, "test", &Config{
|
||||
RedisName: "default",
|
||||
})
|
||||
NewConsumer(ctx, &testRedisConsumer{QueueType: QueueTypeRedis}, &Config{
|
||||
RedisName: "default",
|
||||
})
|
||||
time.Sleep(2 * time.Second)
|
||||
Push(ctx, "test", "content1")
|
||||
Push(ctx, "test", "content2")
|
||||
Push(ctx, "test", "content3")
|
||||
time.Sleep(5 * time.Second)
|
||||
}
|
||||
|
||||
func TestRedisBroadcast(t *testing.T) {
|
||||
ctx := gctx.New()
|
||||
NewProducer(ctx, QueueTypeRedisBroadcast, "test", &Config{
|
||||
RedisName: "default",
|
||||
})
|
||||
consumerHandle := &testRedisConsumer{QueueType: QueueTypeRedisBroadcast}
|
||||
NewConsumer(ctx, consumerHandle, &Config{
|
||||
RedisName: "default",
|
||||
ClientId: "group1",
|
||||
})
|
||||
NewConsumer(ctx, consumerHandle, &Config{
|
||||
RedisName: "default",
|
||||
ClientId: "group2",
|
||||
})
|
||||
Push(ctx, "test", "content1")
|
||||
Push(ctx, "test", "content2")
|
||||
Push(ctx, "test", "content3")
|
||||
time.Sleep(50 * time.Second)
|
||||
UnregisterConsumer(ctx, "test", "group1")
|
||||
UnregisterConsumer(ctx, "test", "group2")
|
||||
UnregisterProducer(ctx, "test")
|
||||
time.Sleep(5 * time.Second)
|
||||
}
|
||||
@@ -0,0 +1,173 @@
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.5iot.com/iotlcm-public/iotlcm-admin-svr/utility/gfredis"
|
||||
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/net/gtrace"
|
||||
"github.com/gogf/gf/v2/os/gctx"
|
||||
"github.com/gogf/gf/v2/os/grpool"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
type tRedisBroadcast struct {
|
||||
p Process
|
||||
RedisName string
|
||||
isRunning bool
|
||||
grpool *grpool.Pool
|
||||
clientId string
|
||||
pubSub *redis.PubSub
|
||||
}
|
||||
|
||||
var (
|
||||
redisBroadcastRwLock sync.RWMutex
|
||||
redisBroadcastQueues = make(map[string]*tRedisBroadcast)
|
||||
)
|
||||
|
||||
func RegisterRedisBroadcast(ctx context.Context, topic string, p Process, config *Config) *tRedisBroadcast {
|
||||
redisBroadcastRwLock.Lock()
|
||||
defer redisBroadcastRwLock.Unlock()
|
||||
q, ok := redisBroadcastQueues[topic+"__"+config.ClientId]
|
||||
if ok {
|
||||
q.p = p
|
||||
q.RedisName = config.RedisName
|
||||
q.isRunning = true
|
||||
q.clientId = config.ClientId
|
||||
} else {
|
||||
q = &tRedisBroadcast{
|
||||
p: p,
|
||||
RedisName: config.RedisName,
|
||||
isRunning: true,
|
||||
clientId: config.ClientId,
|
||||
}
|
||||
redisBroadcastQueues[topic+"__"+config.ClientId] = q
|
||||
}
|
||||
workers := 1
|
||||
if config != nil && config.Workers > 0 {
|
||||
workers = config.Workers
|
||||
}
|
||||
if p != nil {
|
||||
q.p = p
|
||||
q.grpool = grpool.New(workers)
|
||||
|
||||
q.pubSub, _ = q.getConn(ctx)
|
||||
go q.Handle(ctx)
|
||||
}
|
||||
return q
|
||||
}
|
||||
|
||||
func UnregisterRedisBroadcast(ctx context.Context, topic string, clientId string) {
|
||||
redisBroadcastRwLock.Lock()
|
||||
defer redisBroadcastRwLock.Unlock()
|
||||
q, ok := redisBroadcastQueues[topic+"__"+clientId]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
q.isRunning = false
|
||||
delete(redisBroadcastQueues, topic+"__"+clientId)
|
||||
}
|
||||
|
||||
func (q *tRedisBroadcast) GetType() QueueType {
|
||||
return QueueTypeRedis
|
||||
}
|
||||
|
||||
func (q *tRedisBroadcast) GetTopic() string {
|
||||
return q.p.GetTopic()
|
||||
}
|
||||
|
||||
func (q *tRedisBroadcast) Push(ctx context.Context, topic string, data interface{}, expire int64) (err error) {
|
||||
jsonData, err := json.Marshal(data)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
payload := Payload{
|
||||
Ts: time.Now().UnixMilli(),
|
||||
TraceId: gctx.CtxId(ctx),
|
||||
Topic: topic,
|
||||
Data: json.RawMessage(jsonData),
|
||||
Expire: expire,
|
||||
}
|
||||
payloadJson, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
num, err := gfredis.Instance(q.RedisName).Publish(ctx, topic, payloadJson)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
g.Log().Debugf(ctx, "redisqueue Push %s %d", topic, num)
|
||||
return
|
||||
}
|
||||
|
||||
func (q *tRedisBroadcast) getConn(ctx context.Context) (conn *redis.PubSub, err error) {
|
||||
|
||||
conn, err = gfredis.Instance(q.RedisName).Subscribe(ctx, q.p.GetTopic())
|
||||
if err != nil || conn == nil {
|
||||
g.Log().Errorf(ctx, "tRedisBroadcast.Handle Conn %s error: %v", q.p.GetTopic(), err)
|
||||
return
|
||||
}
|
||||
g.Log().Debugf(ctx, "tRedisBroadcast.Handle Subscribe %s 成功, sub:%+v", q.p.GetTopic(), conn)
|
||||
return
|
||||
}
|
||||
|
||||
func (q *tRedisBroadcast) Handle(ctx context.Context) (err error) {
|
||||
g.Log().Debugf(ctx, "tRedisBroadcast.Handle 开始处理 %s, group_id: %s", q.p.GetTopic(), q.clientId)
|
||||
topic := q.p.GetTopic()
|
||||
|
||||
for {
|
||||
if !q.isRunning {
|
||||
break
|
||||
}
|
||||
if q.pubSub == nil {
|
||||
q.pubSub, err = q.getConn(ctx)
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "tRedisBroadcast.Handle getConn %s error: %v", topic, err)
|
||||
time.Sleep(2 * time.Second)
|
||||
continue
|
||||
}
|
||||
}
|
||||
data, err := q.pubSub.ReceiveMessage(ctx)
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "tRedisBroadcast.Handle ReceiveMessage %s error: %v", topic, err)
|
||||
q.pubSub.Close()
|
||||
q.pubSub = nil
|
||||
continue
|
||||
}
|
||||
var payload Payload
|
||||
err = json.Unmarshal([]byte(data.Payload), &payload)
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "tRedisBroadcast.Handle unmarshal error: %v", data)
|
||||
continue
|
||||
}
|
||||
payload.ClientId = q.clientId
|
||||
subCtx := gctx.New()
|
||||
gtrace.WithTraceID(subCtx, payload.TraceId)
|
||||
err = q.grpool.Add(subCtx, func(ctx context.Context) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
g.Log().Errorf(ctx, "tRedisBroadcast.Handle process panic: %v, topic: %s, trace_id: %s, len(data): %d, hex(data): %v",
|
||||
r, q.p.GetTopic(), payload.TraceId, len(payload.Data), hex.EncodeToString(payload.Data))
|
||||
}
|
||||
}()
|
||||
err = q.p.Handle(subCtx, payload)
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "tRedisBroadcast.Handle process error: %v, topic: %s, trace_id: %s, len(data): %d, hex(data): %v",
|
||||
err, q.p.GetTopic(), payload.TraceId, len(payload.Data), hex.EncodeToString(payload.Data))
|
||||
return
|
||||
}
|
||||
})
|
||||
}
|
||||
redisBroadcastRwLock.Lock()
|
||||
delete(redisBroadcastQueues, topic)
|
||||
redisBroadcastRwLock.Unlock()
|
||||
if q.pubSub != nil {
|
||||
q.pubSub.Close()
|
||||
}
|
||||
return
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.5iot.com/iotlcm-public/iotlcm-admin-svr/utility/gfredis"
|
||||
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/net/gtrace"
|
||||
"github.com/gogf/gf/v2/os/gctx"
|
||||
"github.com/gogf/gf/v2/os/grpool"
|
||||
)
|
||||
|
||||
type tRedisQueue struct {
|
||||
p Process
|
||||
RedisName string
|
||||
isRunning bool
|
||||
grpool *grpool.Pool
|
||||
}
|
||||
|
||||
var (
|
||||
redisRwLock sync.RWMutex
|
||||
redisQueues = make(map[string]*tRedisQueue)
|
||||
)
|
||||
|
||||
func RegisterRedisQueue(ctx context.Context, topic string, p Process, config *Config) *tRedisQueue {
|
||||
redisRwLock.Lock()
|
||||
defer redisRwLock.Unlock()
|
||||
q, ok := redisQueues[topic]
|
||||
if ok {
|
||||
if p != nil {
|
||||
q.p = p
|
||||
}
|
||||
q.RedisName = config.RedisName
|
||||
q.isRunning = true
|
||||
} else {
|
||||
q = &tRedisQueue{
|
||||
p: p,
|
||||
RedisName: config.RedisName,
|
||||
isRunning: true,
|
||||
}
|
||||
redisQueues[topic] = q
|
||||
}
|
||||
workers := 1
|
||||
if config != nil && config.Workers > 0 {
|
||||
workers = config.Workers
|
||||
}
|
||||
if p != nil {
|
||||
q.p = p
|
||||
q.grpool = grpool.New(workers)
|
||||
go q.Handle(ctx)
|
||||
}
|
||||
return q
|
||||
}
|
||||
|
||||
func UnregisterRedisQueue(ctx context.Context, topic string) {
|
||||
redisRwLock.Lock()
|
||||
defer redisRwLock.Unlock()
|
||||
q, ok := redisQueues[topic]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
q.isRunning = false
|
||||
}
|
||||
|
||||
func (q *tRedisQueue) GetType() QueueType {
|
||||
return QueueTypeRedis
|
||||
}
|
||||
|
||||
func (q *tRedisQueue) GetTopic() string {
|
||||
return q.p.GetTopic()
|
||||
}
|
||||
|
||||
func (q *tRedisQueue) Push(ctx context.Context, topic string, data interface{}, expire int64) (err error) {
|
||||
jsonData, err := json.Marshal(data)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
payload := Payload{
|
||||
Ts: time.Now().UnixMilli(),
|
||||
TraceId: gctx.CtxId(ctx),
|
||||
Topic: topic,
|
||||
Data: json.RawMessage(jsonData),
|
||||
Expire: expire,
|
||||
}
|
||||
payloadJson, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = gfredis.Instance(q.RedisName).RPush(ctx, topic, payloadJson)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// g.Log().Debugf(ctx, "redisqueue Push %s %d", topic, num)
|
||||
return
|
||||
}
|
||||
|
||||
func (q *tRedisQueue) Handle(ctx context.Context) (err error) {
|
||||
topic := q.p.GetTopic()
|
||||
redisName := q.RedisName
|
||||
for {
|
||||
if !q.isRunning {
|
||||
break
|
||||
}
|
||||
data, err := gfredis.Instance(redisName).BLPop(ctx, 1*time.Second, topic)
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "redisqueue Pop %s error: %v", topic, err)
|
||||
time.Sleep(2 * time.Second)
|
||||
continue
|
||||
}
|
||||
arr := data.Strings()
|
||||
if len(arr) < 2 {
|
||||
continue
|
||||
}
|
||||
// g.Log().Debugf(ctx, "redisqueue Pop %s %v", topic, arr)
|
||||
var payload Payload
|
||||
err = json.Unmarshal([]byte(arr[1]), &payload)
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "redisqueue process unmarshal error: %v, data: %v, topic: %s",
|
||||
err, data, topic)
|
||||
continue
|
||||
}
|
||||
if q.p == nil {
|
||||
g.Log().Errorf(ctx, "redisqueue process is nil, topic: %s, trace_id: %s, data: %v",
|
||||
topic, payload.TraceId, payload.Data)
|
||||
continue
|
||||
}
|
||||
subCtx := gctx.New()
|
||||
gtrace.WithTraceID(subCtx, payload.TraceId)
|
||||
err = q.grpool.Add(subCtx, func(ctx context.Context) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
g.Log().Errorf(ctx, "redisqueue process panic: %v, topic: %s, trace_id: %s, len(data): %d, hex(data): %v",
|
||||
r, q.p.GetTopic(), payload.TraceId, len(payload.Data), hex.EncodeToString(payload.Data))
|
||||
}
|
||||
}()
|
||||
err = q.p.Handle(subCtx, payload)
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "redisqueue process error: %v, topic: %s, trace_id: %s, len(data): %d, hex(data): %v",
|
||||
err, q.p.GetTopic(), payload.TraceId, len(payload.Data), hex.EncodeToString(payload.Data))
|
||||
return
|
||||
}
|
||||
})
|
||||
}
|
||||
redisRwLock.Lock()
|
||||
delete(redisQueues, topic)
|
||||
redisRwLock.Unlock()
|
||||
return
|
||||
}
|
||||
Reference in New Issue
Block a user