Redis のパイプライン機能を使用すると、一括操作の実行やコマンド実行パフォーマンスの向上ができます。パイプラインは、複数のコマンドを一度にサーバーに送信することで、ネットワーク遅延を削減し、パフォーマンスを向上させます。Tair (Redis OSS-compatible) は、ネイティブの Redis パイプラインをサポートしています。
パイプラインの概要
通常、クライアントと Redis サーバーは、ピンポンインタラクションパターンでやり取りします。このパターンでは、クライアントはコマンドを送信し、サーバーからのレスポンスを待ってから次のコマンドを送信します。
Redis はパイプラインモードもサポートしています。このモードは、組み立てラインのように動作します。クライアントはコマンドを送信し、レスポンスを待たずに後続のコマンドを送信します。すべてのコマンドを送信した後、クライアントはすべてのレスポンスを読み取り、送信した順序で元のコマンドに対応付けます。ほとんどの Redis クライアントのパイプラインモード実装では、コマンドは一括で送信されます。すべてのコマンド結果を受信した後、結果はアプリケーションに返されます。
以下の図は、ピンポンインタラクションパターンとパイプラインパターンのネットワーク通信を示しています。
パイプラインは、ネットワークのラウンドトリップタイム (RTT) を削減することで、プログラムの効率とパフォーマンスを向上させます。また、read() および write() システムコールの数と、プロセスのコンテキストスイッチも削減します。
パイプラインは、一部のシナリオで非常に効果的です。たとえば、複数のコマンドを迅速にサーバーに送信し、各コマンドに対して即時レスポンスが不要な場合、パイプラインを、パフォーマンスを最適化するための一括処理ツールとして使用できます。
パイプラインを使用する場合、クライアントはサーバーへの接続を専有します。この間、パイプラインが閉じられるまで、パイプラインを使用しない他の操作は実行できません。他の操作を同時に実行するには、パイプライン操作用に別の接続を作成し、通常の操作とは分けておく必要があります。
詳細については、「Pipeline」をご参照ください。
注意事項
パイプラインは原子性を保証しません。
パイプラインは、クライアントの送信方法を一連のコマンドの一括送信に変更するだけです。サーバーは引き続きデータストリームを個々のコマンドに解析し、順序どおりに実行します。各コマンドは独立しています。これは、このプロセス中にサーバーが他のクライアントからのコマンドを実行する可能性があることを意味します。原子性を保証するには、トランザクションまたは Lua スクリプトを使用する必要があります。
パイプラインの実行中にエラーが発生した場合、ロールバックはサポートされません。
パイプラインにはトランザクションの特性がありません。実行するコマンド間に依存関係がある場合は、パイプラインを使用しないでください。
説明redis-py などの一部のクライアントは、MULTI および EXEC トランザクションコマンドを使用してパイプラインをシミュレートします。これらのクライアントを使用する場合は、パイプラインとトランザクションの違いに注意してください。そうしないと、エラーが発生する可能性があります。トランザクションの制限事項の詳細については、「Transactions」をご参照ください。
サーバーと一部のクライアントにはバッファの制限があります。単一のパイプラインにコマンドを多く含めすぎないでください。
パイプラインはクライアント・サーバー間のインタラクションパターンであり、サーバーアーキテクチャには依存しません。したがって、プロキシモードのクラスターアーキテクチャ、直接接続モードのクラスターアーキテクチャ、または読み書き分離アーキテクチャを使用するインスタンスはすべてパイプラインをサポートします。
説明クラスターアーキテクチャには独自の制限があります。たとえば、単一のコマンドでスロットをまたいでキーにアクセスすることはできません。現在のノードに属していないデータにアクセスしようとすると、
-MOVEDエラーが発生します。クラスターアーキテクチャでパイプラインを使用する場合は、パイプライン内のコマンドがクラスターアーキテクチャの実行条件を満たしていることを確認してください。制限事項の詳細については、「クラスターインスタンスと読み書き分離インスタンスのコマンド制限」をご参照ください。
コード例
パフォーマンスの比較
次のコードは、パイプラインを使用した場合と使用しない場合のパフォーマンスの違いを示しています。
package pipeline.kvstore.aliyun.com;
import java.util.Date;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;
public class RedisPipelinePerformanceTest {
static final String host = "xxxxxx.m.cnhza.kvstore.aliyuncs.com";
static final int port = 6379;
static final String password = "password";
public static void main(String[] args) {
Jedis jedis = new Jedis(host, port);
// Tair インスタンスのパスワード
String authString = jedis.auth(password);// パスワード
if (!authString.equals("OK")) {
System.err.println("AUTH Failed: " + authString);
jedis.close();
return;
}
// 複数のコマンドを連続して実行
final int COUNT=5000;
String key = "KVStore-Tanghan";
// 1. パイプラインを使用しない操作
jedis.del(key);// キーを初期化
Date ts1 = new Date();
for (int i = 0; i < COUNT; i++) {
// リクエストを送信し、レスポンスを受信
jedis.incr(key);
}
Date ts2 = new Date();
System.out.println("Without pipeline > value is:"+jedis.get(key)+" > Time elapsed: " + (ts2.getTime() - ts1.getTime())+ "ms");
// 2. パイプライン操作との比較
jedis.del(key);// キーを初期化
Pipeline p1 = jedis.pipelined();
Date ts3 = new Date();
for (int i = 0; i < COUNT; i++) {
// リクエストを送信
p1.incr(key);
}
// レスポンスを受信
p1.sync();
Date ts4 = new Date();
System.out.println("With pipeline > value is:"+jedis.get(key)+" > Time elapsed: " + (ts4.getTime() - ts3.getTime())+ "ms");
jedis.close();
}
}Tair (Redis OSS 互換) インスタンスの正しいエンドポイントとパスワードを入力して Java プログラムを実行すると、システムは次の出力を返します。この出力は、パイプラインを使用する方がはるかに高速であることを示しています。
Without pipeline > value is:5000 > Time elapsed: 5844ms
With pipeline > value is:5000 > Time elapsed: 78msレスポンスデータの処理方法
Jedis パイプラインからのレスポンスデータは 2 つの方法で処理できます。詳細については、次のコード例をご参照ください。
package pipeline.kvstore.aliyun.com;
import java.util.List;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.Response;
public class PipelineClientTest {
static final String host = "xxxxxxxx.m.cnhza.kvstore.aliyuncs.com";
static final int port = 6379;
static final String password = "password";
public static void main(String[] args) {
Jedis jedis = new Jedis(host, port);
// インスタンスのパスワード
String authString = jedis.auth(password);// パスワード
if (!authString.equals("OK")) {
System.err.println("AUTH Failed: " + authString);
jedis.close();
return;
}
String key = "KVStore-Test1";
jedis.del(key);// 初期化
// 方法 1
Pipeline p1 = jedis.pipelined();
System.out.println("-----Method 1-----");
for (int i = 0; i < 5; i++) {
p1.incr(key);
System.out.println("Pipeline sends request");
}
// リクエスト送信済み。レスポンスの受信を開始。
System.out.println("Requests sent. Start receiving responses.");
List<Object> responses = p1.syncAndReturnAll();
if (responses == null || responses.isEmpty()) {
jedis.close();
throw new RuntimeException("Pipeline error: No response received");
}
for (Object resp : responses) {
System.out.println("Pipeline receives response: " + resp.toString());
}
System.out.println();
// 方法 2
System.out.println("-----Method 2-----");
jedis.del(key);// 初期化
Pipeline p2 = jedis.pipelined();
// 最初に Response オブジェクトの宣言が必要。
Response<Long> r1 = p2.incr(key);
System.out.println("Pipeline sends request");
Response<Long> r2 = p2.incr(key);
System.out.println("Pipeline sends request");
Response<Long> r3 = p2.incr(key);
System.out.println("Pipeline sends request");
Response<Long> r4 = p2.incr(key);
System.out.println("Pipeline sends request");
Response<Long> r5 = p2.incr(key);
System.out.println("Pipeline sends request");
try{
r1.get(); // クライアントはまだレスポンスの受信を開始していないため、この操作は失敗します。
}catch(Exception e){
System.out.println(" <<< Pipeline error: The client has not started to receive responses. >>> ");
}
// リクエスト送信済み。レスポンスの受信を開始。
System.out.println("Requests sent. Start receiving responses.");
p2.sync();
System.out.println("Pipeline receives response: " + r1.get());
System.out.println("Pipeline receives response: " + r2.get());
System.out.println("Pipeline receives response: " + r3.get());
System.out.println("Pipeline receives response: " + r4.get());
System.out.println("Pipeline receives response: " + r5.get());
jedis.close();
}
}Tair (Redis OSS 互換) インスタンスの正しいエンドポイントとパスワードを入力して Java プログラムを実行すると、プログラムは次の出力を返します。
-----Method 1-----
Pipeline sends request
Pipeline sends request
Pipeline sends request
Pipeline sends request
Pipeline sends request
Requests sent. Start receiving responses.
Pipeline receives response: 1
Pipeline receives response: 2
Pipeline receives response: 3
Pipeline receives response: 4
Pipeline receives response: 5
-----Method 2-----
Pipeline sends request
Pipeline sends request
Pipeline sends request
Pipeline sends request
Pipeline sends request
<<< Pipeline error: The client has not started to receive responses. >>>
Requests sent. Start receiving responses.
Pipeline receives response: 1
Pipeline receives response: 2
Pipeline receives response: 3
Pipeline receives response: 4
Pipeline receives response: 5