123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111 |
- package service
- import (
- "context"
- "sync"
- "time"
- "go-common/app/job/main/reply-feed/conf"
- "go-common/app/job/main/reply-feed/dao"
- "go-common/app/job/main/reply-feed/model"
- "go-common/library/log"
- "go-common/library/net/netutil"
- "go-common/library/queue/databus"
- "go-common/library/sync/pipeline/fanout"
- "github.com/ivpusic/grpool"
- "github.com/robfig/cron"
- )
- type Service struct {
- c *conf.Config
- dao *dao.Dao
-
- cron *cron.Cron
-
- bc netutil.BackoffConfig
- statsConsumer *databus.Databus
-
- taskQ *fanout.Fanout
- uvQ *fanout.Fanout
- statQ *fanout.Fanout
- replyListQ *fanout.Fanout
- waiter sync.WaitGroup
-
- calculator *grpool.Pool
- statisticsStats [model.SlotsNum]model.StatisticsStat
- algorithmsLock sync.RWMutex
- statisticsLock sync.RWMutex
- algorithms []model.Algorithm
- }
- func New(c *conf.Config) (s *Service) {
- s = &Service{
- c: c,
- dao: dao.New(c),
- cron: cron.New(),
- bc: netutil.BackoffConfig{
- MaxDelay: 1 * time.Second,
- BaseDelay: 100 * time.Millisecond,
- Factor: 1.6,
- Jitter: 0.2,
- },
- statsConsumer: databus.New(c.Databus.Stats),
-
-
- taskQ: fanout.New("task"),
- uvQ: fanout.New("uv-task", fanout.Worker(4), fanout.Buffer(2048)),
- statQ: fanout.New("memcache", fanout.Worker(4), fanout.Buffer(2048)),
- replyListQ: fanout.New("redis", fanout.Worker(4), fanout.Buffer(2048)),
- calculator: grpool.NewPool(4, 2048),
- }
- var err error
- if err = s.loadAlgorithm(); err != nil {
- panic(err)
- }
- if err = s.loadSlots(); err != nil {
- panic(err)
- }
- go s.loadproc()
-
- s.waiter.Add(1)
- go s.statsproc()
-
-
-
- s.cron.AddFunc("@hourly", func() {
- s.persistStatistics()
- })
- s.cron.Start()
- return s
- }
- func (s *Service) loadproc() {
- for {
- time.Sleep(time.Minute)
- s.loadAlgorithm()
- s.loadSlots()
- }
- }
- func (s *Service) Ping(c context.Context) (err error) {
- return s.dao.Ping(c)
- }
- func (s *Service) Close() {
- s.statsConsumer.Close()
-
- log.Warn("consumer closed")
- s.waiter.Wait()
- s.persistStatistics()
- s.dao.Close()
- }
|