第7章 グリッドでのコード実行
キャッシュの主な利点は、マシン全体でもキーで値を迅速に検索できることです。実際、この理由だけで、おそらく多くのユーザーが Data Grid を使用しています。ただし、Data Grid には、すぐには明らかにならない多くの利点があります。通常、Data Grid はマシンのクラスターで使用されるため、ユーザーのニーズのワークロードを実行するためにクラスター全体を利用するのに役立つ機能もあります。
7.1. クラスターエグゼキューター
マシンのグループがあるため、それらすべてでコードを実行するためにそれらの結合された計算能力を活用することは理にかなっています。Cache Manager には、クラスター内で任意のコードを実行できる優れたユーティリティーが付属しています。この機能にはキャッシュを使用する必要はありません。この クラスターエグゼキューター は、EmbeddedCacheManager
で executor() を呼び出すことで取得できます。このエグゼキュータは、クラスター設定と非クラスター設定の両方で取得できます。
ClusterExecutor は、コードがキャッシュ内のデータに依存しないコードを実行するために特別に設計されており、代わりに、ユーザーがクラスター内でコードを簡単に実行できるようにする方法として使用されます。
このマネージャーは Java のストリーミング API を中心に構築されているため、すべてのメソッドは関数インターフェイスを引数として受け取ります。また、これらの引数は他のノードに送信されるため、シリアライズする必要があります。ラムダがすぐに Serializable になるような策を使用しています。つまり、引数に Serializable と実際の引数タイプ (つまり、Runnable または Function) の両方を実装させることです。JRE は、呼び出す方法を決定する際に最も具体的なクラスを選択するため、ラムダは常にシリアライズ可能です。また、Externalizer を使用してメッセージサイズをさらに減らすこともできます。
マネージャーはデフォルトで、指定されたコマンドを、送信元のノードを含むクラスター内のすべてのノードに送信します。セクションで説明されているように、filterTargets
メソッドを使用して、タスクが実行するノードを制御できます。
7.1.1. 実行ノードのフィルタリング
コマンドを実行するノードを制限できます。たとえば、同じラック内のマシンでのみ計算を実行したい場合があります。または、ローカルサイトで 1 回、別のサイトで操作を再実行することもできます。クラスターエグゼキューターは、同じマシン、ラック、またはサイトレベルのスコープで要求を送信するノードを制限できます。
SameRack.java
EmbeddedCacheManager manager = ...; manager.executor().filterTargets(ClusterExecutionPolicy.SAME_RACK).submit(...)
このトポロジーベースフィルタリングを使用するには、サーバーヒントを介してトポロジー対応のコンシステントハッシュを有効にする必要があります。
ノードの Address
に基づいて述部を使用してフィルタリングすることもできます。これは任意で、以前のコードスニペットでトポロジーベースのフィルタリングと組み合わせることもできます。
また、実行対象と見なすことができるノードを除外する Predicate
を使用して、任意の方法でターゲットノードを選択することもできます。これは同時に Topology フィルタリングと組み合わせて、クラスター内でコードを実行する場所をより詳細に制御できるようにすることもできます。
Predicate.java
EmbeddedCacheManager manager = ...; // Just filter manager.executor().filterTargets(a -> a.equals(..)).submit(...) // Filter only those in the desired topology manager.executor().filterTargets(ClusterExecutionPolicy.SAME_SITE, a -> a.equals(..)).submit(...)
7.1.2. Timeout
クラスターエグゼキューターを使用すると、呼び出しごとにタイムアウトを設定できます。デフォルトは、Transport Configuration で設定された分散同期のタイムアウトになります。このタイムアウトは、クラスター化されたキャッシュマネージャーとクラスター化されていないキャッシュマネージャーの両方で機能します。タイムアウトの期限が切れると、エグゼキューターがタスクを実行しているスレッドを中断する場合と中断しない場合があります。ただし、タイムアウトが発生すると、Consumer
または Future
は TimeoutException
を渡して完了します。この値は、timeout メソッドを呼び出し、希望の期間を指定することで、オーバーライドすることができます。
7.1.3. 単一ノードの提出
クラスターエグゼキューターは、すべてのノードにコマンドを送信する代わりに、単一ノード送信モードで実行することもできます。代わりに、通常はコマンドを受信するノードの 1 つを選択し、1 つだけに送信します。それぞれの送信は、別のノードを使用してタスクが実行される可能性があります。これは、ClusterExecutor が実装する java.util.concurrent.Executor
として ClusterExecutor を使用するのが非常に便利です。
SingleNode.java
EmbeddedCacheManager manager = ...; manager.executor().singleNodeSubmission().submit(...)
7.1.3.1. Failover
シングルノード送信で実行する場合は、コマンドを再試行することにより、特定のコマンドの処理中に例外が発生した場合にクラスターエグゼキューターが処理できるようにすることが望ましい場合があります。これが発生すると、クラスターエグゼキューターは単一のノードを再度選択し、任意のフェイルオーバー試行までコマンドを再実行します。選択したノードは、トポロジーまたは述部のチェックをパスするノードである可能性があることに注意してください。フェイルオーバーは、オーバーライドされた singleNodeSubmission メソッドを呼び出すことで有効になります。指定されたコマンドは、コマンドが例外なく完了するか、送信の合計量が指定されたフェイルオーバーカウントと等しくなるまで、単一のノードに再送信されます。
7.1.4. 例: PI アプローチ
この例は、ClusterExecutor を使用して PI の値を見積もる方法を示しています。
Pi 近似は、クラスターエグゼキューターを介した並列分散実行から大きな利点を得ることができます。正方形の面積は Sa = 4r2 であり、円の面積は Ca=pi*r2 であることを思い出してください。2 つ目の式からの r2 を置き換えると、pi = 4 * Ca/S になります。ここで、正方形に非常に多くのダーツを射ることができると仮定して、射ったダーツの総数に対して円の中に入ったダーツの割合を取ると、Ca/Sa の値が近似します。pi = 4 * Ca/Sa であるため、pi の近似値を簡単に導き出すことができます。ダーツを多く撃つほど、より良い近似が得られます。以下の例では、10 億本のダーツを撃ちますが、それらを連続して撃つのではなく、Data Grid クラスター全体でダーツ射撃の作業を並列化します。これは 1 のクラスターで正常に機能しますが、遅くなることに注意してください。
public class PiAppx { public static void main (String [] arg){ EmbeddedCacheManager cacheManager = .. boolean isCluster = .. int numPoints = 1_000_000_000; int numServers = isCluster ? cacheManager.getMembers().size() : 1; int numberPerWorker = numPoints / numServers; ClusterExecutor clusterExecutor = cacheManager.executor(); long start = System.currentTimeMillis(); // We receive results concurrently - need to handle that AtomicLong countCircle = new AtomicLong(); CompletableFuture<Void> fut = clusterExecutor.submitConsumer(m -> { int insideCircleCount = 0; for (int i = 0; i < numberPerWorker; i++) { double x = Math.random(); double y = Math.random(); if (insideCircle(x, y)) insideCircleCount++; } return insideCircleCount; }, (address, count, throwable) -> { if (throwable != null) { throwable.printStackTrace(); System.out.println("Address: " + address + " encountered an error: " + throwable); } else { countCircle.getAndAdd(count); } }); fut.whenComplete((v, t) -> { // This is invoked after all nodes have responded with a value or exception if (t != null) { t.printStackTrace(); System.out.println("Exception encountered while waiting:" + t); } else { double appxPi = 4.0 * countCircle.get() / numPoints; System.out.println("Distributed PI appx is " + appxPi + " using " + numServers + " node(s), completed in " + (System.currentTimeMillis() - start) + " ms"); } }); // May have to sleep here to keep alive if no user threads left } private static boolean insideCircle(double x, double y) { return (Math.pow(x - 0.5, 2) + Math.pow(y - 0.5, 2)) <= Math.pow(0.5, 2); } }