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) + 1taskId := int64(randNum)                                    //假数据，1000个任务roomId := fmt.Sprintf("room_%d", randNum)                   //假数据，1000个直播间userId := fmt.Sprintf("user_%d", randNum)                   //假数据，1000个用户IDorderMoney := 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) // 删除keyres := data.(*UserOrders)       //获取值fmt.Println("发送IM消息", res)      //发送IM消息}
}

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

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