案例目標
簡單介紹 redis pipeline 的機制，結(jié)合一段實例說明pipeline 在提升吞吐量方面發(fā)生的效用。
案例背景
應用系統(tǒng)在數(shù)據(jù)推送或事件處理過程中，往往出現(xiàn)數(shù)據(jù)流經(jīng)過多個網(wǎng)元；
然而在某些服務中，數(shù)據(jù)操作對redis 是強依賴的，在最近的一次分析中發(fā)現(xiàn)：
一次數(shù)據(jù)推送會對 redis 產(chǎn)生近30次讀寫操作！
在數(shù)據(jù)推送業(yè)務中的性能壓測中，以數(shù)據(jù)上報 -> 下發(fā)應答為一次事務；而對于這樣的讀寫模型，redis 的操作過于頻繁，很快便導致系統(tǒng)延時過高，吞吐量低下，無法滿足目標；
優(yōu)化過程 主要針對業(yè)務代碼做的優(yōu)化，其中redis 操作經(jīng)過大量合并，最終降低到原來的1/5，而系統(tǒng)吞吐量也提升明顯。
其中，redis pipeline(管道機制) 的應用是一個關鍵手段。
pipeline的解釋
Pipeline指的是管道技術，指的是客戶端允許將多個請求依次發(fā)給服務器，過程中而不需要等待請求的回復，在最后再一并讀取結(jié)果即可。
管道技術使用廣泛，例如許多POP3協(xié)議已經(jīng)實現(xiàn)支持這個功能，大大加快了從服務器下載新郵件的過程。 Redis很早就支持管道（pipeline）技術。(因此無論你運行的是什么版本，你都可以使用管道（pipelining）操作Redis)
普通請求模型
[圖-pipeline1]
Pipeline請求模型
[圖-pipeline2]
從兩個圖的對比中可看出，普通的請求模型是同步的，每次請求對應一次IO操作等待；
而Pipeline 化之后所有的請求合并為一次IO，除了時延可以降低之外，還能大幅度提升系統(tǒng)吞吐量。
代碼實例
說明
本地開啟50個線程，每個線程完成1000個key的寫入，對比pipeline開啟及不開啟兩種場景下的性能表現(xiàn)。
相關常量
// 并發(fā)任務
private static final int taskCount = 50;
// pipeline大小
private static final int batchSize = 10;
// 每個任務處理命令數(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ā)啟動任務，統(tǒng)計執(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寫入的細節(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);
}
}
}
運行結(jié)果
不使用Pipeline，整體執(zhí)行26s；而使用Pipeline優(yōu)化后的代碼，執(zhí)行時間僅需要3s！
NoPipeline-stat 
[圖-nopipeline]
Pipeline-stat
[圖-pipeline]
注意事項
pipeline機制可以優(yōu)化吞吐量，但無法提供原子性/事務保障，而這個可以通過Redis-Multi等命令實現(xiàn)。
參考這里
部分讀寫操作存在相關依賴，無法使用pipeline實現(xiàn)，可利用Script機制，但需要在可維護性方面做好取舍。
擴展閱讀
官方文檔-Redis-Pipelining
官方文檔-Redis-Transaction
以上這篇redis通過pipeline提升吞吐量的方法就是小編分享給大家的全部內(nèi)容了，希望能給大家一個參考，也希望大家多多支持腳本之家。
您可能感興趣的文章:- python中sklearn的pipeline模塊實例詳解
- Redis利用Pipeline加速查詢速度的方法
- 在Redis集群中使用pipeline批量插入的實現(xiàn)方法
- 使用Jenkins Pipeline自動化構建發(fā)布Java項目的方法
- python使用pipeline批量讀寫redis的方法
- Python：Scrapy框架中Item Pipeline組件使用詳解
- 詳解redis大幅性能提升之使用管道（PipeLine）和批量（Batch）操作
- 如何使用pipeline和jacoco獲取自動化測試代碼覆蓋率
