チャレンジ

書き込み 並行プログラム は難しい。スレッド、ロック、競合状態などを処理する必要があると、エラーが発生しやすくなり、コードの読み取り、テスト、および保守が困難になる可能性があります。

したがって、多くの人はマルチスレッドを完全に避けることを好みます。代わりに、シングルスレッドプロセスのみを採用し、外部サービス（データベース、キューなど）に依存して、必要な同時または非同期操作を処理します。このアプローチは正当な代替手段である場合もありますが、それが単に実行可能なオプションではない多くのシナリオがあります。取引や銀行のアプリケーション、リアルタイムゲームなど、多くのリアルタイムシステムには、シングルスレッドプロセスが完了するのを待つ余裕がありません（今すぐ答えが必要です！）。他のシステムは、計算またはリソースを大量に消費するため、コードに並列化を導入せずに実行するのに非常に長い時間（場合によっては数時間または数日）かかります。

かなり一般的なシングルスレッドアプローチの1つ（ Node.js たとえば、世界）は、イベントベースの非ブロッキングパラダイムを使用することです。これは、コンテキストスイッチ、ロック、およびブロックを回避することでパフォーマンスを向上させますが、複数のプロセッサを同時に使用する問題には対処しません（そのためには、複数の独立したプロセスを起動して調整する必要があります）。

デザインの5つの原則

つまり、並行アプリケーションを構築するために、スレッド、ロック、および競合状態の内部に深く入り込むしかないということですか？

Akkaフレームワークのおかげで、答えはノーです。このチュートリアルでは、Akkaの例を紹介し、並行分散アプリケーションの実装を容易にし、簡素化する方法を探ります。

Akkaフレームワークとは何ですか？

この投稿では、Akkaを紹介し、並行分散アプリケーションの実装を容易にし、簡素化する方法を探ります。

Akka は、JVM上で高度に同時、分散、およびフォールトトレラントなアプリケーションを構築するためのツールキットおよびランタイムです。アッカはで書かれています はしご 、ScalaとJavaの両方に言語バインディングが提供されています。

並行性を処理するためのAkkaのアプローチは、 アクターモデル 。アクターベースのシステムでは、すべてがオブジェクト指向設計のオブジェクトであるのとほぼ同じように、すべてがアクターです。ただし、主な違いは、特に私たちの議論に関連して、アクターモデルが並行モデルとして機能するように特別に設計および設計されているのに対し、オブジェクト指向モデルはそうではないことです。より具体的には、Scalaアクターシステムでは、アクターは、連続性を前提とせずに、相互作用して情報を共有します。アクターが互いに情報を共有し、互いにタスクを実行するメカニズムは、メッセージパッシングです。

スレッドの作成とスケジューリング、メッセージの受信とディスパッチ、競合状態と同期の処理のすべての複雑さは、透過的に処理するためのフレームワークに委ねられています。

Akkaは、アクターがメッセージを処理するだけでよいように、アクターと基盤となるシステムの間にレイヤーを作成します。スレッドの作成とスケジューリング、メッセージの受信とディスパッチ、競合状態と同期の処理のすべての複雑さは、透過的に処理するためのフレームワークに委ねられています。

アッカは厳守します 反応性マニフェスト 。リアクティブアプリケーションは、従来のマルチスレッドアプリケーションを、次の要件の1つ以上を満たすアーキテクチャに置き換えることを目的としています。

• イベント駆動型。 アクターを使用すると、リクエストを非同期で処理し、非ブロッキング操作を排他的に使用するコードを記述できます。
• スケーラブル。 Akkaでは、メッセージパッシングと場所の透過性の両方のおかげで、コードを変更せずにノードを追加できます。
• 弾力性。 すべてのアプリケーションでエラーが発生し、ある時点で失敗します。 Akkaは、自己回復システムを促進するための「監視」（フォールトトレランス）戦略を提供します。
• レスポンシブ。 今日の高性能で迅速な応答アプリケーションの多くは、ユーザーに迅速なフィードバックを提供する必要があるため、非常にタイムリーにイベントに対応する必要があります。 Akkaのノンブロッキング、メッセージベースの戦略はこれを達成するのに役立ちます。

アッカの俳優とは何ですか？

アクターは本質的に、メッセージを受信し、それらを処理するためのアクションを実行するオブジェクトにすぎません。メッセージの送信元から切り離されており、受信したメッセージのタイプを適切に認識し、それに応じてアクションを実行することが唯一の責任です。

メッセージを受信すると、アクターは次の1つ以上のアクションを実行できます。

• 一部の操作自体を実行します（計算の実行、データの永続化、外部Webサービスの呼び出しなど）
• メッセージまたは派生メッセージを別のアクターに転送する
• 新しいアクターをインスタンス化し、メッセージを転送します

あるいは、アクターは、メッセージを無視することが適切であると判断した場合、メッセージを完全に無視することを選択できます（つまり、何もしないことを選択できます）。

アクターを実装するには、akka.actor.Actorトレイトを拡張し、receiveメソッドを実装する必要があります。アクターのreceiveメソッドは、メッセージがそのアクターに送信されるときに（Akkaによって）呼び出されます。その典型的な実装は、次のAkkaの例に示すように、メッセージタイプを識別し、それに応じて反応するパターンマッチングで構成されています。

import akka.actor.Actor import akka.actor.Props import akka.event.Logging class MyActor extends Actor { def receive = { case value: String => doSomething(value) case _ => println('received unknown message') } }

パターンマッチングは、メッセージを処理するための比較的洗練された手法であり、コールバックに基づく同等の実装よりも「クリーン」でナビゲートしやすいコードを生成する傾向があります。たとえば、単純なHTTP要求/応答の実装について考えてみます。

まず、JavaScriptでコールバックベースのパラダイムを使用してこれを実装しましょう。

route(url, function(request){ var query = buildQuery(request); dbCall(query, function(dbResponse){ var wsRequest = buildWebServiceRequest(dbResponse); wsCall(wsRequest, function(wsResponse) { sendReply(wsResponse); }); }); });

次に、これをパターンマッチングベースの実装と比較してみましょう。

msg match { case HttpRequest(request) => { val query = buildQuery(request) dbCall(query) } case DbResponse(dbResponse) => { var wsRequest = buildWebServiceRequest(dbResponse); wsCall(dbResponse) } case WsResponse(wsResponse) => sendReply(wsResponse) }

コールバックベースのJavaScriptコードは確かにコンパクトですが、読みやすく、ナビゲートするのは確かに困難です。比較すると、パターンマッチングベースのコードを使用すると、どのケースが考慮され、それぞれがどのように処理されているかがすぐにわかります。

アクターシステム

複雑な問題を取り上げ、それをより小さなサブ問題に再帰的に分割することは、一般的に適切な問題解決手法です。このアプローチは、コンピュータサイエンスで特に有益です（ 単一責任の原則 ）、冗長性がほとんどまたはまったくない、クリーンでモジュール化されたコードを生成する傾向があるため、保守が比較的簡単です。

アクターベースの設計では、この手法を使用すると、アクターを論理的に編成して、 アクターシステム 。アクターシステムは、アクターが相互作用するためのインフラストラクチャを提供します。

Akkaでは、アクターと通信する唯一の方法はActorRefを使用することです。 ActorRef他のオブジェクトがそのアクターの内部および状態に直接アクセスまたは操作することを妨げるアクターへの参照を表します。メッセージはActorRefを介してアクターに送信できます次の構文プロトコルのいずれかを使用します。

• ! （「伝える」）–メッセージを送信し、すぐに戻ります
• ? （「尋ねる」）–メッセージを送信し、 未来 可能な返信を表す

各アクターには、受信メッセージの配信先となるメールボックスがあります。選択できるメールボックスの実装は複数あり、デフォルトの実装はFIFOです。

アクターには、複数のメッセージを処理しながら状態を維持するための多くのインスタンス変数が含まれています。 Akkaは、アクターの各インスタンスが独自の軽量スレッドで実行され、メッセージが一度に1つずつ処理されるようにします。このようにして、開発者が同期や競合状態について明示的に心配することなく、各アクターの状態を確実に維持できます。

各アクターには、Akka ActorAPIを介してタスクを実行するための次の有用な情報が提供されます。

• sender：ActorRef現在処理中のメッセージの送信者へ
• context：アクターが実行されているコンテキストに関連する情報とメソッド（たとえば、新しいアクターをインスタンス化するためのactorOfメソッドを含む）
• supervisionStrategy：エラーからの回復に使用する戦略を定義します
• self：ActorRef俳優自身のために

Akkaは、アクターの各インスタンスが独自の軽量スレッドで実行され、メッセージが一度に1つずつ処理されるようにします。このようにして、開発者が同期や競合状態について明示的に心配する必要なしに、各アクターの状態を確実に維持できます。

これらのチュートリアルを結び付けるために、テキストファイル内の単語数を数える簡単な例を考えてみましょう。

Akkaの例では、問題を2つのサブタスクに分解します。つまり、（1）1行の単語数をカウントする「子」タスクと、（2）1行あたりの単語数を合計してファイル内の単語の総数を取得する「親」タスクです。

親アクターはファイルから各行をロードしてから、その行の単語をカウントするタスクを子アクターに委任します。子が完了すると、結果とともにメッセージが親に返送されます。親は（各行の）単語数を含むメッセージを受信し、ファイル全体の単語の総数のカウンターを保持します。これは、完了時に呼び出し元に返されます。

（以下に示すAkkaチュートリアルコードサンプルは教訓的なもののみを目的としているため、必ずしもすべてのエッジ条件やパフォーマンスの最適化などに関係するわけではないことに注意してください。また、以下に示すコードサンプルの完全なコンパイル可能なバージョンは、この 要旨 。）

レスポンシブウェブデザインの画面サイズ

まず、子の実装例を見てみましょうStringCounterActorクラス：

case class ProcessStringMsg(string: String) case class StringProcessedMsg(words: Integer) class StringCounterActor extends Actor { def receive = { case ProcessStringMsg(string) => { val wordsInLine = string.split(' ').length sender ! StringProcessedMsg(wordsInLine) } case _ => println('Error: message not recognized') } }

このアクターのタスクは非常に単純です。消費ProcessStringMsgメッセージ（テキスト行を含む）、指定された行の単語数をカウントし、StringProcessedMsgを介して結果を送信者に返します。メッセージ。 !を使用するようにクラスを実装したことに注意してください。 （「tell」）StringProcessedMsgを送信するメソッドメッセージ（つまり、メッセージを送信してすぐに戻る）。

では、親に注意を向けましょうWordCounterActorクラス：

1. case class StartProcessFileMsg() 2. 3. class WordCounterActor(filename: String) extends Actor { 4. 5. private var running = false 6. private var totalLines = 0 7. private var linesProcessed = 0 8. private var result = 0 9. private var fileSender: Option[ActorRef] = None 10. 11. def receive = { 12. case StartProcessFileMsg() => { 13. if (running) { 14. // println just used for example purposes; 15. // Akka logger should be used instead 16. println('Warning: duplicate start message received') 17. } else { 18. running = true 19. fileSender = Some(sender) // save reference to process invoker 20. import scala.io.Source._ 21. fromFile(filename).getLines.foreach { line => 22. context.actorOf(Props[StringCounterActor]) ! ProcessStringMsg(line) 23. totalLines += 1 24. } 25. } 26. } 27. case StringProcessedMsg(words) => { 28. result += words 29. linesProcessed += 1 30. if (linesProcessed == totalLines) { 31. fileSender.map(_ ! result) // provide result to process invoker 32. } 33. } 34. case _ => println('message not recognized!') 35. } 36. }

ここでは多くのことが起こっているので、それぞれをさらに詳しく調べてみましょう。 （以下の説明で参照されている行番号は、上記のコードサンプルに基づいていることに注意してください） ..。

まず、処理するファイルの名前がWordCounterActorに渡されることに注意してください。コンストラクター（3行目）。これは、アクターが単一のファイルの処理にのみ使用されることを示しています。これにより、ジョブの完了時に状態変数（running、totalLines、linesProcessed、およびresult）をリセットする必要がなくなるため、開発者のコ​​ーディングジョブも簡素化されます。インスタンスは一度だけ使用され（つまり、単一のファイルを処理するため）、その後破棄されるためです。

次に、WordCounterActorに注意してください。次の2種類のメッセージを処理します。

• StartProcessFileMsg （12行目）
  • WordCounterActorを最初に開始する外部アクターから受信します。
  • 受信すると、WordCounterActor最初に、冗長な要求を受信して​​いないことを確認します。
  • リクエストが冗長な場合、WordCounterActor警告を生成し、それ以上何も行われません（16行目）。
  • リクエストが冗長でない場合：
    • WordCounterActor送信者への参照をfileSenderに格納しますインスタンス変数（これはOption[ActorRef]ではなくOption[Actor]であることに注意してください-9行目を参照）。これActorRef後で最終的なStringProcessedMsgを処理するときにアクセスして応答するために必要です（これは、以下で説明するように、StringCounterActorの子から受信されます）。
    • WordCounterActor次にファイルを読み取り、ファイルの各行がロードされると、StringCounterActor子が作成され、処理される行を含むメッセージが子に渡されます（21〜24行目）。
• StringProcessedMsg （27行目）
  • 子供から受け取ったStringCounterActor割り当てられた行の処理が完了したとき。
  • 受信すると、WordCounterActorファイルの行カウンターをインクリメントし、ファイル内のすべての行が処理された場合（つまり、totalLinesとlinesProcessedが等しい場合）、最終結果を元のfileSenderに送信します。 （28〜31行目）。

繰り返しになりますが、Akkaでは、アクター間の通信の唯一のメカニズムはメッセージパッシングであることに注意してください。アクターが共有するのはメッセージだけです。アクターは同じメッセージに同時にアクセスできる可能性があるため、競合状態や予期しない動作を回避するために、メッセージが不変であることが重要です。

ケースクラス Scalaには、パターンマッチングを介して再帰的な分解メカニズムを提供する通常のクラスがあります。

したがって、メッセージはデフォルトで不変であり、パターンマッチングとシームレスに統合されるため、ケースクラスの形式でメッセージを渡すのが一般的です。

アプリ全体を実行するためのコードサンプルで例を締めくくりましょう。

object Sample extends App { import akka.util.Timeout import scala.concurrent.duration._ import akka.pattern.ask import akka.dispatch.ExecutionContexts._ implicit val ec = global override def main(args: Array[String]) { val system = ActorSystem('System') val actor = system.actorOf(Props(new WordCounterActor(args(0)))) implicit val timeout = Timeout(25 seconds) val future = actor ? StartProcessFileMsg() future.map { result => println('Total number of words ' + result) system.shutdown } } } 並行プログラミングでは、「future」は本質的に、まだ知られていない結果のプレースホルダーオブジェクトです。

今回は?に注目してくださいメソッドはメッセージの送信に使用されます。このようにして、発信者は返されたを使用できます 未来 これが利用可能な場合に最終結果を出力し、ActorSystemをシャットダウンしてプログラムを終了します。

Akkaのフォールトトレランスとスーパーバイザー戦略

アクターシステムでは、各アクターはその子のスーパーバイザーです。アクターがメッセージの処理に失敗した場合、アクターは自身とそのすべての子を一時停止し、通常は例外の形式でスーパーバイザーにメッセージを送信します。

ソフトウェアプロジェクト管理におけるコスト見積もり

Akkaでは、スーパーバイザー戦略は、システムのフォールトトレラントな動作を定義するための主要で直接的なメカニズムです。

Akkaでは、スーパーバイザーがその子からそこに浸透する例外に反応して処理する方法は、スーパーバイザー戦略と呼ばれます。 スーパーバイザー戦略 は、システムのフォールトトレラントな動作を定義するための主要で直接的なメカニズムです。

障害を示すメッセージがスーパーバイザーに到達すると、次のいずれかのアクションを実行できます。

• 内部状態を維持しながら、子（およびその子）を再開します。 この戦略は、子の状態がエラーによって破損しておらず、正しく機能し続けることができる場合に適用できます。
• 子（およびその子）を再起動して、内部状態をクリアします。 この戦略は、今説明したものとは逆のシナリオで使用できます。子の状態がエラーによって破損している場合は、将来使用する前に状態をリセットする必要があります。
• 子供（およびその子供）を永久に停止します。 この戦略は、エラー状態が修正可能であるとは考えられないが、実行中の残りの操作を危険にさらさない場合に使用できます。これは、失敗した子がなくても完了できます。
• 自分自身を停止し、エラーをエスカレーションします。 スーパーバイザーが障害の処理方法を知らないため、自身のスーパーバイザーにエスカレーションする場合に使用されます。

さらに、アクターは、失敗した子またはその兄弟だけにアクションを適用することを決定できます。これには、2つの事前定義された戦略があります。

• OneForOneStrategy：指定されたアクションを失敗した子にのみ適用します
• AllForOneStrategy：指定されたアクションをそのすべての子に適用します

OneForOneStrategyを使用した簡単な例を次に示します。

import akka.actor.OneForOneStrategy import akka.actor.SupervisorStrategy._ import scala.concurrent.duration._ override val supervisorStrategy = OneForOneStrategy() { case _: ArithmeticException => Resume case _: NullPointerException => Restart case _: IllegalArgumentException => Stop case _: Exception => Escalate }

戦略が指定されていない場合、次のデフォルト戦略が採用されます。

• アクターの初期化中にエラーが発生した場合、またはアクターが強制終了された場合、アクターは停止されます。
• 他の種類の例外があった場合、アクターは単に再起動されます。

このデフォルト戦略のAkka提供の実装は次のとおりです。

final val defaultStrategy: SupervisorStrategy = { def defaultDecider: Decider = { case _: ActorInitializationException ⇒ Stop case _: ActorKilledException ⇒ Stop case _: Exception ⇒ Restart } OneForOneStrategy()(defaultDecider) }

Akkaはの実装を可能にします カスタムスーパーバイザー戦略 ただし、Akkaのドキュメントで警告されているように、実装が正しくないとアクターシステムがブロックされる（つまり、アクターが永続的に停止される）などの問題が発生する可能性があるため、注意して行ってください。

場所の透明性

Akkaアーキテクチャはサポートします 場所の透明性 、アクターが受信したメッセージの発信元を完全に認識できないようにします。メッセージの送信者は、アクターと同じJVMに存在する場合と、別のJVM（同じノードまたは別のノードで実行されている）に存在する場合があります。 Akkaを使用すると、これらの各ケースを、アクター（したがって開発者）に対して完全に透過的な方法で処理できます。唯一の注意点は、複数のノードを介して送信されるメッセージはシリアル化可能でなければならないということです。

Akkaアーキテクチャは場所の透過性をサポートしているため、アクターは受信したメッセージの発信元を完全に認識できません。

アクターシステムは、特別なコードを必要とせずに分散環境で実行するように設計されています。 Akkaは、メッセージの送信先のノードを指定する構成ファイル（application.conf）の存在のみを必要とします。構成ファイルの簡単な例を次に示します。

akka { actor { provider = 'akka.remote.RemoteActorRefProvider' } remote { transport = 'akka.remote.netty.NettyRemoteTransport' netty { hostname = '127.0.0.1' port = 2552 } } }

いくつかの別れのヒント…

Akkaフレームワークが並行性と高性能の実現にどのように役立つかを見てきました。ただし、このチュートリアルで指摘したように、Akkaの能力を最大限に活用するために、システムを設計および実装する際に留意すべき点がいくつかあります。

• 可能な限り、各アクターには可能な限り最小のタスクを割り当てる必要があります（前述のように、 単一責任の原則 ）

• アクターはイベント（つまり、メッセージの処理）を非同期で処理し、ブロックしないでください。そうしないと、パフォーマンスに悪影響を与える可能性のあるコンテキストスイッチが発生します。具体的には、アクターをブロックしないように、将来的にブロック操作（IOなど）を実行するのが最善です。つまり：

  case evt => blockingCall() // BAD case evt => Future { blockingCall() // GOOD }
• メッセージを相互に渡すアクターはすべて、独自のスレッドで同時に実行されるため、メッセージがすべて不変であることを確認してください。変更可能なメッセージは、予期しない動作を引き起こす可能性があります。
• ノード間で送信されるメッセージはシリアル化可能である必要があるため、メッセージが大きいほど、メッセージのシリアル化、送信、および逆シリアル化に時間がかかり、パフォーマンスに悪影響を与える可能性があることに注意してください。

結論

アッカ、で書かれた はしご は、高度に並行した分散型のフォールトトレラントアプリケーションの開発を簡素化および促進し、開発者から複雑さの多くを隠します。 Akkaの完全な正義を行うには、この1つのチュートリアルよりもはるかに多くのことが必要になりますが、この紹介とその例が、もっと読みたくなるほど魅力的であったことを願っています。

Amazon、VMWare、およびCSCは、Akkaを積極的に使用している大手企業のほんの一例です。訪問 アッカ公式サイト 詳細を学び、Akkaがあなたのプロジェクトにとっても正しい答えであるかどうかを探求するために。