// 傳統(tǒng)單次讀取模式（性能低下）
for (String key : keys) {
    String value = jedis.get(key);  // 產(chǎn)生N次網(wǎng)絡(luò)請求
}

// Jedis實現(xiàn)
try (Jedis jedis = pool.getResource()) {
    List<String> values = jedis.mget("key1", "key2", "key3");
}

// Lettuce實現(xiàn)（異步）
RedisClient client = RedisClient.create("redis://localhost");
StatefulRedisConnection<String, String> connection = client.connect();
List<String> values = connection.sync().mget("key1", "key2").stream()
        .map(KeyValue::getValue)
        .collect(Collectors.toList());

// Jedis Pipeline
try (Jedis jedis = pool.getResource()) {
    Pipeline p = jedis.pipelined();
    for (String key : keys) {
        p.get(key);  // 將命令放入緩沖區(qū)
    }
    List<Object> results = p.syncAndReturnAll();  // 一次性發(fā)送
}

???????// Lettuce Pipeline（異步）
List<RedisFuture<String>> futures = new ArrayList<>();
for (String key : keys) {
    futures.add(commands.get(key));  // 非阻塞提交
}
// 統(tǒng)一獲取結(jié)果
List<String> values = futures.stream()
        .map(RedisFuture::get)
        .collect(Collectors.toList());

// 基于Spring Data Redis的批量實現(xiàn)
@Autowired
private RedisTemplate<String, UserProfile> redisTemplate;

???????public Map<String, UserProfile> batchGetUserProfiles(List<String> userIds) {
    // 1. 構(gòu)建Key列表
    List<String> keys = userIds.stream()
            .map(id -> "user:profile:" + id)
            .collect(Collectors.toList());
    
    // 2. 執(zhí)行批量查詢
    List<UserProfile> profiles = redisTemplate.opsForValue().multiGet(keys);
    
    // 3. 組裝返回結(jié)果
    Map<String, UserProfile> result = new HashMap<>();
    for (int i = 0; i < userIds.size(); i++) {
        result.put(userIds.get(i), profiles.get(i));
    }
    return result;
}

// 原始Key：user_profile_{userId}
// 優(yōu)化后：u:p:{userId}  （減少內(nèi)存占用30%+）

# application.yml
spring:
  redis:
    jedis:
      pool:
        max-active: 100   # 最大連接數(shù)
        max-idle: 50
        min-idle: 10

// 使用本地緩存減少Redis訪問
Cache<String, UserProfile> localCache = Caffeine.newBuilder()
    .maximumSize(10_000)
    .expireAfterWrite(5, TimeUnit.MINUTES)
    .build();

// 分批次處理（每批100個Key）
int batchSize = 100;
List<List<String>> partitions = Lists.partition(keys, batchSize);

Map<String, String> result = new HashMap<>();
for (List<String> batch : partitions) {
    List<String> values = jedis.mget(batch.toArray(new String[0]));
    // 合并結(jié)果...
}

// 返回結(jié)果與輸入Key順序一致，不存在時為null
List<String> values = jedis.mget(keys);
for (int i = 0; i < keys.size(); i++) {
    if (values.get(i) != null) {
        // 處理有效數(shù)據(jù)
    }
}

// 限制單次批量操作Key數(shù)量
if (keys.size() > MAX_BATCH_SIZE) {
    throw new IllegalArgumentException("Too many keys");
}

// 通過分片分散壓力
int shard = key.hashCode() % SHARD_COUNT;
Jedis jedis = shardPool[shard].getResource();

// 使用JMH進行基準測試
@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
public class RedisBatchBenchmark {
    
    @Benchmark
    public void testMget(Blackhole bh) {
        List<String> values = jedis.mget(keys);
        bh.consume(values);
    }
}

# redis.conf 關(guān)鍵參數(shù)
tcp-keepalive 60      # 保持連接活躍
maxmemory-policy allkeys-lru  # 內(nèi)存淘汰策略
client-output-buffer-limit normal 2gb 1gb 60 # 客戶端輸出緩沖

// 使用JedisCluster
Set<HostAndPort> nodes = new HashSet<>();
nodes.add(new HostAndPort("127.0.0.1", 7000));
try (JedisCluster cluster = new JedisCluster(nodes)) {
    cluster.mget("key1", "key2");  // 自動路由
}