### 使用 Redis 的 ZSET 实现多条件排序在一些需求中，经常需要实现一个排行榜。数据量较少时可以使用 RDB 数据库排序，数据量较大时可以自己实现算法或使用 NoSQL 数据库排序。利用 Redis 的 zset 来实现排行榜是非常方便的方法。以下是具体实现过程。 #### 需求示例实现一个玩家成就点数的排行榜： ```shell > zadd r 100 A "1" > zadd r 200 B "1" > zadd r 200 C "1" > zadd r 300 D "1" > zrange r 0 -1 withscores 1) "A" 2) "100" 3) "B" 4) "200" 5) "C" 6) "200" 7) "D" 8) "300" ``` 在上例中，B 和 C 的分数相同，我们希望先达到对应分数的人排在前面，这相当于增加了一个排序维度。仅用 score 无法实现，需要做一些转换。下面分享几种实现方法。 ### 实现方法假设需要三个维度的排序：具体数值、是否氪金、时间戳。氪金的（0 否，1 是）和时间早的排序靠前。 #### 原始数据 ``` 玩家成就是否氪金时间戳 A 100 1 1571819021259 B 200 0 1571819021259 C 200 1 1571819021259 D 400 0 1571819021259 E 200 1 1571810001259 ``` ### 方法一：利用 key 排序如果后两个维度初始化后不再变化，可以利用 Redis 的排序特性，将不变的维度写入 key 中。这样当 score 相同时会用 key 排序。 ```shell # 反转时间戳(9999999999999 - 1571810001259)，让时间早的排在前面 > zadd r 100 A-1-8428180978740 "1" > zadd r 200 B-0-8428180978740 "1" > zadd r 200 C-1-8428180978740 "1" > zadd r 400 D-0-8428180978740 "1" > zadd r 200 E-1-8428189998740 "1" > zrange r 0 -1 withscores 1) "A-1-8428180978740" 2) "100" 3) "B-0-8428180978740" 4) "200" 5) "C-1-8428180978740" 6) "200" 7) "E-1-8428189998740" 8) "200" 9) "D-0-8428180978740" 10) "400" ``` 此方法实现非常简单。 ### 方法二：二进制分段利用二进制 64 位 long 分段存储各个维度。首先时间戳缩小一些，忽略毫秒，和一个大数计算差值： ```java int MAX_SECOND = 1861891200; // 2029.1.1 0:0:0 int sts = (MAX_SECOND - (int)(ts / 1000)); ``` 时间戳缩小到 300000000 以内。划分 63 位：高 33 位存分数，1 位存标志位，低 29 位存时间戳。 ```java int size1 = 33; int size2 = 1; int size3 = 29; long d1Max = maxBySize(size1); long d2Max = maxBySize(size2); long d3Max = maxBySize(size3); // 计算 score public long genScore(long d1, long d2, long d3) { return ((d1 & d1Max) << (size2 + size3)) | ((d2 & d2Max) << size3) | (d3 & d3Max); } // 计算增加 value 值 public void incValue(long d1) { return ((d1 & d1Max) << (size2 + size3)) | ((0 & d2Max) << size3) | (0 & d3Max); } // 根据二进制长度计算最大值 private long maxBySize(int len) { long r = 0; while (len-- > 0) { r = ((r << 1) | 1); } return r; } ``` 增加高位分数直接将低维度设为 0，计算出分数后调用 `ZINCRBY` 即可。改变低位维度值需要调用 `ZADD`，用自旋加 CAS 更新防止覆盖掉新更新的值。 ```java public boolean setD(String key, String value, Long d2, Long d3) { long start = System.currentTimeMillis(); Long oldScore; long score; byte[] originBytes; do { if ((System.currentTimeMillis() - start) > 2000) { return false; } long d1 = 0; originBytes = zscoreBytes(key, value); oldScore = convertZscore(originBytes); if (originBytes != null) { d1 = getD1ByScore(oldScore); } if (d2 == null) { d2 = getD2ByScore(oldScore); } if (d3 == null) { d3 = getD3ByScore(oldScore); } score = genScore(d1, d2, d3); } while (!compareAndSetDimension(key, value, originBytes, score)); return true; } Long SUCCESS = 1L; String script = "if (((not (redis.call('zscore', KEYS[1], ARGV[1]))) and (ARGV[2] == '')) " + "or redis.call('zscore', KEYS[1], ARGV[1]) == ARGV[2]) " + " then redis.call('zadd',KEYS[1],ARGV[3],ARGV[1]) return 1 else return 0 end"; private boolean compareAndSetDimension(String setKey, String key, byte[] oldScore, long newScore) { Long result = redisTemplate.execute((RedisCallback) connection -> { try { return connection.eval(script.getBytes(), ReturnType.fromJavaType(Long.class), 1, setKey.getBytes(), key.getBytes(), oldScore, om.writeValueAsBytes(newScore)); } catch (JsonProcessingException e) { throw new RuntimeException(e); } }); return SUCCESS.equals(result); } // 获取原始值 public byte[] zscoreBytes(String key, String value) { String script = "return redis.call('zscore', KEYS[1], ARGV[1])"; return redisTemplate.execute((RedisCallback) connection -> connection.eval(script.getBytes(), ReturnType.fromJavaType(String.class), 1, key.getBytes(), value.getBytes())); } ``` 使用 `zscoreBytes()` 方法是为了避免精度损失，因为 Redis 返回的原始值可能会在转换为 Double 时损失精度。下面利用 `genScore` 方法生成 score 值并排序： ```java # 生成score值 # genScore(100, 1, 290072179) > zadd r 108201125491 A "1" # genScore(200, 0, 290072179) > zadd r 215038436979 B "1" # genScore(200, 1, 290072179) > zadd r 215575307891 C "1" # genScore(400, 0, 290072179) > zadd r 429786801779 D "1" # genScore(200, 1, 290081199) > zadd r 215575316911 E "1" > zrange r 0 -1 withscores 1) "A" 2) "108201125491" 3) "B" 4) "215038436979" 5) "C" 6) "215575307891" 7) "E" 8) "215575316911" 9) "D" 10) "429786801779" ``` ### 方法三：直接拆分十进制除了二进制分段外，还可以直接拆分十进制数，为了方便，可以用小数划分维度，例如将分数放在整数位，标志位和时间戳放在小数位。 ```java # 假设分数、标志位和时间戳分开存储 A 100.1290072179 B 200.0290072179 C 200.1290072179 D 400.0290072179 E 200.1290081199 ``` 原理与二进制分段相同，只是计算简单且可读性更好。以下是实现方法： ```java public double genScore(double score, int flag, int timestamp) { return score + flag * 0.1 + timestamp * 1e-10; } // 示例： double scoreA = genScore(100, 1, 290072179); // 100.1290072179 double scoreB = genScore(200, 0, 290072179); // 200.0290072179 double scoreC = genScore(200, 1, 290072179); // 200.1290072179 double scoreD = genScore(400, 0, 290072179); // 400.0290072179 double scoreE = genScore(200, 1, 290081199); // 200.1290081199 // 将生成的score值添加到Redis中： redisTemplate.opsForZSet().add("r", "A", scoreA); redisTemplate.opsForZSet().add("r", "B", scoreB); redisTemplate.opsForZSet().add("r", "C", scoreC); redisTemplate.opsForZSet().add("r", "D", scoreD); redisTemplate.opsForZSet().add("r", "E", scoreE); // 获取并展示结果： Set> rangeWithScores = redisTemplate.opsForZSet().rangeWithScores("r", 0, -1); for (ZSetOperations.TypedTuple tuple : rangeWithScores) { System.out.println(tuple.getValue() + ": " + tuple.getScore()); } ``` ### 总结在数据量不大的情况下，直接使用 RDB 进行排序是比较方便的选择。如果使用 Redis，在后两个维度不变的情况下可以直接利用 key 排序，这样实现简单且高效。如果维度会变化，可以采用二进制或十进制的方法存储排序维度。 - **二进制分段**：分配每个维度固定的位数，高效且适合数值较大的场景。 - **十进制拆分**：将不同维度的数据组合成一个浮点数，计算简单且可读性好。通过这些方法，可以实现多维度的排序，并根据实际需求选择合适的实现方式。

1 个回答