modify redis single client

Author: champly

ctxs/component/component.mq.go

@@ -15,12 +15,13 @@ type IComponentMQ interface {
 }
 
 type ComponentMQ struct {
-	client *redis.Client
 }
 
 var (
 	componentMQ *ComponentMQ
 	lockMQ      sync.Mutex
+	client      *redis.Client
+	l           sync.Mutex
 )
 
 func NewComponentMQ() *ComponentMQ {
@@ -37,38 +38,49 @@ func NewComponentMQ() *ComponentMQ {
 	return componentMQ
 }
 
-func (m *ComponentMQ) getClient() {
-	if m.client != nil {
-		return
+func (m *ComponentMQ) Produce(queueName, value string) error {
+	if client == nil {
+		GetSingleClient()
+	}
+
+	cmd := client.LPush(context.TODO(), queueName, value)
+	_, err := cmd.Result()
+	if err != nil {
+		return fmt.Errorf("lpush %s %s fail:err:%+v", queueName, value, err)
+	}
+	return nil
+}
+
+func GetSingleClient() *redis.Client {
+	if client != nil {
+		return client
 	}
 
-	client := redis.NewClient(&redis.Options{
+	l.Lock()
+	defer l.Unlock()
+
+	if client != nil {
+		return client
+	}
+	cli := redis.NewClient(&redis.Options{
 		Addr:     configs.MQServer.Addr,
 		Password: configs.MQServer.Password,
 		DB:       configs.MQServer.DB,
 	})
 	// secret auth
 	if configs.MQServer.Auth != "" {
-		err := client.Do(context.TODO(), "AUTH", configs.MQServer.Auth).Err()
+		err := cli.Do(context.TODO(), "AUTH", configs.MQServer.Auth).Err()
 		if err != nil {
 			panic("config component mq do auth failed:" + err.Error())
 		}
 	}
-	_, err := client.Ping(context.TODO()).Result()
+	_, err := cli.Ping(context.TODO()).Result()
 	if err != nil {
 		panic("config mqserver reture err:" + err.Error())
 	}
-	klog.Infof("connect redis succ.")
-	m.client = client
-}
 
-func (m *ComponentMQ) Produce(queueName, value string) error {
-	m.getClient()
+	klog.Infof("connect redis {%s} succ.", configs.MQServer.Addr)
 
-	cmd := m.client.LPush(context.TODO(), queueName, value)
-	_, err := cmd.Result()
-	if err != nil {
-		return fmt.Errorf("lpush %s %s fail:err:%+v", queueName, value, err)
-	}
-	return nil
+	client = cli
+	return client
 }

example/config.yaml

@@ -51,7 +51,6 @@ cronserver:
 
 # mq server config
 mqserver:
-  addr: "10.13.3.2:6379"
+  addr: "localhost:6379"
   password: ""
   db: 0
-

example/main.go

@@ -22,4 +22,6 @@ func (a *App) register() {
 	a.API("/api/demo", api.Api)
 
 	a.Cron("auto.demo", cron.NewCron)
+
+	a.MQ("mq.test", cron.Receive)
 }

example/services/cron/auto.go

@@ -1,6 +1,11 @@
 package cron
 
-import "github.com/champly/hercules/ctxs"
+import (
+	"fmt"
+	"time"
+
+	"github.com/champly/hercules/ctxs"
+)
 
 type Cron struct{}
 
@@ -16,5 +21,15 @@ func (c *Cron) Handler(ctx *ctxs.Context) (err error) {
 	ctx.Log.Debug("debug")
 	ctx.Log.Warn("warn")
 
+	ctx.ToolBox.Produce("mq.test", fmt.Sprintf(`{"time":"%s"}`, time.Now().Format("2006-01-02 15:04:05")))
+
+	return nil
+}
+
+func Receive(ctx *ctxs.Context) (err error) {
+
+	ctx.Log.Info("==========Receive Mq Info=========")
+	ctx.Log.Info(ctx.GetBody())
+
 	return nil
 }

go.mod

@@ -5,12 +5,20 @@ go 1.15
 require (
 	github.com/champly/lib4go v0.0.0-20201011013828-08c00d776af2
 	github.com/gin-gonic/gin v1.6.3
-	github.com/go-redis/redis/v8 v8.3.1
-	github.com/google/gops v0.3.8
+	github.com/go-redis/redis/v8 v8.3.3-0.20201022064020-38caa12762e7
+	github.com/google/gops v0.3.13-0.20200910081257-5d514cabbb21
+	github.com/magiconair/properties v1.8.4 // indirect
+	github.com/pelletier/go-toml v1.8.1 // indirect
 	github.com/robfig/cron v1.2.0
-	github.com/sirupsen/logrus v1.6.0
-	github.com/spf13/viper v1.7.0
+	github.com/sirupsen/logrus v1.7.1-0.20200930112625-d131c24e23ba
+	github.com/spf13/afero v1.4.1 // indirect
+	github.com/spf13/cast v1.3.1 // indirect
+	github.com/spf13/jwalterweatherman v1.1.0 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/spf13/viper v1.7.2-0.20201012074520-4938331709c1
+	golang.org/x/sys v0.0.0-20201020230747-6e5568b54d1a // indirect
 	gopkg.in/go-playground/assert.v1 v1.2.1 // indirect
 	gopkg.in/go-playground/validator.v9 v9.31.0
+	gopkg.in/ini.v1 v1.62.0 // indirect
 	k8s.io/klog/v2 v2.3.0
 )