案例目標

簡單介紹 redis pipeline 的機制，結合一段例項說明pipeline 在提升吞吐量方面發生的效用。

案例背景

應用系統在資料推送或事件處理過程中，往往出現數據流經過多個網元；
然而在某些服務中，資料操作對redis 是強依賴的，在最近的一次分析中發現：
一次資料推送會對 redis 產生近30次讀寫操作！

在資料推送業務中的效能壓測中，以資料上報 -> 下發應答為一次事務；
而對於這樣的讀寫模型，redis 的操作過於頻繁，很快便導致系統延時過高，吞吐量低下，無法滿足目標；

優化過程 主要針對業務程式碼做的優化，其中redis 操作經過大量合併，最終降低到原來的1/5，而系統吞吐量也提升明顯。
其中，redis pipeline(管道機制) 的應用是一個關鍵手段。

pipeline的解釋

Pipeline指的是管道技術，指的是客戶端允許將多個請求依次發給伺服器，過程中而不需要等待請求的回覆，在最後再一併讀取結果即可。
管道技術使用廣泛，例如許多POP3協議已經實現支援這個功能，大大加快了從伺服器下載新郵件的過程。
Redis很早就支援管道（pipeline）技術。(因此無論你執行的是什麼版本，你都可以使用管道（pipelining）操作Redis)

程式碼例項

說明
本地開啟50個執行緒，每個執行緒完成1000個key的寫入，對比pipeline開啟及不開啟兩種場景下的效能表現。

相關常量

// 併發任務
    private static final int taskCount = 50;
    // pipeline大小
    private static final int batchSize = 10;
    // 每個任務處理命令數
    private static final int cmdCount = 1000;

    private static final boolean usePipeline = true;
 

初始連結

 JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxActive(200);
        poolConfig.setMaxIdle(100);
        poolConfig.setMaxWait(2000);
        poolConfig.setTestOnBorrow(false);
        poolConfig.setTestOnReturn(false);

        jedisPool = new JedisPool(poolConfig, host, port);
 

併發啟動任務，統計執行時間

public static void main(String[] args) throws InterruptedException {
        init();

        flushDB();

        long t1 = System.currentTimeMillis();
        ExecutorService executor = Executors.newCachedThreadPool();

        CountDownLatch latch = new CountDownLatch(taskCount);
        for (int i = 0; i < taskCount; i++) {
            executor.submit(new DemoTask(i, latch));
        }

        latch.await();
        executor.shutdownNow();

        long t2 = System.currentTimeMillis();

        System.out.println("execution finish time(s):" + (t2 - t1) / 1000.0);

    }

DemoTask 封裝了執行key寫入的細節，區分不同場景

public void run() {
            logger.info("Task[{}] start.", id);
            try {
                if (usePipeline) {
                    runWithPipeline();
                } else {
                    runWithNonPipeline();
                }
            } finally {
                latch.countDown();
            }

            logger.info("Task[{}] end.", id);
     }

不使用Pipeline的場景比較簡單，迴圈執行set操作

  for (int i = 0; i < cmdCount; i++) {
                Jedis jedis = get();
                try {
                    jedis.set(key(i), UUID.randomUUID().toString());
                } finally {
                    if (jedis != null) {
                        jedisPool.returnResource(jedis);
                    }
                }
                if (i % batchSize == 0) {
                    logger.info("Task[{}] process -- {}", id, i);
                }
            }

使用Pipeline，需要處理分段，如10個作為一批命令執行

for (int i = 0; i < cmdCount;) {
                Jedis jedis = get();

                try {
                    Pipeline pipeline = jedis.pipelined();
                    int j;
                    for (j = 0; j < batchSize; j++) {
                        if (i + j < cmdCount) {
                            pipeline.set(key(i + j), UUID.randomUUID().toString());
                        } else {
                            break;
                        }
                    }
                    pipeline.sync();
                    logger.info("Task[{}] pipeline -- {}", id, i + j);

                    i += j;

                } finally {
                    if (jedis != null) {
                        jedisPool.returnResource(jedis);
                    }
                }

            }

執行結果

不使用Pipeline，整體執行26s；而使用Pipeline優化後的程式碼，執行時間僅需要3s！

NoPipeline-stat

注意事項

• pipeline機制可以優化吞吐量，但無法提供原子性/事務保障，而這個可以通過Redis-Multi等命令實現。
  參考這裡
• 部分讀寫操作存在相關依賴，無法使用pipeline實現，可利用Script機制，但需要在可維護性方面做好取捨。

擴充套件閱讀

• 官方文件-Redis-Pipelining
• 官方文件-Redis-Transaction