Jedis 進階應用詳解：事務、管道、發布訂閱與分片集群

前言

Jedis 是 Redis 官方推薦的 Java 客戶端。除基本 CRUD 外，掌握事務、管道、發布/訂閱及分片集群（ShardedJedis）等高級特性，對構建高性能、高可用的 Redis 應用至關重要。本文結合代碼示例詳細講解這些核心功能。

一、事務

Redis 事務通過 MULTI、EXEC、WATCH 實現原子性操作。Jedis 中使用 Transaction 對象封裝事務邏輯。

基本語法

jedis.watch(key1, key2);       // 樂觀鎖，監聽鍵變化
Transaction tx = jedis.multi(); // 開啟事務
tx.set("foo", "bar");
tx.exec();                      // 執行事務

事務中獲取值

事務塊內不可直接用 jedis.get()，需通過 Transaction 返回的 Response 對象，並在 exec() 後調用 get() 獲取結果。

try (Jedis jedis = pool.getResource()) {
    Transaction t = jedis.multi();
    t.set("fool", "bar");
    // jedis.get("fool"); // 錯誤：Cannot use Jedis when in Multi

    Response<String> result1 = t.get("fool"); // 正確方式
    t.zadd("foo", 1, "barowitch");
    t.zadd("foo", 0, "barinsky");
    Response<Set<String>> sose = t.zrange("foo", 0, -1);

    t.exec(); // 必須執行，否則命令不生效

    String foolbar = result1.get(); // 此處才能拿到結果
    int soseSize = sose.get().size();
    System.out.println(foolbar);
    System.out.println(soseSize);
}

注意：Response.get() 必須在 t.exec() 之後調用，否則返回 null。

二、管道

管道可將多條命令一次性發送至 Redis，減少網絡往返延遲，適用於命令間無依賴關係的批量操作。

Pipeline p = jedis.pipelined();
p.set("fool", "bar");
p.zadd("foo", 1, "barowitch");
p.zadd("foo", 0, "barinsky");
p.zadd("foo", 0, "barikoviev");

Response<String> pipeString = p.get("fool");
Response<Set<String>> sose = p.zrange("foo", 0, -1);

p.sync(); // 同步執行並獲取結果

int soseSize = sose.get().size();
Set<String> setBack = sose.get();

三、發布/訂閱

Redis 發布/訂閱模式實現消息廣播。關鍵限制：subscribe() 是阻塞操作，且發布與訂閱不能共用同一個 Jedis 連接。

監聽器實現

class MyListener extends JedisPubSub {
    @Override
    public void onMessage(String channel, String message) {
        System.out.println("收到消息: channel=" + channel + ", msg=" + message);
    }
    @Override
    public void onSubscribe(String channel, int subscribedChannels) {
        System.out.println("訂閱成功: " + channel);
    }
    // 其他 onUnsubscribe, onPSubscribe, onPUnsubscribe, onPMessage 略
}

生產者與消費者示例

final JedisPool pool = new JedisPool(new JedisPoolConfig(), "localhost");

// 兩個訂閱者線程
new Thread(() -> {
    try (Jedis jedis = pool.getResource()) {
        jedis.subscribe(new MyListener(), "channel01");
    }
}).start();

new Thread(() -> {
    try (Jedis jedis = pool.getResource()) {
        jedis.subscribe(new MyListener(), "channel01");
    }
}).start();

// 一個發布者線程
new Thread(() -> {
    try (Jedis jedis = pool.getResource()) {
        while (true) {
            TimeUnit.SECONDS.sleep(1);
            String msg = new Date().toLocaleString();
            System.out.println("發布消息: " + msg);
            jedis.publish("channel01", msg);
        }
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}).start();

運行輸出

channel:channel01, subscribedChannels:1
channel:channel01, subscribedChannels:1
發布消息:2016-7-20 10:03:42
收到消息: channel=channel01, message=2016-7-20 10:03:42
收到消息: channel=channel01, message=2016-7-20 10:03:42
發布消息:2016-7-20 10:03:43
...

四、分片集群

ShardedJedis 使用一致性哈希算法將 Key 分佈到多個 Redis 實例，實現負載均衡與水平擴展。

環境準備

啟動三個 Redis 實例，端口分別為 6379、6380、6381。

代碼示例

JedisPoolConfig poolConfig = new JedisPoolConfig();
poolConfig.setMaxTotal(2);
poolConfig.setMaxIdle(1);
poolConfig.setMaxWaitMillis(2000);

String host = "127.0.0.1";
List<JedisShardInfo> shards = Arrays.asList(
    new JedisShardInfo(host, 6379, 500),
    new JedisShardInfo(host, 6380, 500),
    new JedisShardInfo(host, 6381, 500)
);

ShardedJedisPool jedisPool = new ShardedJedisPool(poolConfig, shards);

try (ShardedJedis jedis = jedisPool.getResource()) {
    jedis.set("test", "test");
    System.out.println(jedis.get("test")); // 輸出: test
}

// 驗證 Key 分佈情況
try (ShardedJedis jedis = jedisPool.getResource()) {
    jedis.set("1111", "1111");
    jedis.set("2222", "2222");
    jedis.set("3333", "3333");
    jedis.set("4444", "4444");

    System.out.println("1111 在: " + jedis.getShard("1111").getClient().getHost() + ":" + jedis.getShard("1111").getClient().getPort());
    System.out.println("2222 在: " + jedis.getShard("2222").getClient().getHost() + ":" + jedis.getShard("2222").getClient().getPort());
    System.out.println("3333 在: " + jedis.getShard("3333").getClient().getHost() + ":" + jedis.getShard("3333").getClient().getPort());
    System.out.println("4444 在: " + jedis.getShard("4444").getClient().getHost() + ":" + jedis.getShard("4444").getClient().getPort());
}

輸出結果

test
1111 在: 127.0.0.1:6380
2222 在: 127.0.0.1:6379
3333 在: 127.0.0.1:6381
4444 在: 127.0.0.1:6381

參考資源

• Jedis Wiki - AdvancedUsage: https://github.com/xetorthio/jedis/wiki/AdvancedUsage
• Redis 客戶端 Jedis 使用總結: cnblogs.com/coder2012
• Jedis 事務與管道實踐: CSDN lihao21