feat: 迁移送礼事件

cron/cron.go

+package cron
+
+import (
+	"git.hilo.cn/hilo-common/resource/config"
+	"hilo-group/cron/gift_cron"
+)
+
+func Init() {
+	if !config.IsMaster() {
+		return
+	}
+	gift_cron.SendGiftEventInit() // 礼物消息
+	gift_cron.GiftRemark()        // 礼物消息补偿
+}

cron/gift_cron/remark.go

+package gift_cron
+
+import (
+	"git.hilo.cn/hilo-common/domain"
+	"git.hilo.cn/hilo-common/resource/config"
+	"github.com/robfig/cron"
+	"hilo-group/domain/model/event_m"
+)
+
+// 补偿检查mark=2
+func GiftRemark() {
+	c := cron.New()
+	// 每5分钟
+	spec := "0 */5 * * * ?"
+	if !config.AppIsRelease() {
+		spec = "0 * * * * ?" // 测服1分钟
+	}
+	_ = c.AddFunc(spec, func() {
+		var model = domain.CreateModelNil()
+		unmarks, err := event_m.FetchUnMarkEvents(model, 10)
+		if err != nil {
+			model.Log.Errorf("FetchUnMarkEvents fail:%v", err)
+			return
+		}
+		for _, unmark := range unmarks {
+			model.Log.Infof("FetchUnMarkEvent start:%v", unmark.ID)
+			// 重新入库
+			if err := event_m.AddUnMarkEvent(model, &event_m.EventGiftSend{
+				Proto:         unmark.Proto,
+				Payload:       unmark.Payload,
+				MarkHiloGroup: unmark.MarkHiloGroup,
+				Mark:          unmark.Mark, // 不能影响别的服务的mark状态
+			}); err != nil {
+				model.Log.Errorf("FetchUnMarkEvent add unmark fail:%v", err)
+				continue
+			}
+			// 旧的标记已处理
+			unmark.Model = model
+			if err := unmark.MarkDone(); err != nil {
+				model.Log.Errorf("FetchUnMarkEvent mark fail:%v", err)
+			}
+			model.Log.Infof("FetchUnMarkEvent success:%v", unmark.ID)
+		}
+	})
+	c.Start()
+}

cron/gift_cron/send_gift.go

+package gift_cron
+
+import (
+	"git.hilo.cn/hilo-common/mycontext"
+	"git.hilo.cn/hilo-common/mylogrus"
+	"hilo-group/domain/service/event_s"
+	"time"
+)
+
+// 送礼事件
+func SendGiftEventInit() {
+
+	mylogrus.MyLog.Infof("SendGiftEventInit")
+	go func() {
+		ticker := time.NewTicker(time.Millisecond * 500)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				//start := time.Now()
+				myCtx := mycontext.CreateMyContext(nil)
+				// 消费送礼事件
+				if err := event_s.NewGiftSendEventService(myCtx).Consume(); err != nil {
+					myCtx.Log.Errorf("eventServcie consume fail:%v", err)
+				} else {
+					//myCtx.Log.Infof("eventServcie consume success,cost:%v", time.Now().Sub(start))
+				}
+			}
+		}
+	}()
+}

domain/event/gift_ev/send_gift.go

+package gift_ev
+
+import (
+	"git.hilo.cn/hilo-common/domain"
+	"git.hilo.cn/hilo-common/resource/mysql"
+	"hilo-group/_const/enum/gift_e"
+)
+
+var sendGiftListen = new(domain.EventBase)
+
+// 送礼事件
+type SendGiftEvent struct {
+	SendUserId       mysql.ID
+	ReceiveUserIds   []mysql.ID
+	ResGift          EventResGift
+	GiftOperateIds   []mysql.ID
+	GiftN            mysql.Num
+	SceneType        gift_e.GiftOperateSceneType
+	SceneUid         mysql.Str
+	NoDiamondConsume bool   // 不要消费钻石
+	TotalConsume     uint64 // 房间的总消费额
+}
+
+type EventResGift struct {
+	ID                mysql.ID
+	Name              mysql.Str
+	IconUrl           mysql.Str
+	SvgaUrl           mysql.Str
+	MusicUrl          mysql.Str
+	DiamondNum        mysql.Num
+	BeanNum           mysql.Num
+	ReceiveDiamondNum mysql.Num
+	Second            mysql.Num
+	N                 mysql.Num
+	GroupBroadcast    bool
+	Cp                bool
+	Together          bool
+	Status            mysql.UserYesNo
+	GiftType          mysql.Type
+}
+
+func AddSendGiftEventSync(callback func(model *domain.Model, event interface{}) error) {
+	domain.AddEventSync(sendGiftListen, callback)
+}
+
+func AddSendGiftEventAsync(callback func(model *domain.Model, event interface{}) error) {
+	domain.AddEventAsync(sendGiftListen, callback)
+}
+
+func PublishSendGiftEvent(model *domain.Model, event interface{}) error {
+	return domain.PublishEvent(sendGiftListen, model, event)
+}

domain/model/event_m/repo.go

+package event_m
+
+import "hilo-group/domain/model"
+
+func (p *EventGiftSendOffsetHiloGroup) Persistence() error {
+	return model.Persistent(p.Db, p)
+}
+
+func (p *EventGiftSend) Persistence() error {
+	return model.Persistent(p.Db, p)
+}

domain/model/event_m/send_gift.go

+package event_m
+
+import (
+	"git.hilo.cn/hilo-common/domain"
+	"git.hilo.cn/hilo-common/resource/mysql"
+	"github.com/pkg/errors"
+	"gorm.io/gorm"
+	"time"
+)
+
+// 送礼事件消息
+type EventGiftSend struct {
+	mysql.Entity
+	*domain.Model `gorm:"-"`
+	Proto         mysql.Type
+	Payload       []byte
+	Mark          mysql.YesNo
+	MarkHiloGroup mysql.YesNo
+}
+
+func (EventGiftSend) TableName() string {
+	return "event_gift_send2"
+}
+
+// 偏移值
+type EventGiftSendOffsetHiloGroup struct {
+	mysql.Entity
+	*domain.Model `gorm:"-"`
+	MarkOffset    mysql.ID
+}
+
+// 获取当前偏移值
+func Offset(model *domain.Model) (*EventGiftSendOffsetHiloGroup, error) {
+	offset := new(EventGiftSendOffsetHiloGroup)
+	if err := model.Db.WithContext(model).First(offset).Error; err != nil {
+		if err != gorm.ErrRecordNotFound {
+			model.Log.Errorf("Offset fail:%v", err)
+			return nil, err
+		}
+		// gorm.ErrRecordNotFound
+	}
+	offset.Model = model
+	return offset, nil
+}
+
+// 批量获取送礼
+func FetchEventGiftSend(model *domain.Model, limit int) ([]*EventGiftSend, *EventGiftSendOffsetHiloGroup, error) {
+	offset, err := Offset(model)
+	if err != nil {
+		return nil, nil, err
+	}
+	var events []*EventGiftSend
+	if err := model.Db.WithContext(model).Model(EventGiftSend{}).
+		Where("id > ?", offset.MarkOffset).
+		Order("id asc").Limit(limit).Find(&events).Error; err != nil {
+		model.Log.Errorf("FetchEventGiftSend fail:%v", err)
+		return nil, nil, err
+	}
+	return events, offset, nil
+}
+
+// 标记已完成
+func (p *EventGiftSend) MarkDone() error {
+	p.MarkHiloGroup = mysql.YES
+	result := p.Db.WithContext(p.Model).Model(EventGiftSend{}).Where("id = ?", p.ID).Update("mark_hilo_group", p.MarkHiloGroup).Limit(1)
+	if result.Error != nil {
+		return result.Error
+	}
+	if result.RowsAffected <= 0 {
+		p.Model.Log.Errorf("MarkDone row affeced 0")
+		return errors.New("row affect 0")
+	}
+	return nil
+}
+
+// 查询过去1小时-5分钟前未mark的事件
+// 用来补偿
+// 记得加limit
+func FetchUnMarkEvents(model *domain.Model, limit int) ([]*EventGiftSend, error) {
+	offset, err := Offset(model)
+	if err != nil {
+		return nil, err
+	}
+	t := time.Now().Add(-time.Minute * 5)
+	lt := t.Add(-time.Hour)
+	var events []*EventGiftSend
+	if err := model.Db.WithContext(model).Model(EventGiftSend{}).
+		Where("mark_hilo_group = ?", mysql.NO).
+		Where("id <= ?", offset.MarkOffset).
+		Where("created_time < ?", t).
+		Where("created_time > ?", lt).
+		Order("id asc").Limit(limit).Find(&events).Error; err != nil {
+		model.Log.Errorf("FetchUnMarkEvents fail:%v", err)
+		return nil, err
+	}
+	return events, nil
+}
+
+// 补偿加上unmark的event
+func AddUnMarkEvent(model *domain.Model, event *EventGiftSend) error {
+	return model.Db.WithContext(model).Create(event).Error
+}

domain/service/event_s/event_init.go

@@ -8,6 +8,7 @@ import (
 	"hilo-group/_const/enum/group_e"
 	"hilo-group/_const/enum/msg_e"
 	"hilo-group/_const/enum/task_e"
+	"hilo-group/domain/event/gift_ev"
 	"hilo-group/domain/event/group_ev"
 	"hilo-group/domain/event/group_power_ev"
 	"hilo-group/domain/model/diamond_m"
@@ -27,6 +28,7 @@ func EventInit() {
 	GroupEvents()
 	GroupImMass()
 	GroupTheme()
+	SendGift()
 }
 
 func GroupSupportEvents() {
@@ -393,3 +395,16 @@ func FlushHiloInfo(extId string, isVip bool, isPrettyCode bool, medals []uint32,
 	}
 	return nil
 }
+
+func SendGift() {
+	// 送礼事件-火箭
+	gift_ev.AddSendGiftEventAsync(func(model *domain.Model, event interface{}) error {
+		sendGiftEvent, ok := event.(*gift_ev.SendGiftEvent)
+		if !ok {
+			model.Log.Errorf("AddSendGiftEventAsync event type err")
+			return nil
+		}
+		model.Log.Infof("AddSendGiftEventAsync %+v", sendGiftEvent)
+		return nil
+	})
+}

domain/service/event_s/send_gift.go

+package event_s
+
+import (
+	"encoding/json"
+	"git.hilo.cn/hilo-common/domain"
+	"git.hilo.cn/hilo-common/mycontext"
+	"git.hilo.cn/hilo-common/resource/mysql"
+	"hilo-group/domain/event/gift_ev"
+	"hilo-group/domain/model/event_m"
+	"runtime/debug"
+)
+
+// 每次处理500条
+const BatchCount = 500
+
+type GiftSendEventService struct {
+	svc *domain.Service
+}
+
+func NewGiftSendEventService(myContext *mycontext.MyContext) *GiftSendEventService {
+	svc := domain.CreateService(myContext)
+	return &GiftSendEventService{svc}
+}
+
+//
+func (s *GiftSendEventService) Consume() error {
+	defer func() {
+		if err := recover(); err != nil {
+			s.svc.Log.Errorf("ExceptionHandle GiftSendEventService Consume SYSTEM ACTION PANIC: %v, stack: %v", err, string(debug.Stack()))
+		}
+	}()
+	var model = domain.CreateModel(s.svc.CtxAndDb)
+	return model.Transaction(func(model *domain.Model) error {
+		events, offset, err := event_m.FetchEventGiftSend(model, BatchCount)
+		if err != nil {
+			return err
+		}
+		for k := range events {
+			cpEvent := &event_m.EventGiftSend{
+				Entity: mysql.Entity{
+					ID:          events[k].ID,
+					CreatedTime: events[k].CreatedTime,
+					UpdatedTime: events[k].UpdatedTime,
+				},
+				Proto:         events[k].Proto,
+				Payload:       events[k].Payload,
+				MarkHiloGroup: events[k].MarkHiloGroup,
+			}
+			if cpEvent.MarkHiloGroup == mysql.YES {
+				model.Log.Warnf("already consume msg :%v", cpEvent)
+				continue
+			}
+			sendGiftEvent := new(gift_ev.SendGiftEvent)
+			if err := json.Unmarshal(cpEvent.Payload, sendGiftEvent); err != nil {
+				model.Log.Errorf("json msg fail,event:%v,err:%v", cpEvent, err)
+				continue
+			}
+			if err := gift_ev.PublishSendGiftEvent(model, sendGiftEvent); err != nil {
+				model.Log.Errorf("PublishSendGiftEvent fail,event:%v,err:%v", cpEvent, err)
+				return err
+			}
+			// 标记已经处理
+			cpEvent.Model = model
+			err = cpEvent.MarkDone()
+			if err != nil {
+				model.Log.Errorf("consume msg fail,event:%v,err:%v", cpEvent, err)
+				return err
+			}
+		}
+		if err != nil {
+			model.Log.Errorf("batch consume msg has fail,event,err:%v", err)
+			// 暂时先允许丢数据,继续mark offset
+		}
+		// 最后一次提交offset
+		if len(events) > 0 {
+			offset.MarkOffset = events[len(events)-1].ID
+			return offset.Persistence()
+		}
+		return nil
+	})
+}

go.mod

@@ -83,6 +83,7 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.1 // indirect
 	github.com/rifflock/lfshook v0.0.0-20180920164130-b9218ef580f5 // indirect
+	github.com/robfig/cron v1.2.0 // indirect
 	github.com/spf13/cast v1.5.0 // indirect
 	github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/common v1.0.479 // indirect
 	github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/ims v1.0.479 // indirect

go.sum

@@ -282,6 +282,8 @@ github.com/posener/complete v1.2.3/go.mod h1:WZIdtGGp+qx0sLrYKtIRAruyNpv6hFCicSg
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/rifflock/lfshook v0.0.0-20180920164130-b9218ef580f5 h1:mZHayPoR0lNmnHyvtYjDeq0zlVHn9K/ZXoy17ylucdo=
 github.com/rifflock/lfshook v0.0.0-20180920164130-b9218ef580f5/go.mod h1:GEXHk5HgEKCvEIIrSpFI3ozzG5xOKA2DVlEX/gGnewM=
+github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ=
+github.com/robfig/cron v1.2.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=

main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"fmt"
 	"git.hilo.cn/hilo-common/resource/consul"
+	"hilo-group/cron"
 	"hilo-group/domain/service/event_s"
 	"hilo-group/route"
 )
@@ -14,7 +15,7 @@ const (
 )
 
 func main() {
-	//cron.Init()                     // 开启定时任务
+	cron.Init()                                              // 开启定时任务
 	event_s.EventInit()                                      // 注册事件(内部事件+mysql拟kafka)
 	r := route.InitRouter()                                  // 注册路由
 	consul.RegisterToConsul(PORT, RegisterName, RegisterTag) // 服务注册