增加当前节点本地定时缓存，修改部分潜在BUG，启动时Mongo或者redis无法正常连接时，进入启动等待

backend/database/mongo.go

@@ -1,9 +1,14 @@
 package database
 
 import (
+	"crawlab/constants"
+	"github.com/apex/log"
+	"github.com/cenkalti/backoff/v4"
 	"github.com/globalsign/mgo"
+	"github.com/globalsign/mgo/bson"
 	"github.com/spf13/viper"
 	"net"
+	"reflect"
 	"time"
 )
 
@@ -31,6 +36,28 @@ func GetGridFs(prefix string) (*mgo.Session, *mgo.GridFS) {
 	return s, gf
 }
 
+func FillNullObjectId(doc interface{}) {
+	t := reflect.TypeOf(doc)
+	if t.Kind() == reflect.Ptr {
+		t = t.Elem()
+	}
+	if t.Kind() != reflect.Struct {
+		return
+	}
+	v := reflect.ValueOf(doc)
+	for i := 0; i < t.NumField(); i++ {
+		ft := t.Field(i)
+		fv := v.Elem().Field(i)
+		val := fv.Interface()
+		switch val.(type) {
+		case bson.ObjectId:
+			if !val.(bson.ObjectId).Valid() {
+				v.FieldByName(ft.Name).Set(reflect.ValueOf(bson.ObjectIdHex(constants.ObjectIdNull)))
+			}
+		}
+	}
+}
+
 func InitMongo() error {
 	var mongoHost = viper.GetString("mongo.host")
 	var mongoPort = viper.GetString("mongo.port")
@@ -61,37 +88,16 @@ func InitMongo() error {
 			dialInfo.Password = mongoPassword
 			dialInfo.Source = mongoAuth
 		}
+		bp := backoff.NewExponentialBackOff()
+		var err error
 
-		// mongo session
-		var sess *mgo.Session
-
-		// 错误次数
-		errNum := 0
-
-		// 重复尝试连接mongo
-		for {
-			var err error
-
-			// 连接mongo
-			sess, err = mgo.DialWithInfo(&dialInfo)
-
+		err = backoff.Retry(func() error {
+			Session, err = mgo.DialWithInfo(&dialInfo)
 			if err != nil {
-				// 如果连接错误，休息1秒，错误次数+1
-				time.Sleep(1 * time.Second)
-				errNum++
-
-				// 如果错误次数超过30，返回错误
-				if errNum >= 30 {
-					return err
-				}
-			} else {
-				// 如果没有错误，退出循环
-				break
+				log.WithError(err).Warnf("waiting for connect mongo database, after %f seconds try again.", bp.NextBackOff().Seconds())
 			}
-		}
-
-		// 赋值给全局mongo session
-		Session = sess
+			return err
+		}, bp)
 	}
 	//Add Unique index for 'key'
 	keyIndex := mgo.Index{

backend/database/pubsub.go

@@ -77,6 +77,7 @@ func (r *Redis) Subscribe(ctx context.Context, consume ConsumeFunc, channel ...s
 			fmt.Println(err)
 
 			if err == nil {
+				index = 0
 				break
 			}
 			time.Sleep(5 * time.Second)

backend/database/redis.go

@@ -6,6 +6,7 @@ import (
 	"crawlab/utils"
 	"errors"
 	"github.com/apex/log"
+	"github.com/cenkalti/backoff/v4"
 	"github.com/gomodule/redigo/redis"
 	"github.com/spf13/viper"
 	"runtime/debug"
@@ -77,11 +78,15 @@ func (r *Redis) HSet(collection string, key string, value string) error {
 	}
 	return nil
 }
-
+func (r *Redis) Ping() error {
+	c := r.pool.Get()
+	defer utils.Close(c)
+	_, err2 := redis.String(c.Do("PING"))
+	return err2
+}
 func (r *Redis) HGet(collection string, key string) (string, error) {
 	c := r.pool.Get()
 	defer utils.Close(c)
-
 	value, err2 := redis.String(c.Do("HGET", collection, key))
 	if err2 != nil && err2 != redis.ErrNil {
 		log.Error(err2.Error())
@@ -167,7 +172,17 @@ func NewRedisPool() *redis.Pool {
 
 func InitRedis() error {
 	RedisClient = NewRedisClient()
-	return nil
+	b := backoff.NewExponentialBackOff()
+	b.MaxInterval = 20 * time.Second
+	err := backoff.Retry(func() error {
+		err := RedisClient.Ping()
+
+		if err != nil {
+			log.WithError(err).Warnf("waiting for redis pool active connection. will after %f seconds try  again.", b.NextBackOff().Seconds())
+		}
+		return err
+	}, b)
+	return err
 }
 
 func Pub(channel string, msg entity.NodeMessage) error {