案例目標
簡單介紹 redis pipeline 的機制，結合一段實例說明pipeline 在提升吞吐量方面發生的效用。
案例背景
應用系統在數據推送或事件處理過程中，往往出現數據流經過多個網元；
然而在某些服務中，數據操作對redis 是強依賴的，在最近的一次分析中發現：
一次數據推送會對 redis 產生近30次讀寫操作！
在數據推送業務中的性能壓測中，以數據上報 -> 下發應答為一次事務；而對于這樣的讀寫模型，redis 的操作過于頻繁，很快便導致系統延時過高，吞吐量低下，無法滿足目標；
優化過程 主要針對業務代碼做的優化，其中redis 操作經過大量合并，最終降低到原來的1/5，而系統吞吐量也提升明顯。
其中，redis pipeline(管道機制) 的應用是一個關鍵手段。
pipeline的解釋
Pipeline指的是管道技術，指的是客戶端允許將多個請求依次發給服務器，過程中而不需要等待請求的回復，在最后再一并讀取結果即可。
管道技術使用廣泛，例如許多POP3協議已經實現支持這個功能，大大加快了從服務器下載新郵件的過程。 Redis很早就支持管道（pipeline）技術。(因此無論你運行的是什么版本，你都可以使用管道（pipelining）操作Redis)
普通請求模型
[圖-pipeline1]
Pipeline請求模型
[圖-pipeline2]
從兩個圖的對比中可看出，普通的請求模型是同步的，每次請求對應一次IO操作等待；
而Pipeline 化之后所有的請求合并為一次IO，除了時延可以降低之外，還能大幅度提升系統吞吐量。
代碼實例
說明
本地開啟50個線程，每個線程完成1000個key的寫入，對比pipeline開啟及不開啟兩種場景下的性能表現。
相關常量
|
1
2
3
4
5
6
7
8
|
// 并發任務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; |
初始化連接
|
1
2
3
4
5
6
7
8
|
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); |
并發啟動任務，統計執行時間
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
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寫入的細節，區分不同場景
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
public void run() { logger.info("Task[{}] start.", id); try { if (usePipeline) { runWithPipeline(); } else { runWithNonPipeline(); } } finally { latch.countDown(); } logger.info("Task[{}] end.", id); } |
不使用Pipeline的場景比較簡單，循環執行set操作
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
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個作為一批命令執行
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
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
[圖-nopipeline]
Pipeline-stat
[圖-pipeline]
注意事項
pipeline機制可以優化吞吐量，但無法提供原子性/事務保障，而這個可以通過Redis-Multi等命令實現。
參考這里
部分讀寫操作存在相關依賴，無法使用pipeline實現，可利用Script機制，但需要在可維護性方面做好取舍。
擴展閱讀
以上這篇redis通過pipeline提升吞吐量的方法就是小編分享給大家的全部內容了，希望能給大家一個參考，也希望大家多多支持服務器之家。
