openbilibili/app/service/main/thumbup/dao/dao.go

package dao

import (
	"context"
	"time"

	"go-common/app/service/main/thumbup/conf"
	"go-common/app/service/main/thumbup/model"
	"go-common/library/cache/memcache"
	xredis "go-common/library/cache/redis"
	"go-common/library/database/tidb"
	"go-common/library/log"
	"go-common/library/queue/databus"
	"go-common/library/stat/prom"
	"go-common/library/sync/pipeline/fanout"
)

// PromError prom error
func PromError(name string) {
	prom.BusinessErrCount.Incr(name)
}

// Dao dao
type Dao struct {
	// config
	c *conf.Config
	// db
	tidb *tidb.DB
	// memcache
	mc            *memcache.Pool
	mcStatsExpire int32
	//redis
	redis                *xredis.Pool
	redisStatsExpire     int64
	redisUserLikesExpire int64
	redisItemLikesExpire int64
	// redisSortExpire int64
	// stmt
	businessesStmt        *tidb.Stmts
	likeStateStmt         *tidb.Stmts
	userLikeCountStmt     *tidb.Stmts
	itemLikeListStmt      *tidb.Stmts
	userLikeListStmt      *tidb.Stmts
	statsOriginStmt       *tidb.Stmts
	statStmt              *tidb.Stmts
	updateLikeStmt        *tidb.Stmts
	updateCountChangeStmt *tidb.Stmts
	statDbus              *databus.Databus
	likeDbus              *databus.Databus
	itemDbus              *databus.Databus
	userDbus              *databus.Databus
	cache                 *fanout.Fanout
	async                 *fanout.Fanout
	tidbAsync             *fanout.Fanout
	BusinessMap           map[string]*model.Business
	BusinessIDMap         map[int64]*model.Business
}

// New dao new
func New(c *conf.Config) (d *Dao) {
	d = &Dao{
		// config
		c: c,
		// mc
		mc:            memcache.NewPool(c.Memcache.Config),
		mcStatsExpire: int32(time.Duration(c.Memcache.StatsExpire) / time.Second),
		// redis
		redis:                xredis.NewPool(c.Redis.Config),
		redisStatsExpire:     int64(time.Duration(c.Redis.StatsExpire) / time.Second),
		redisUserLikesExpire: int64(time.Duration(c.Redis.UserLikesExpire) / time.Second),
		redisItemLikesExpire: int64(time.Duration(c.Redis.ItemLikesExpire) / time.Second),
		// db
		tidb:      tidb.NewTiDB(c.Tidb),
		statDbus:  databus.New(c.StatDatabus),
		likeDbus:  databus.New(c.LikeDatabus),
		itemDbus:  databus.New(c.ItemDatabus),
		userDbus:  databus.New(c.UserDatabus),
		cache:     fanout.New("cache", fanout.Worker(1), fanout.Buffer(10240)),
		async:     fanout.New("async", fanout.Worker(2), fanout.Buffer(10240)),
		tidbAsync: fanout.New("tidb-async", fanout.Worker(10), fanout.Buffer(10240)),
	}
	d.businessesStmt = d.tidb.Prepared(_tidbBusinessesSQL)
	d.likeStateStmt = d.tidb.Prepared(_tidbLikeMidSQL)
	d.userLikeCountStmt = d.tidb.Prepared(_tidbUserLikeCountSQL)
	d.itemLikeListStmt = d.tidb.Prepared(_tidbItemLikeListSQL)
	d.userLikeListStmt = d.tidb.Prepared(_tidbUserLikeListSQL)
	d.statsOriginStmt = d.tidb.Prepared(_tidbStatsOriginSQL)
	d.statStmt = d.tidb.Prepared(_tidbStatSQL)
	d.updateLikeStmt = d.tidb.Prepared(_tidbUpdateLikeSQL)
	d.updateCountChangeStmt = d.tidb.Prepared(_tidbupdateCountChange)
	d.loadBusiness()
	go d.loadBusinessproc()
	return d
}

// Ping check connection success.
func (d *Dao) Ping(c context.Context) (err error) {
	if err = d.pingMC(c); err != nil {
		PromError("mc:Ping")
		log.Error("d.pingMC error(%v)", err)
		return
	}
	if err = d.pingRedis(c); err != nil {
		PromError("redis:Ping")
		log.Error("d.pingRedis error(%v)", err)
		return
	}
	return
}

// Close close  resource.
func (d *Dao) Close() {
	d.async.Close()
	d.tidbAsync.Close()
	d.tidb.Close()
	d.mc.Close()
	d.redis.Close()
}

// pingMc ping memcache
func (d *Dao) pingMC(c context.Context) (err error) {
	conn := d.mc.Get(c)
	defer conn.Close()
	item := memcache.Item{Key: "ping", Value: []byte{1}, Expiration: 100}
	err = conn.Set(&item)
	return
}

// pingRedis ping redis.
func (d *Dao) pingRedis(c context.Context) (err error) {
	conn := d.redis.Get(c)
	if _, err = conn.Do("SET", "PING", "PONG"); err != nil {
		PromError("redis: ping remote")
		log.Error("remote redis: conn.Do(SET,PING,PONG) error(%v)", err)
	}
	conn.Close()
	return
}

// LoadBusiness .
func (d *Dao) loadBusiness() {
	var business []*model.Business
	var err error
	businessMap := make(map[string]*model.Business)
	businessIDMap := make(map[int64]*model.Business)
	for {
		if business, err = d.Businesses(context.TODO()); err != nil {
			time.Sleep(time.Second)
			continue
		}
		for _, b := range business {
			businessMap[b.Name] = b
			businessIDMap[b.ID] = b
		}
		d.BusinessMap = businessMap
		d.BusinessIDMap = businessIDMap
		return
	}
}

func (d *Dao) loadBusinessproc() {
	for {
		time.Sleep(time.Minute * 5)
		d.loadBusiness()
	}
}