go高并发之路——数据聚合处理

数据聚合处理，指的是在某个请求或者脚本处理中，我们不会把这个数据立刻响应给前端或者立刻发送给下游，而是对数据先进行聚合处理一下，等到达某个阈值（时间或者量级），再响应给前端或者发送给下游。

举个实际的业务场景：直播间有一个做任务的功能，用户满足购买了多少金额我们就会给该用户发放一些奖励。此时如果有用户在购买了某些商品，那么我们得发送IM消息实时告知用户，该任务还差多少金额。那么这种情况下我们是不是每个用户下了一笔订单，我们就往直播间推送一次消息呢？

很显然不是，先讲一个直播间常用的IM消息推送逻辑：一般我们不会用户下了一笔订单，我们就往直播间推送一次消息，因为消息IM推送是有频率限制的。想想看，一个直播间如果有30w用户，下单QPS如果有1W，那么那一秒就要往直播间推送1w次，那肯定是不行的，量太大了，等下把IM服务都打爆了。那么如何解决这个问题呢？

答案就是要对数据做聚合处理了，我们一般一个直播间都会生成一个RoomId，然后推送消息推送到直播间。这时候我们可以以任务ID为key值（假设一个任务ID只属于一个直播间），将某段时间比如1秒钟内的用户订单数据进行聚合，然后再一起发送IM给直播间。最后直播前端根据实际登录的UserId，校验是否在推送的IM消息中存在对应的信息，存在就展示和处理该用户的IM消息。

上面的方案已定，那我们去实现了，下面是使用go实现的一个简单的demo：

type UserOrders struct {
	RoomId   string      `json:"room_id"`
	TaskId   int64       `json:"task_id"`
	UserList []*UserList `json:"user_list"`
}
//用户list
type UserList struct {
	UserId     string `json:"user_id"`
	OrderMoney int    `json:"order_money"`
}

var UserOrdersMap sync.Map //定义sync.Map，存储每个任务的用户订单信息

const cacheKey = "TaskIm-%d" //sync.Map键值

//入口
func main() { 
	ctx := context.Background()
	rand.Seed(time.Now().UnixNano()) // 设置随机数种子
	randNum := rand.Intn(1000) + 1
	taskId := int64(randNum)                                    //假数据，1000个任务
	roomId := fmt.Sprintf("room_%d", randNum)                   //假数据，1000个直播间
	userId := fmt.Sprintf("user_%d", randNum)                   //假数据，1000个用户ID
	orderMoney := randNum                                       //假数据，用户的购买金额数
	SendUserOrderMoney(ctx, taskId, roomId, userId, orderMoney) //往直播间发送IM消息

}

//往直播间发送IM消息
func SendUserOrderMoney(ctx context.Context, taskId int64, roomId string, userId string, orderMoney int) {
	key := fmt.Sprintf(cacheKey, taskId)
	userInfo := &UserList{
		UserId:     userId,
		OrderMoney: orderMoney,
	}
	if v, ok := UserOrdersMap.Load(key); ok { // key存在
		userOrders := v.(*UserOrders)
		userOrders.UserList = append(userOrders.UserList, userInfo) //往用户列表中追加用户订单信息
		UserOrdersMap.Store(key, userOrders)
		if len(userOrders.UserList) > 100 { //用户list大于100个用户时，发送IM消息
			SendOrder(ctx, taskId) //发送IM消息
			return
		}
	} else { // key不存在
		userOrders := &UserOrders{
			RoomId:   roomId,
			TaskId:   taskId,
			UserList: []*UserList{userInfo},
		}
		UserOrdersMap.Store(key, userOrders)
	}
	go func() {
		_ = TimerDeal(ctx, taskId)
	}()

}

//定时器
func TimerDeal(ctx context.Context, taskId int64) (err error) {
	t := time.NewTimer(time.Second * 1) // 设置定时器1秒钟后执行
	for {
		select {
		case <-t.C:
			SendOrder(ctx, taskId)
			return
		}
	}
}

//发送IM消息
func SendOrder(ctx context.Context, taskId int64) {
	key := fmt.Sprintf(cacheKey, taskId)
	if data, ok := UserOrdersMap.Load(key); ok {
		defer UserOrdersMap.Delete(key) // 删除key
		res := data.(*UserOrders)       //获取值
		fmt.Println("发送IM消息", res)      //发送IM消息
	}
}

上面这段demo是使用了sync.Map来存储任务维度的IM数据，主要是考虑到有多个协程或线程在往Map里面写数据。当然，我们也可以考虑使用其它方式，比如Redis当存储介质。然后这段代码实现的逻辑就是当某个任务的用户list大于100或者时间大于1s时，我们就开始发送IM消息给直播间前端。

这就是一个数据聚合的一个简单案例，当然数据聚合还有很多的应用场景，比如我们要往表里插入数据，如果高并发的场景下，一条条插入会对DB造成较大的压力，这时候也可以考虑使用这种聚合方式，将数据聚合起来（比如到100条），再一起insert。