20251209

archive/bbs-go/server/internal/pkg/event/ants_test.go

@@ -0,0 +1,54 @@
+package event
+
+import (
+	"fmt"
+	"github.com/panjf2000/ants/v2"
+	"sync"
+	"testing"
+)
+
+func TestType(t *testing.T) {
+	//e1 := FollowEvent{}
+	//e2 := &FollowEvent{}
+	//fmt.Println("type:", reflect.TypeOf(e1))
+	//fmt.Println("type:", reflect.TypeOf(e2))
+	//
+	//reflect.TypeOf(e1)
+
+	//var e1 interface{}
+	//
+	//e1 = &mq.FollowEvent{}
+	//
+	//switch e1.(type) {
+	//case string:
+	//	//...
+	//case int:
+	//	//...
+	//case mq.FollowEvent:
+	//case *mq.FollowEvent:
+	//	fmt.Println("FollowEvent!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
+	//}
+	//return
+}
+
+func TestAnts(t *testing.T) {
+	var wg sync.WaitGroup
+	runTimes := 1000
+
+	// Use the pool with a function,
+	// set 10 to the capacity of goroutine pool and 1 second for expired duration.
+	p, _ := ants.NewPoolWithFunc(10, func(i interface{}) {
+		n := i.(int32)
+		fmt.Printf("run with %d\n", n)
+		wg.Done()
+	})
+	defer p.Release()
+
+	// Submit tasks one by one.
+	for i := 0; i < runTimes; i++ {
+		wg.Add(1)
+		_ = p.Invoke(int32(i))
+	}
+	wg.Wait()
+	fmt.Printf("running goroutines: %d\n", p.Running())
+}

archive/bbs-go/server/internal/pkg/event/data.go

@@ -0,0 +1,53 @@
+package event
+
+// FollowEvent 关注
+type FollowEvent struct {
+	UserId  int64 `json:"userId"`
+	OtherId int64 `json:"otherId"`
+}
+
+// UnFollowEvent 取消关注
+type UnFollowEvent struct {
+	UserId  int64 `json:"userId"`
+	OtherId int64 `json:"otherId"`
+}
+
+type TopicCreateEvent struct {
+	UserId     int64 `json:"userId"`
+	TopicId    int64 `json:"topicId"`
+	CreateTime int64 `json:"createTime"`
+}
+
+type TopicDeleteEvent struct {
+	UserId       int64 `json:"userId"`
+	TopicId      int64 `json:"topicId"`
+	DeleteUserId int64 `json:"deleteUserId"`
+}
+
+type UserLikeEvent struct {
+	UserId     int64  `json:"userId"`
+	EntityId   int64  `json:"entityId"`
+	EntityType string `json:"entityType"`
+}
+
+type UserUnLikeEvent struct {
+	UserId     int64  `json:"userId"`
+	EntityId   int64  `json:"entityId"`
+	EntityType string `json:"entityType"`
+}
+
+type UserFavoriteEvent struct {
+	UserId     int64  `json:"userId"`
+	EntityId   int64  `json:"entityId"`
+	EntityType string `json:"entityType"`
+}
+
+type CommentCreateEvent struct {
+	UserId    int64 `json:"userId"`
+	CommentId int64 `json:"commentId"`
+}
+
+type TopicRecommendEvent struct {
+	TopicId   int64 `json:"topicId"`
+	Recommend bool  `json:"recommend"`
+}

archive/bbs-go/server/internal/pkg/event/event.go

@@ -0,0 +1,68 @@
+package event
+
+import (
+	"log/slog"
+	"reflect"
+	"sync"
+
+	"github.com/panjf2000/ants/v2"
+)
+
+var (
+	m         sync.RWMutex
+	eventPool *ants.PoolWithFunc
+	handlers  map[reflect.Type][]func(i interface{})
+	// wg        sync.WaitGroup
+)
+
+func init() {
+	var err error
+	eventPool, err = ants.NewPoolWithFunc(4, dispatch, ants.WithMaxBlockingTasks(1000))
+	if err != nil {
+		slog.Error(err.Error(), slog.Any("err", err))
+	}
+	handlers = make(map[reflect.Type][]func(i interface{}))
+}
+
+func dispatch(i interface{}) {
+	handlerList := getHandlerList(i)
+	if len(handlerList) == 0 {
+		return
+	}
+	for _, handler := range handlerList {
+		handler(i)
+		// wg.Done()
+	}
+}
+
+func Send(e interface{}) {
+	if err := eventPool.Invoke(e); err != nil {
+		slog.Error(err.Error(), slog.Any("err", err))
+	} else {
+		// wg.Add(len(getHandlerList(e)))
+		// wg.Wait()
+	}
+}
+
+func RegHandler(t reflect.Type, handler func(i interface{})) {
+	m.Lock()
+	defer m.Unlock()
+
+	handlerList := handlers[t]
+	handlerList = append(handlerList, handler)
+	handlers[t] = handlerList
+}
+
+func getHandlerList(i interface{}) []func(i interface{}) {
+	m.RLock()
+	defer m.RUnlock()
+
+	t := reflect.TypeOf(i)
+	handlerList, ok := handlers[t]
+	if ok {
+		return handlerList
+	} else {
+		slog.Error("没找到任务处理器", slog.String("type", t.String()))
+		return nil
+	}
+}

archive/bbs-go/server/internal/pkg/event/event_test.go

@@ -0,0 +1,28 @@
+package event
+
+import (
+	"bbs-go/internal/models"
+	"fmt"
+	"reflect"
+	"testing"
+
+	"github.com/mlogclub/simple/common/jsons"
+	"github.com/mlogclub/simple/sqls"
+)
+
+func TestEvent(t *testing.T) {
+	//var w sync.WaitGroup
+	//w.Add(1)
+	RegHandler(reflect.TypeOf(models.User{}), func(i interface{}) {
+		fmt.Println("处理用户1")
+		fmt.Println(jsons.ToStr(i))
+	})
+	RegHandler(reflect.TypeOf(models.User{}), func(i interface{}) {
+		fmt.Println("处理用户2")
+		fmt.Println(jsons.ToStr(i))
+	})
+	Send(models.User{
+		Username: sqls.SqlNullString("test"),
+	})
+	//w.Wait()
+}