redis集群模式下根据key计算槽从而得到该key位于哪个redis服务器上批处理是建立一次连接并批处理所有数据集群下批处理的数据可能位于不同槽不同服务器因此一个连接处理不了因此报错解决方案1. 方案一哈希标签Hash Tags—— “强行合群”这是从数据设计层面解决问题的最快方法。逻辑原理Redis 集群在计算 Key 的槽位时如果发现 Key 包含{}则只会对{}里面的内容进行哈希计算。具体做法将需要批量操作的 Key 加上相同的标签。原本feed:user:1feed:user:2随机散落在不同节点。改造feed:{group_A}:user:1feed:{group_A}:user:2。结果所有带{group_A}的 Key 必定落在同一个槽Slot并在同一个物理节点上。优缺点利Pipeline 效率最高完全像操作单机 Redis 一样。弊容易造成“数据倾斜”Data Skew。如果某个{group}下的数据量巨大会导致集群中某个节点被撑爆而其他节点无所事事。2. 方案二客户端分片路由Client-side Sharding—— “分治法”这是在逻辑控制层面最通用的做法。逻辑原理既然一个连接只能去一个节点那就先把这一批 Key 按“家乡”分好组分头派发。具体步骤分拣Mapping遍历你要处理的 1000 个 Key利用 CRC16 算法计算每个 Key 的 Slot并根据集群拓扑图Cluster Nodes Info找到对应的节点地址。分组Grouping将 Key 归类。例如节点 A 负责 300 个节点 B 负责 400 个节点 C 负责 300 个。并发执行Parallel Execution同时开启三个 Pipeline分别向节点 A、B、C 发送请求。汇总Aggregation等待三个 Pipeline 全部返回结果最后合并给业务层。优缺点利符合集群分布式的初衷负载均衡不会产生数据倾斜。弊逻辑复杂虽然是并行但网络往返RTT受限于最慢的那个节点。基于Jedis的方案实现在Jedis环境下针对 Redis Cluster 的 Pipeline 陷阱这两个方案的实现思路完全不同。一个是“改数据命名”一个是“改代码逻辑”。方案一哈希标签Hash Tags这是逻辑上最简单的做法。通过在 Key 中加入{}强制让这一批粉丝的收件箱落在同一个 Redis 节点上。代码逻辑publicvoidpushWithHashTag(StringblogId,ListLongfollowerIds){// 假设我们要把这批推送任务锁定在同一个槽位Slot// 只要 {} 里的内容相同Redis 就会计算出相同的 SlotStringslotTag{group_1};longcurrentTimeSystem.currentTimeMillis();// 在 Jedis 环境下由于 Key 都在同一个节点普通的 Pipeline 即可生效try(JedisjedisjedisPool.getResource()){// 注意这里必须是连到对应节点的 JedisPipelinepipelinejedis.pipelined();for(LongfId:followerIds){// Key 变成feed:{group_1}:user:123Stringkeyfeed:slotTag:user:fId;pipeline.zadd(key,(double)currentTime,blogId);}pipeline.sync();// 一次性发送}}方案二客户端手动分拣分组Manual Grouping由于JedisCluster官方对象并不直接提供跨槽位的pipelined()方法因为它无法确定你这一堆 Key 到底要去哪个节点我们需要手动按照节点JedisPool进行分组分发。代码逻辑importredis.clients.jedis.*;importredis.clients.jedis.util.JedisClusterCRC16;importjava.lang.reflect.Field;importjava.util.List;importjava.util.Map;importjava.util.TreeMap;publicclassJedisClusterPipelineHelper{/** * 获取 JedisCluster 内部维护的 Slot 与 JedisPool 的映射关系 * 这里的逻辑是通过反射进入 JedisCluster 的内部缓存 */SuppressWarnings(unchecked)publicstaticJedisPoolgetPoolBySlot(JedisClusterjedisCluster,intslot){try{// 1. 获取 JedisCluster 中的 connectionHandler// Jedis 3.x 及以上版本字段名为 connectionHandlerFieldhandlerFieldJedisCluster.class.getDeclaredField(connectionHandler);handlerField.setAccessible(true);ObjecthandlerhandlerField.get(jedisCluster);// 2. 从 handler 中获取 JedisClusterInfoCache (它存储了槽位分布)FieldcacheFieldhandler.getClass().getSuperclass().getDeclaredField(cache);cacheField.setAccessible(true);ObjectcachecacheField.get(handler);// 3. 从 cache 中获取 slots 映射表// slots 是一个 MapInteger, JedisPool 或者 JedisPool[] 数组FieldslotsFieldcache.getClass().getDeclaredField(slots);slotsField.setAccessible(true);// Jedis 3.x 以上版本通常返回 JedisPool[] 或 ListJedisPoolObjectslotsslotsField.get(cache);if(slotsinstanceofMap){return((MapInteger,JedisPool)slots).get(slot);}elseif(slotsinstanceofList){return((ListJedisPool)slots).get(slot);}elseif(slots.getClass().isArray()){return((JedisPool[])slots)[slot];}returnnull;}catch(Exceptione){thrownewRuntimeException(无法从 JedisCluster 中提取槽位映射信息,e);}}}importredis.clients.jedis.*;importredis.clients.jedis.util.JedisClusterCRC16;publicvoidpushWithGrouping(StringblogId,ListLongfollowerIds){// 1. 准备一个 MapKey 是节点连接池Value 是属于该节点的粉丝 ID 列表MapJedisPool,ListLongnodeTasksnewHashMap();// 假设 jedisCluster 是你注入的集群对象// 获取集群中所有的节点映射关系Slot - PoolMapString,JedisPoolclusterNodesjedisCluster.getClusterNodes();for(LongfId:followerIds){StringkeyFEED_KEYfId;// 计算这个 Key 所在的 SlotintslotJedisClusterCRC16.getSlot(key);// 2. 找到该 Slot 所在的连接池 (这里简化了逻辑实际需根据 Slot 映射查找)JedisPoolconnectionPoolgetPoolBySlot(slot);nodeTasks.computeIfAbsent(connectionPool,k-newArrayList()).add(fId);}// 3. 对每个节点分别开启 PipelinenodeTasks.forEach((pool,subList)-{try(Jedisjedispool.getResource()){Pipelinepipelinejedis.pipelined();for(LongfId:subList){StringkeyFEED_KEYfId;pipeline.zadd(key,(double)System.currentTimeMillis(),blogId);}pipeline.sync();// 针对单台服务器的批量提交}});}基于Lettuce的方案实现在 Lettuce 中你完全不需要像 Jedis 那样写复杂的反射代码去查 Slot、找节点、分连接池。因为 Lettuce 的底层是基于Netty的异步框架它天生就支持“自动路由”。1. Lettuce 的核心逻辑自动分拣当你使用 Lettuce 连接 Redis 集群并执行批量操作时逻辑是这样的你只管发你把 1000 个zAdd丢给 Lettuce。它帮你拆Lettuce 的RedisAdvancedClusterAsyncCommands会在内存里自动计算每个 Key 的 Slot并发现它们分属于哪些节点。并发分发Lettuce 会针对每个涉及到的节点异步地发起 TCP 请求。结果聚合等所有节点的响应都回来后它再把结果拼好还给你。2. 基于 Spring Data Redis (Lettuce) 的实现如果你在项目中去掉了 Jedis 的排除逻辑回到默认的 Lettuce代码会变得异常简洁。你之前的executePipelined代码几乎不需要改动它在底层就会自动支持集群分发。publicvoidhandleInsertWithLettuce(ListLongfollowerIds,StringblogIdStr){longcurrentTimeSystem.currentTimeMillis();// 在 Lettuce 环境下executePipelined 内部会自动处理集群路由ListObjectresultsstringRedisTemplate.executePipelined((RedisCallbackObject)connection-{for(LongfId:followerIds){StringkeyFEED_KEYfId;// 这里的 zAdd 虽然看起来是连着写的// 但 Lettuce 底层会异步地将它们发往不同的集群节点connection.zAdd(key.getBytes(),(double)currentTime,blogIdStr.getBytes());}returnnull;});// results 会包含所有操作的结果log.info(Lettuce 集群批量推送完成处理条数{},results.size());}3. 如果你想用原生 Lettuce API脱离 Spring如果你想绕过 Spring 的封装直接用原生的 Lettuce 追求极致性能代码逻辑是“异步并发”// 1. 获取异步集群命令对象RedisAdvancedClusterAsyncCommandsString,StringasyncCommandsclusterConnection.async();// 2. 禁用自动刷新触发可选为了让命令更像 Pipeline 一样堆积asyncCommands.setAutoFlushCommands(false);ListRedisFuture?futuresnewArrayList();for(LongfId:followerIds){// 发起异步操作这些操作会被自动路由到对应的节点futures.add(asyncCommands.zadd(FEED_KEYfId,System.currentTimeMillis(),blogIdStr));}// 3. 一次性将缓冲区命令刷出去asyncCommands.flushCommands();// 4. 等待所有异步任务完成CompletableFuture.allOf(futures.toArray(newCompletableFuture[0])).join();
Redis集群批处理下的陷阱
redis集群模式下根据key计算槽从而得到该key位于哪个redis服务器上批处理是建立一次连接并批处理所有数据集群下批处理的数据可能位于不同槽不同服务器因此一个连接处理不了因此报错解决方案1. 方案一哈希标签Hash Tags—— “强行合群”这是从数据设计层面解决问题的最快方法。逻辑原理Redis 集群在计算 Key 的槽位时如果发现 Key 包含{}则只会对{}里面的内容进行哈希计算。具体做法将需要批量操作的 Key 加上相同的标签。原本feed:user:1feed:user:2随机散落在不同节点。改造feed:{group_A}:user:1feed:{group_A}:user:2。结果所有带{group_A}的 Key 必定落在同一个槽Slot并在同一个物理节点上。优缺点利Pipeline 效率最高完全像操作单机 Redis 一样。弊容易造成“数据倾斜”Data Skew。如果某个{group}下的数据量巨大会导致集群中某个节点被撑爆而其他节点无所事事。2. 方案二客户端分片路由Client-side Sharding—— “分治法”这是在逻辑控制层面最通用的做法。逻辑原理既然一个连接只能去一个节点那就先把这一批 Key 按“家乡”分好组分头派发。具体步骤分拣Mapping遍历你要处理的 1000 个 Key利用 CRC16 算法计算每个 Key 的 Slot并根据集群拓扑图Cluster Nodes Info找到对应的节点地址。分组Grouping将 Key 归类。例如节点 A 负责 300 个节点 B 负责 400 个节点 C 负责 300 个。并发执行Parallel Execution同时开启三个 Pipeline分别向节点 A、B、C 发送请求。汇总Aggregation等待三个 Pipeline 全部返回结果最后合并给业务层。优缺点利符合集群分布式的初衷负载均衡不会产生数据倾斜。弊逻辑复杂虽然是并行但网络往返RTT受限于最慢的那个节点。基于Jedis的方案实现在Jedis环境下针对 Redis Cluster 的 Pipeline 陷阱这两个方案的实现思路完全不同。一个是“改数据命名”一个是“改代码逻辑”。方案一哈希标签Hash Tags这是逻辑上最简单的做法。通过在 Key 中加入{}强制让这一批粉丝的收件箱落在同一个 Redis 节点上。代码逻辑publicvoidpushWithHashTag(StringblogId,ListLongfollowerIds){// 假设我们要把这批推送任务锁定在同一个槽位Slot// 只要 {} 里的内容相同Redis 就会计算出相同的 SlotStringslotTag{group_1};longcurrentTimeSystem.currentTimeMillis();// 在 Jedis 环境下由于 Key 都在同一个节点普通的 Pipeline 即可生效try(JedisjedisjedisPool.getResource()){// 注意这里必须是连到对应节点的 JedisPipelinepipelinejedis.pipelined();for(LongfId:followerIds){// Key 变成feed:{group_1}:user:123Stringkeyfeed:slotTag:user:fId;pipeline.zadd(key,(double)currentTime,blogId);}pipeline.sync();// 一次性发送}}方案二客户端手动分拣分组Manual Grouping由于JedisCluster官方对象并不直接提供跨槽位的pipelined()方法因为它无法确定你这一堆 Key 到底要去哪个节点我们需要手动按照节点JedisPool进行分组分发。代码逻辑importredis.clients.jedis.*;importredis.clients.jedis.util.JedisClusterCRC16;importjava.lang.reflect.Field;importjava.util.List;importjava.util.Map;importjava.util.TreeMap;publicclassJedisClusterPipelineHelper{/** * 获取 JedisCluster 内部维护的 Slot 与 JedisPool 的映射关系 * 这里的逻辑是通过反射进入 JedisCluster 的内部缓存 */SuppressWarnings(unchecked)publicstaticJedisPoolgetPoolBySlot(JedisClusterjedisCluster,intslot){try{// 1. 获取 JedisCluster 中的 connectionHandler// Jedis 3.x 及以上版本字段名为 connectionHandlerFieldhandlerFieldJedisCluster.class.getDeclaredField(connectionHandler);handlerField.setAccessible(true);ObjecthandlerhandlerField.get(jedisCluster);// 2. 从 handler 中获取 JedisClusterInfoCache (它存储了槽位分布)FieldcacheFieldhandler.getClass().getSuperclass().getDeclaredField(cache);cacheField.setAccessible(true);ObjectcachecacheField.get(handler);// 3. 从 cache 中获取 slots 映射表// slots 是一个 MapInteger, JedisPool 或者 JedisPool[] 数组FieldslotsFieldcache.getClass().getDeclaredField(slots);slotsField.setAccessible(true);// Jedis 3.x 以上版本通常返回 JedisPool[] 或 ListJedisPoolObjectslotsslotsField.get(cache);if(slotsinstanceofMap){return((MapInteger,JedisPool)slots).get(slot);}elseif(slotsinstanceofList){return((ListJedisPool)slots).get(slot);}elseif(slots.getClass().isArray()){return((JedisPool[])slots)[slot];}returnnull;}catch(Exceptione){thrownewRuntimeException(无法从 JedisCluster 中提取槽位映射信息,e);}}}importredis.clients.jedis.*;importredis.clients.jedis.util.JedisClusterCRC16;publicvoidpushWithGrouping(StringblogId,ListLongfollowerIds){// 1. 准备一个 MapKey 是节点连接池Value 是属于该节点的粉丝 ID 列表MapJedisPool,ListLongnodeTasksnewHashMap();// 假设 jedisCluster 是你注入的集群对象// 获取集群中所有的节点映射关系Slot - PoolMapString,JedisPoolclusterNodesjedisCluster.getClusterNodes();for(LongfId:followerIds){StringkeyFEED_KEYfId;// 计算这个 Key 所在的 SlotintslotJedisClusterCRC16.getSlot(key);// 2. 找到该 Slot 所在的连接池 (这里简化了逻辑实际需根据 Slot 映射查找)JedisPoolconnectionPoolgetPoolBySlot(slot);nodeTasks.computeIfAbsent(connectionPool,k-newArrayList()).add(fId);}// 3. 对每个节点分别开启 PipelinenodeTasks.forEach((pool,subList)-{try(Jedisjedispool.getResource()){Pipelinepipelinejedis.pipelined();for(LongfId:subList){StringkeyFEED_KEYfId;pipeline.zadd(key,(double)System.currentTimeMillis(),blogId);}pipeline.sync();// 针对单台服务器的批量提交}});}基于Lettuce的方案实现在 Lettuce 中你完全不需要像 Jedis 那样写复杂的反射代码去查 Slot、找节点、分连接池。因为 Lettuce 的底层是基于Netty的异步框架它天生就支持“自动路由”。1. Lettuce 的核心逻辑自动分拣当你使用 Lettuce 连接 Redis 集群并执行批量操作时逻辑是这样的你只管发你把 1000 个zAdd丢给 Lettuce。它帮你拆Lettuce 的RedisAdvancedClusterAsyncCommands会在内存里自动计算每个 Key 的 Slot并发现它们分属于哪些节点。并发分发Lettuce 会针对每个涉及到的节点异步地发起 TCP 请求。结果聚合等所有节点的响应都回来后它再把结果拼好还给你。2. 基于 Spring Data Redis (Lettuce) 的实现如果你在项目中去掉了 Jedis 的排除逻辑回到默认的 Lettuce代码会变得异常简洁。你之前的executePipelined代码几乎不需要改动它在底层就会自动支持集群分发。publicvoidhandleInsertWithLettuce(ListLongfollowerIds,StringblogIdStr){longcurrentTimeSystem.currentTimeMillis();// 在 Lettuce 环境下executePipelined 内部会自动处理集群路由ListObjectresultsstringRedisTemplate.executePipelined((RedisCallbackObject)connection-{for(LongfId:followerIds){StringkeyFEED_KEYfId;// 这里的 zAdd 虽然看起来是连着写的// 但 Lettuce 底层会异步地将它们发往不同的集群节点connection.zAdd(key.getBytes(),(double)currentTime,blogIdStr.getBytes());}returnnull;});// results 会包含所有操作的结果log.info(Lettuce 集群批量推送完成处理条数{},results.size());}3. 如果你想用原生 Lettuce API脱离 Spring如果你想绕过 Spring 的封装直接用原生的 Lettuce 追求极致性能代码逻辑是“异步并发”// 1. 获取异步集群命令对象RedisAdvancedClusterAsyncCommandsString,StringasyncCommandsclusterConnection.async();// 2. 禁用自动刷新触发可选为了让命令更像 Pipeline 一样堆积asyncCommands.setAutoFlushCommands(false);ListRedisFuture?futuresnewArrayList();for(LongfId:followerIds){// 发起异步操作这些操作会被自动路由到对应的节点futures.add(asyncCommands.zadd(FEED_KEYfId,System.currentTimeMillis(),blogIdStr));}// 3. 一次性将缓冲区命令刷出去asyncCommands.flushCommands();// 4. 等待所有异步任务完成CompletableFuture.allOf(futures.toArray(newCompletableFuture[0])).join();
