feat(greet): Add timed tasks to synchronize data

conf/config.sample.yaml

@@ -39,4 +39,5 @@ open:
 
 bleve:
   index: "./data/greet.blv"
-  setup_path: "./deploy/setup"
+  setup_path: "./deploy/setup"
+  sync_cron: "@every 10m" # 每十分钟进行一次单向数据同步

config/bleve.go

@@ -3,4 +3,5 @@ package config
 type Bleve struct {
 	Index     string `mapstructure:"index"`
 	SetupPath string `mapstructure:"setup_path"`
+	SyncCron  string `mapstructure:"sync_cron"`
 }

dao/bleve/bleve.go

@@ -7,15 +7,14 @@ import (
 	"strconv"
 	"sync"
 
-	"apihut-server/config"
 	"apihut-server/dao/mysql"
 	"apihut-server/logger"
+	"apihut-server/utils/gen"
 
 	"github.com/blevesearch/bleve/v2"
 	"github.com/pkg/errors"
 	gse "github.com/vcaesar/gse-bleve"
 	"go.uber.org/zap"
-	"gorm.io/gorm"
 )
 
 var i *index
@@ -51,15 +50,17 @@ func Init(indexPath string) error {
 	i = new(index)
 	i.greet = greetIndex
 
-	// 同步索引
-	if err = migrate(); err != nil {
+	// 同步索引，只增改不删除
+	if err = SyncFromDB(); err != nil {
 		return err
 	}
 
 	return nil
 }
 
-func migrate() error {
+var lastSyncMD5 string
+
+func SyncFromDB() error {
 	greetList, err := mysql.GetGreetList()
 	if errors.Is(gorm.ErrRecordNotFound, err) || len(greetList) == 0 {
 		if err = loadSQLFile(); err != nil {
@@ -70,6 +71,17 @@ func migrate() error {
 		return err
 	}
 
+	// 缓存上次变动的md5，避免频繁更新
+	m, _ := json.Marshal(greetList)
+	newMD5 := gen.MD5(m)
+	if lastSyncMD5 == newMD5 {
+		logger.L().Debug("md5相同，数据无变动", zap.String("md5", newMD5))
+		return nil
+	} else {
+		logger.L().Debug("md5变动，数据更新", zap.String("old.md5", lastSyncMD5), zap.String("new.md5", newMD5))
+		lastSyncMD5 = newMD5
+	}
+
 	batch := i.greet.NewBatch()
 	for _, greet := range greetList {
 		err = batch.Index(strconv.Itoa(int(greet.ID)), greet)

go.mod

@@ -63,6 +63,7 @@ require (
 	github.com/onsi/gomega v1.20.2 // indirect
 	github.com/pelletier/go-toml v1.9.5 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.5 // indirect
+	github.com/robfig/cron/v3 v3.0.1 // indirect
 	github.com/spf13/afero v1.8.2 // indirect
 	github.com/spf13/cast v1.5.0 // indirect
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect

go.sum

@@ -281,6 +281,8 @@ github.com/pkg/sftp v1.13.1/go.mod h1:3HaPG6Dq1ILlpPZRO0HVMrsydcdLt6HRDccSgb87qR
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
 github.com/rogpeppe/go-internal v1.8.0 h1:FCbCCtXNOY3UtUuHUYaghJg4y7Fd14rXifAYUAtL9R8=

main.go

@@ -4,6 +4,7 @@ import (
 	"flag"
 
 	"apihut-server/dao/bleve"
+	"apihut-server/utils/cron"
 
 	"apihut-server/config"
 	"apihut-server/dao/mysql"
@@ -55,6 +56,8 @@ func main() {
 		logger.L().DPanic("bleve panic", zap.Error(err))
 		return
 	}
+	// 开启定时任务
+	cron.Init()
 
 	_ = r.Run(config.Conf.GetSitePort())
 }

utils/cron/cron.go

@@ -0,0 +1,27 @@
+package cron
+
+import (
+	"apihut-server/config"
+	"apihut-server/dao/bleve"
+	"apihut-server/logger"
+
+	"github.com/robfig/cron/v3"
+	"go.uber.org/zap"
+)
+
+func Init() {
+	c := cron.New()
+
+	// 一句招呼数据同步
+	_, _ = c.AddFunc(config.Conf.Bleve.SyncCron, func() {
+		logger.L().Info("【定时任务】开始同步一句招呼数据...")
+		err := bleve.SyncFromDB()
+		if err != nil {
+			logger.L().Error("【定时任务】同步一句招呼数据失败", zap.Error(err))
+			return
+		}
+		logger.L().Info("【定时任务】同步一句招呼数据成功！")
+	})
+
+	c.Start()
+}

utils/gen/md5.go

@@ -0,0 +1,12 @@
+package gen
+
+import (
+	"crypto/md5"
+	"encoding/hex"
+)
+
+func MD5(str []byte) string {
+	h := md5.New()
+	h.Write(str)
+	return hex.EncodeToString(h.Sum(nil))
+}