案例目標(biāo)

簡單介紹 redis pipeline 的機(jī)制，結(jié)合一段實(shí)例說明pipeline 在提升吞吐量方面發(fā)生的效用。

案例背景

應(yīng)用系統(tǒng)在數(shù)據(jù)推送或事件處理過程中，往往出現(xiàn)數(shù)據(jù)流經(jīng)過多個網(wǎng)元；
然而在某些服務(wù)中，數(shù)據(jù)操作對redis 是強(qiáng)依賴的，在最近的一次分析中發(fā)現(xiàn)：
一次數(shù)據(jù)推送會對 redis 產(chǎn)生近30次讀寫操作！

在數(shù)據(jù)推送業(yè)務(wù)中的性能壓測中，以數(shù)據(jù)上報(bào) -> 下發(fā)應(yīng)答為一次事務(wù)；
而對于這樣的讀寫模型，redis 的操作過于頻繁，很快便導(dǎo)致系統(tǒng)延時過高，吞吐量低下，無法滿足目標(biāo)；

優(yōu)化過程 主要針對業(yè)務(wù)代碼做的優(yōu)化，其中redis 操作經(jīng)過大量合并，最終降低到原來的1/5，而系統(tǒng)吞吐量也提升明顯。
其中，redis pipeline(管道機(jī)制)?的應(yīng)用是一個關(guān)鍵手段。

pipeline的解釋

Pipeline指的是管道技術(shù)，指的是客戶端允許將多個請求依次發(fā)給服務(wù)器，過程中而不需要等待請求的回復(fù)，在最后再一并讀取結(jié)果即可。
管道技術(shù)使用廣泛，例如許多POP3協(xié)議已經(jīng)實(shí)現(xiàn)支持這個功能，大大加快了從服務(wù)器下載新郵件的過程。
Redis很早就支持管道（pipeline）技術(shù)。(因此無論你運(yùn)行的是什么版本，你都可以使用管道（pipelining）操作Redis)

普通請求模型

[圖-pipeline1]

Pipeline請求模型

[圖-pipeline2]

從兩個圖的對比中可看出，普通的請求模型是同步的，每次請求對應(yīng)一次IO操作等待；
而Pipeline 化之后所有的請求合并為一次IO，除了時延可以降低之外，還能大幅度提升系統(tǒng)吞吐量。

代碼實(shí)例

說明
本地開啟50個線程，每個線程完成1000個key的寫入，對比pipeline開啟及不開啟兩種場景下的性能表現(xiàn)。

相關(guān)常量

// 并發(fā)任務(wù)private static final int taskCount = 50;// pipeline大小private static final int batchSize = 10;// 每個任務(wù)處理命令數(shù)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);

并發(fā)啟動任務(wù)，統(tǒng)計(jì)執(zhí)行時間

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?封裝了執(zhí)行key寫入的細(xì)節(jié)，區(qū)分不同場景

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

不使用Pipeline的場景比較簡單，循環(huán)執(zhí)行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個作為一批命令執(zhí)行

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);}}}

運(yùn)行結(jié)果

不使用Pipeline，整體執(zhí)行26s；而使用Pipeline優(yōu)化后的代碼，執(zhí)行時間僅需要3s！

NoPipeline-stat

[圖-nopipeline]

Pipeline-stat

[圖-pipeline]

注意事項(xiàng)

• pipeline機(jī)制可以優(yōu)化吞吐量，但無法提供原子性/事務(wù)保障，而這個可以通過Redis-Multi等命令實(shí)現(xiàn)。
  參考這里
• 部分讀寫操作存在相關(guān)依賴，無法使用pipeline實(shí)現(xiàn)，可利用Script機(jī)制，但需要在可維護(hù)性方面做好取舍。

擴(kuò)展閱讀

官方文檔-Redis-Pipelining
官方文檔-Redis-Transaction

來源:https://www.cnblogs.com/littleatp/p/8419796.html

總結(jié)

以上是生活随笔為你收集整理的redis通过pipeline提升吞吐量的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。