redis如何通過pipeline提升吞吐量

220次閱讀

沒有評論

共計 2893 個字符，預計需要花費 8 分鐘才能閱讀完成。

這篇文章給大家分享的是有關 redis 如何通過 pipeline 提升吞吐量的內容。丸趣 TV 小編覺得挺實用的，因此分享給大家做個參考，一起跟隨丸趣 TV 小編過來看看吧。

案例目標

簡單介紹 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 開啟及不開啟兩種場景下的性能表現。

相關常量

//  并發任務
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-stat

注意事項

pipeline 機制可以優化吞吐量，但無法提供原子性 / 事務保障，而這個可以通過 Redis-Multi 等命令實現。

感謝各位的閱讀！關于“redis 如何通過 pipeline 提升吞吐量”這篇文章就分享到這里了，希望以上內容可以對大家有一定的幫助，讓大家可以學到更多知識，如果覺得文章不錯，可以把它分享出去讓更多的人看到吧！

正文完