redis有序集合写入和求交集的速度

背景

团队小伙伴做了一个需求。大概的需求是有很多的图片作品，图片作品有一些类别，每个人进入到每个类别的作品业，根据权重优先查看权重最高的的作品，权重大概是基于每个人对该作品的浏览计算，浏览过的作品放在最后展示。小伙伴是基于redis的有序集合的方式实现的，主要用到写入和求交集这两个操作。我们这个系统的作品数量不算多，目前只有5000左右，用户也只有5000左右，这么小的数据量用这种方式实现，我原本觉得是没什么问题的。但是，实际测试下来，就很离谱，不知道他的程序里哪里导致的异常耗时。出于好奇，我写了一段测试代码，测试一下不同数量级的数据，写入和求交集的时长。

测试程序

package mainimport ("context""fmt""log""math/rand""time""github.com/redis/go-redis/v9"
)// BatchZAdd function to batch insert elements into a Redis sorted set
func BatchZAdd(ctx context.Context, rdb *redis.Client, key string, members []redis.Z) error {// Pipeline to batch the ZADD commandspipe := rdb.Pipeline()if err := pipe.ZAdd(ctx, key, members...).Err(); err != nil {return fmt.Errorf("failed to add member to sorted set: %v", err)}// Execute the pipeline_, err := pipe.Exec(ctx)if err != nil {return fmt.Errorf("failed to execute pipeline: %v", err)}return nil
}func main() {// Create a new Redis clientrdb := redis.NewClient(&redis.Options{Addr:     "10.10.37.100:6379", // Replace with your Redis server addressPassword: "dreame@2020",DB:       11,})// Define the context for the Redis operationctx := context.Background()startTime := time.Now()// Define the key for the sorted setkey := "mySortedSet|test1"// Prepare the members to be added to the sorted setmembers := make([]redis.Z, 1000000)rand.Seed(time.Now().UnixNano())for i := 0; i < len(members); i++ {// Generate a random score and member for each element in the setrandomNumber := rand.Intn(50)members[i] = redis.Z{Score: float64(randomNumber), Member: fmt.Sprintf("member%d", i)}}// Call the BatchZAdd function to batch insert the membersif err := BatchZAdd(ctx, rdb, key, members); err != nil {log.Fatalf("failed to batch insert members into sorted set: %v", err)}fmt.Println("mySortedSet|test1写入序列执行时间：", time.Since(startTime).Seconds(), "s")// 写入第二个序列startTime = time.Now()key2 := "mySortedSet|test2"members2 := make([]redis.Z, 1000000)for i := 0; i < len(members2); i++ {// Generate a random score and member for each element in the setrandomNumber := rand.Intn(50)members2[i] = redis.Z{Score: float64(randomNumber), Member: fmt.Sprintf("member%d", i)}}// Call the BatchZAdd function to batch insert the membersif err := BatchZAdd(ctx, rdb, key2, members2); err != nil {log.Fatalf("failed to batch insert members into sorted set: %v", err)}fmt.Println("mySortedSet|test2写入序列执行时间：", time.Since(startTime).Seconds(), "s")log.Println("Members successfully added to the sorted set")destinationKey := "mySortedSet|destination"cmd := rdb.ZInterStore(ctx, destinationKey, &redis.ZStore{Keys:      []string{key, key2},Aggregate: "MIN",})if cmd.Err() != nil {log.Fatalf("failed to create intersection of sorted sets: %v", cmd.Err())}startTime = time.Now()nums := int64(0)var err1 errortimeout := time.After(10 * time.Second) // 设置超时为5秒ticker := time.NewTicker(10 * time.Millisecond)defer ticker.Stop()for {select {case <-timeout:fmt.Println("Timeout reached, exiting loop.", time.Now())returncase <-ticker.C:cmd = rdb.ZCard(ctx, destinationKey)nums, err1 = cmd.Result()if err1 != nil {fmt.Println("Error getting ZCard:", err1)return // 或者其他错误处理逻辑}fmt.Println("*********************nums:*************", nums)if nums != 0 {fmt.Println("执行时间：", time.Since(startTime).Seconds(), "s")log.Println("Members successfully ZInterStore the sorted set")return // 退出循环}}}
}

 

这种写法和测试程序中的方法相比，100万一下的数据，时间稍微长了一点。但是，测试程序中的方法在100万的数据写入时就会报错了，但是，图中的方法不会报错。

	写入时长（单位：s）	求交集时长
5千	0.02	0.014
1万	0.03	0.026
10万	0.19	0.012
100万	5.22	0.015
1000万

 

我们发现，求交集的时间都还好。但是，写入时长是呈线性增长，实际执行1000万数据写入，报了超时错误，也可以理解，毕竟时间呈线性增长，如果没有超时限制，应该也需要一分钟左右。因为其写入时间复杂度是O(log(N)) for each item added, where N is the number of elements in the sorted set。而求交集的时间复杂度是O(N*K)+O(M*log(M)) worst case with N being the smallest input sorted set, K being the number of input sorted sets and M being the number of elements in the resulting sorted set.

思考

出了上文使用管道（Pipeline）的方式，还有更优的写入方案么？最先冒出来的想法是并发写入，但是，我想到了一个问题，由于有序集合涉及到排序，并发写，是否会有锁竞争的问题？甚至排序会出现问题？我们可以实际测试一下看看结果如何。

func ConcurrentBatchZAdd(ctx context.Context, rdb *redis.Client, key string, members []redis.Z, numGoroutines int) error {var wg sync.WaitGroupbatchSize := len(members) / numGoroutinesfor i := 0; i < numGoroutines; i++ {start := i * batchSizeend := start + batchSizeif i == numGoroutines-1 {end = len(members) // 最后一个 goroutine 处理剩余的成员}wg.Add(1)go func(members []redis.Z) {defer wg.Done()pipe := rdb.Pipeline()cmd := pipe.ZAdd(ctx, key, members...)_, err := pipe.Exec(ctx)if err != nil {log.Printf("failed to execute pipeline: %v", err)}if err := cmd.Err(); err != nil {log.Printf("failed to batch insert members into sorted set: %v", err)}}(members[start:end])}wg.Wait() // 等待所有 goroutine 完成return nil
}

10万及以下的数据，似乎效率没有差异。到了100万的情况，似乎有了差异，并发写100万的时长是2.3s。1000万的数据，并发写入时会报超时错误，按照我的理解，并发写入其实并不能提升写入效率。