原文はこちら。
https://community.oracle.com/community/cloud_computing/oracle-cloud-developer-solutions/blog/2017/01/05/microservices-messaging-on-oracle-cloud-using-apache-kafka

この記事は2部構成の第1部です。Oracle Cloud Platformと幅広く使われているオープンソーステクノロジーを組み合わせ、サンプルアプリケーション（ソースコードはこちら）を使って、マイクロサービス間を疎結合の非同期メッセージベースで連携する方法を紹介します。

Cloud Developer Portal Solutions
https://cloud.oracle.com/developer/solutions

第1部では、以下の内容を取り扱います。

個々のマイクロサービスの開発
非同期メッセージングを利用した疎結合連携
Oracle Cloud Serviceのセットアップおよびデプロイ

第2部ではマイクロサービスのスケールについて取り扱います。

利用するコンポーネント

Oracle Cloud

以下のOracle Cloud Serviceを使います。

	Description
Application Container Cloud	Java SEマイクロサービスをデプロイ可能な、スケーラブルなプラットフォーム
Compute Cloud	Kafka クラスタ (broker) をホストするために利用

オープンソース

以下のオープンソーステクノロジーを使ってサンプルアプリケーションを構築します。

	Description
Apache Kafka	スケーラブルなPub-Subメッセージハブ
Jersey	RESTおよびSSEサービスを実装するために利用。（プラガブル）ランタイム/コンテナとしてGrizzlyを利用
Maven	（assemblyプラグインとともに）標準のJavaビルドツールとして利用。

Microservicesにおけるメッセージング

マイクロサービスベースのシステムは複数のアプリケーション（サービス）から構成されており、これらのサービスは、通常システム全体の特殊な側面（ビジネスシナリオ）にフォーカスしています。個々のサービスは、相互に連携せず独立して機能することもできますが、実際のところサービスは孤立して機能することはできず、お互いに通信して仕事を終わらせる必要があります。マイクロサービス間の通信を実装する上で複数の戦略を使うことができ、同期 vs 非同期スタイル、コレオグラフィ vs オーケストレーション、REST（HTTP）vs メッセージングといったくくりで分類されることがよくあります。

サンプルアプリケーションのアーキテクチャ

この記事のサンプルアプリケーションのユースケースはシンプルで、ランダムに生成されたデータ（Producerマイクロサービス）を別のエンティティ（Consumerマイクロサービス）が受け取り、最終的にはリアルタイムでユーザーが確認できるようにします。

第1部では、高可用性の設定を考慮しておらず、1個のKafkaノードです。つまり、Kafkaクラスタに1個のサーバのみ存在し、ProducerとConsumerマイクロサービス（両方とも1インスタンスのみ）をApplication Container Cloudにデプロイします。

上図で図示されている個々のコンポーネントを見ていきましょう。

Apache Kafka

Apache Kafkaは、「メッセージングシステムまたは分散コミットログとして実装されたストリーミングプラットフォーム」として広く知られています。もっと簡単な説明があればいいのですが。

基本
KafkaはJVM上で動作するScalaで記述された、Publish-Subscribeベースのメッセージングシステムで、パブリッシャ（Publisher）がトピック（Topic）にデータを書き込み、コンシューマ（Consumer）がこれらのトピックをポーリングしてデータを取得します。
分散指向
コンポーネント（broker、パブリッシャ、コンシューマ）はスケールアウト可能なように設計されています。
マスタ・スレーブアーキテクチャ
トピック内のデータはクラスタ内の複数ノード間に（レプリケーション・ファクタに基づいて）分散されます。1個のノードのみが特定データのマスタとして機能し、0個以上のノードがそのデータのコピーを持つ、つまりフォロワーとして動作することができます。
パーティション
トピックはさらにパーティション（Partition）に分割され、各パーティションは基本的にデータ（Key-Valueのペア）を格納するコミット・ログ（commit log）として振る舞います。データは不変で、厳密に順序付け（オフセットが各データエントリに割り当てられます）された上で、（構成に基づいて）永続化され、ディスクに保持されます。
適した領域
Kafkaは大容量の高速なリアルタイムストリーミングデータの処理に適しています。
JMSではない
JMSと似てはいますが違います。JMS仕様を実装したものではなく、JMSベースのソリューションを置き換えるようなものではありません。

Kafka brokerはKafkaサーバプロセス（ノード）以外の何者でもありません。複数のノードで、分散された、フォールトトレラントおよび水平方向にスケール可能なメッセージハブとして機能するクラスタを構成することができます。

Producerマイクロサービス

Producerマイクロサービスは、Kafka Java APIとJersey （JAX-RS参照実装）を使います。リアルタイムデータのPub-Subパイプラインを紹介するため、このマイクロサービスはハイペースでサンプルイベントを発行します。

サンプルデータ

Producerが生成したデータをメトリックのモデルとします。今回は、特定のマシンのCPU使用率であり、単純なキーと値のペア（名前、使用率など）と見做すことができます。以下はその例です（属性Partitionは無視してください）

: Partition 0 event: machine-2 id: 19 data: 14% : Partition 1 event: machine-1 id: 20 data: 5%

Consumerマイクロサービス

このシステムの2個目のマイクロサービスです。Producerと同様に、JerseyとKafkaのJava （Consumer） APIを使っています。このサービスで利用しているJerseyコンポーネントの中で注目すべきは、Server Sent Eventsモジュールです。このモジュールを使うと、サンプルアプリケーションで必要なsubscribe-and-broadcastセマンティクスを実装が容易になります（詳細は後で説明します）。

両マイクロサービスは個別のアプリケーションとしてApplication Container Cloud Serviceにデプロイされ、独立して管理、スケールすることができます。

Oracle Compute CloudへのApache Kafkaの構築

複数の方法でApache KafkaをOracle Compute Cloud （IaaS）上に構築できます。

Oracle Cloud Marketplaceの利用

マーケットプレイスのApache KafkaのBitnamiイメージを使います。詳細ドキュメントは以下のURLをご覧ください。

Oracle® Cloud
Using Oracle Compute Cloud Service (IaaS)
Creating an Instance from Oracle Cloud Marketplace
http://docs.oracle.com/cloud/latest/stcomputecs/STCSG/GUID-D7CE3F3E-CC1B-443B-BAF9-E0F6B4FD7762.htm#STCSG-GUID-D7CE3F3E-CC1B-443B-BAF9-E0F6B4FD7762

Oracle Compute CloudのVMを使う

お好みのOSが動作するCompute CloudのVMをプロビジョニングするところから始めましょう。以下のドキュメントが役にたつことでしょう。

Oracle® Cloud
Using Oracle Compute Cloud Service (IaaS)
Workflow for Creating Your First Instance
http://docs.oracle.com/cloud/latest/stcomputecs/STCSG/GUID-DD966FCF-624B-45A3-88AF-8EC123CCBCFC.htm#STCSG-GUID-DD966FCF-624B-45A3-88AF-8EC123CCBCFC

VMへのSSHアクセスの有効化

構成を進める上で、まずOracle Compute Cloud VMへのSSHアクセスを有効化（セキュリティポリシーやセキュリティルールを作成）する必要があります。Oracle LinuxやOracle SolarisベースのVMに対する手順はそれぞれ以下を確認してください。

Oracle® Cloud
Using Oracle Compute Cloud Service (IaaS)
Accessing an Oracle Linux Instance Using SSH
http://docs.oracle.com/cloud/latest/stcomputecs/STCSG/GUID-D947E2CC-0D4C-43F4-B2A9-A517037D6C11.htm#STCSG-GUID-D947E2CC-0D4C-43F4-B2A9-A517037D6C11
Accessing an Oracle Solaris Instance Using SSH
http://docs.oracle.com/cloud/latest/stcomputecs/STCSG/GUID-C72CC64D-73C7-4150-8B5A-EB39D7F72327.htm#STCSG-GUID-C72CC64D-73C7-4150-8B5A-EB39D7F72327

VMへのKafkaのインストール

このセクションではOracle LinuxベースのVMを前提としています。

以下のコマンドを使います。

sudo yum install java-1.8.0-openjdk
sudo yum install wget
mkdir -p ~/kafka-download
wget "http://redrockdigimark.com/apachemirror/kafka/0.10.1.0/kafka_2.11-0.10.1.0.tgz" -O ~/kafka-download/kafka-binary.tgz
mkdir -p ~/kafka-install && cd ~/kafka-install
tar -xvzf ~/kafka-download/kafka-binary.tgz --strip 1

Kafkaのリスナーポートをオープン

Oracle Application Container Cloudにデプロイされたマイクロサービスのため、Kafka brokerサービスへのアクセス（今回の場合は9092/tcp）を許可する必要があります。以下のドキュメントにユースケースの形ですばらしいリファレンスが提供されています。

Oracle® Cloud
Using Oracle Compute Cloud Service (IaaS)
Setting Up Firewalls and Opening Ports for a Sample Scenario
http://docs.oracle.com/cloud/latest/stcomputecs/STCSG/GUID-DE568AAF-39CE-462C-B605-B96AE4036825.htm#OCSUG292

Security Applicationを作成し、プロトコルと対応するポートを指定します。詳細は以下のドキュメントをご覧ください。

Oracle® Cloud
Using Oracle Compute Cloud Service (IaaS)
Creating a Security Application
http://docs.oracle.com/cloud/latest/stcomputecs/STCSG/GUID-4B073EF2-0D7C-4AD8-A40A-585C4E2F938C.htm#OCSUG183

先ほど作成したSecurity Applicationを参照し、Security Ruleを作成します。この設定で、（ルールで定義したように）パブリック・インターネットから9092/tcpへのトラフィックを（Security Application構成毎に）許可します。詳細は以下のドキュメントを参照してください。

Oracle® Cloud
Using Oracle Compute Cloud Service (IaaS)
Creating a Security Rule
http://docs.oracle.com/cloud/latest/stcomputecs/STCSG/GUID-630622EC-160B-4523-88AD-F7B46463A0BE.htm#GUID-A3BEB363-C262-4F1E-909D-AA73AF2D65C9

以下のような構成になるはずです。

Kafka brokerの構成

Compute Cloud環境用にKafkaサーバのプロパティファイル（<KAFKA_INSTALL>/config/server.properties）中の以下の属性を編集します。

Compute CloudインスタンスのパブリックDNS

パブリックIPが140.44.88.200であれば、パブリックIPに対応するFQDNはoc-140-44-88-200.compute.oraclecloud.comになります。

属性	値
listeners	PLAINTEXT://<oracle-compute-private-IP>:<kafka-listen-port> （例） PLAINTEXT://10.190.210.199:9092
advertised.listeners	PLAINTEXT://<oracle-compute-public-DNS>:<kafka-listen-port> （例） PLAINTEXT://oc-140-44-88-200.compute.oraclecloud.com:9092

以下はserver.propertiesファイルのスナップショットです。

KAFKA_INSTALL/bin/zookeeper-server-start.sh config/zookeeper.propertiesを実行して、Zookeeperを起動します。

KAFKA_INSTALL/bin/kafka-server-start.sh config/server.propertiesを実行して、Kafka Brokerを起動します。

Zookeeper起動前にKafka brokerを起動しないでください。

ソリューション概要

イベントフロー・シーケンス

これらのコンポーネントが協調してどのようにユースケース全体をサポートするのかを説明します。

ProducerがKafka brokerにイベントをPush

Consumer側では…

アプリケーションはデータ取得のためKafka brokerをポーリングします（KafkaではPoll/Pullモデルが使われています。よく見られるPushモデルではありません）
（ブラウザやHTTPクライアントといった）クライアントはシンプルなHTTP GETリクエストを特定のURL（例：https://<acc-app-url>/metrics）に送信してイベントをサブスクライブします。アプリケーション内でイベントが発生したときにクライアントがイベントを取得するという、1回のサブスクライブ後、クライアントはいつでも切断することができます。

非同期＋疎結合
Consumerがメトリックデータを生成します。あるConsumerがブラウザベースのクライアント用のリアルタイムフィードを利用可能にしますが、同じデータに関する異なるビジネスロジックが実装されている複数のConsumer（例えば、処理や分析用途のためにメトリックデータを永続データストアにプッシュする）が存在する可能性があります。

Server Sent Events (SSE)について

SSEは、HTTPとWebSocketの中間的なものです。クライアントは接続リクエストを送信し、接続が確立すると接続は開いたままになり、サーバからデータを引き続き受信できます。

これは、サーバーのポーリングを回避できるため、単一のリクエストごとにHTTPリクエスト/レスポンスを張る方式に比べて効率的です。
全二重であるWebSocketとは異なります。具体的には、クライアントとサーバは、接続が確立された後はいつでもメッセージを交換することができるますが、SSEでは、クライアントはリクエストを1回だけ送信します。

このモデルは、クライアントが接続してデータが到着するのを待つだけなので、今回のサンプルアプリケーションに適しています（サブスクリプションした後は、サーバーと対話する必要がないためです）。

その他の着目すべきポイント

SSEはW3Cで正式な仕様です。

Server-Sent Events
W3C Working Draft 08 February 2011
https://www.w3.org/TR/2011/WD-eventsource-20110208/
データに対する具体的なメディアタイプが定められています。
ほとんどのブラウザでJavaScript実装があります。

スケーラビリティ

高いスループットとパフォーマンスを維持するために、このシステムのすべての部分がステートレスで水平方向にスケール可能であることに注意する必要があります。第2部では、スケーラビリティの側面をさらに深く理解し、Application Container Cloud Serviceで簡単にスケールさせる方法について説明します。

コード

このセクションでは、このサンプルで使った両方のマイクロサービスのコードの概要と重要なポイントを紹介します。

Producer マイクロサービス

アプリケーションのブートストラップ、イベント生成などを処理するクラスで構成されています。

Class	Details
ProducerBootstrap.java	アプリケーションのエントリポイント。 Grizzlyコンテナを起動する。
Producer.java	専用スレッドで動作。イベント生成の中核ロジックを含む。
ProducerManagerResource.java	producerプロセスの開始・停止のためのHTTP(s)エンドポイントを公開。
ProducerLifecycleManager.java	ExecutorServiceを使いProducerスレッドを管理するロジックを実装（ProducerManagerResourceが内部で利用）。

Consumer マイクロサービス

Class	Details
ConsumerBootstrap.java	アプリケーションのエントリポイント。 Grizzlyコンテナを起動してConsumerプロセスを呼び出す。
Consumer.java	専用スレッドで動作。イベント消費の中核ロジックを含む。
ConsumerEventResource.java	エンドユーザーがイベント消費するためのHTTP(s)エンドポイントを公開。
EventCoordinator.java	イベントサブスクリプションおよびブロードキャスティングを実装するためのJersey SSEBroadcasterのラッパー（ConsumerEventResourceが内部で利用）。

以下の理由で、Jersey SSE Broadcaster を使っています。

クライアントの統計情報を追跡する
クライアント接続が切断されると自動的にサーバーリソースを破棄する
スレッドセーフである

Oracle Application Container Cloudへのデプロイ

ここまででアプリケーションについて情報を得たので、ビルド、パッケージング、デプロイについて見ていきます。

まずソースコード (zipファイル) をダウンロードしてください。このエントリの最初（原文からダウンロードする場合は最後）にリンクがあります。

Metadata files

manifest.json:

このファイルはいじらずにそのまま使うことができます。

deployment.json

Kafka brokerに対応する環境変数が含まれています。値はデプロイ前にユーザーが埋めるためにプレースホルダーとして残してあります。

{
    "environment": {
        "KAFKA_CLUSTER":"<as-configured-in-kafka-server-properties>"
    }
}

この値（Oracle Compute CloudインスタンスのパブリックDNSのFQDN）は、Kafkaの server.properties ファイル中の属性 advertised.listeners で構成した値と一致する必要があります。

metadataファイルの詳細は、以下のドキュメントをご覧ください。

Oracle® Cloud Developing for Oracle Application Container Cloud Service
Creating Metadata Files
http://docs.oracle.com/en/cloud/paas/app-container-cloud/dvcjv/creating-meta-data-files.html

Build & zip

manifest.json ファイルだけを使ってJARをビルドしてZipで圧縮し、クラウドにデプロイできるアーティファクトを生成します。

Producerアプリケーション

cd <download_dir>/producer // mavenプロジェクトの展開場所
mvn clean install
zip accs-kafka-producer.zip manifest.json target/accs-kafka-producer.jar //you can also use tar to create a tgz file

Consumerアプリケーション

cd <download_dir>/consumer //mavenプロジェクトの展開場所
mvn clean install
zip accs-kafka-consumer.zip manifest.json target/accs-kafka-consumer.jar

アプリケーションのzipファイルをOracle Storage cloudへアップロード

まずOracle Storage Cloudに先ほど作成したアプリケーションのZipファイルをアップロードした上で、後続の手順でそのZipファイルを参照していきます。以下は必要なcURLコマンドです。

（まだ作成していない場合は）Oracle Storage Cloudにコンテナを作成する

curl -i -X PUT -u <USER_ID>:<USER_PASSWORD> <STORAGE_CLOUD_CONTAINER_URL>  
（例） curl -X PUT –u jdoe:foobar "https://domain007.storage.oraclecloud.com/v1/Storage-domain007/accs-kafka-consumer/"

Zipファイルをコンテナにアップロード（ZipファイルはStorage Cloudのオブジェクトにすぎません）

curl -X PUT -u <USER_ID>:<USER_PASSWORD> <STORAGE_CLOUD_CONTAINER_URL> -T <zip_file> "<storage_cloud_object_URL>" //template
（例） curl -X PUT –u jdoe:foobar -T accs-kafka-consumer.zip "https://domain007.storage.oraclecloud.com/v1/Storage-domain007/accs-kafka-consumer/accs-kafka-consumer.zip"

Producerマイクロサービスについても同じ手順を繰り返してください。

Application Container Cloudへのデプロイ

Zipファイルのアップロードが完了したら、アプリケーションをデプロイするために利用するApplication Container Cloud REST APIを使って当該ZipファイルのOracle Storage Cloudにおけるパスを参照することができます。以下はREST APIを使うcURLコマンドの例です。

curl -X POST -u joe@example.com:password \    
-H "X-ID-TENANT-NAME:domain007" \    
-H "Content-Type: multipart/form-data" -F "name=accs-kafka-consumer" \    
-F "runtime=java" -F "subscription=Monthly" \    
-F "deployment=@deployment.json" \    
-F "archiveURL=accs-kafka-consumer/accs-kafka-consumer.zip" \    
-F "notes=notes for deployment" \    
https://apaas.oraclecloud.com/paas/service/apaas/api/v1.1/apps/domain007

Producerマイクロサービスについても同じ手順を繰り返してください。

デプロイ後

Container Cloudコンソ―ルのApplicationsセクションの下にマイクロサービスを確認できるはずです。

特定のアプリケーションの詳細情報を見ると、環境変数も存在するはずです。

アプリケーションのテスト

Producer

accs-kafka-producer マイクロサービスのために、Kafka Producerプロセス（スレッド）をユーザーが開始する必要があります（これは単に柔軟性を提供するためです）。下表に記載の適切なコマンドを（cURLやPostmanなどを使って）発行し、Producerプロセスを管理します。

Action	HTTP verb	URI
開始	GET	https://<ACCS-APP-URL>/producer （例）https://accs-kafka-producer-domain007.apaas.us.oraclecloud.com/producer
停止	DELETE	上と同様

Producerを開始した後は、停止するまでイベントを発行し続けます。

Consumer

accs-kafka-consumer マイクロサービスでは、Kafka Consumerプロセスはアプリケーションとともに起動します。つまりメトリック収集のためにKafka brokerへのポーリングを開始します。前述の通り、Consumerアプリケーションは、リアルタイムにメトリックデータを見るための（Server Sent Eventsを使った）HTTP(s)エンドポイントを提供しています。

確認するには、ブラウザ（Google Chromeがよいでしょう）を使って以下のURLにアクセスします。
https://<ACCS-CONSUMER-APP-URL>/metrics
（例）
https://accs-kafka-consumer-domain007.apaas.us.oraclecloud.com/metrics
切断する場合は単純にブラウザ（タブ）を閉じてください。

下図のようなリアルタイムデータストリームを確認できるはずです。属性eventはマシン名とID、属性dataはCPU使用率を表します。

属性Partitionは無視してください。これはスケーラビリティや負荷分散といった第2部で説明予定です。

参考資料

Oracle Compute Cloud Service (IaaS)
http://docs.oracle.com/cloud/latest/stcomputecs/index.html
Oracle Application Container Cloud Service
http://docs.oracle.com/en/cloud/paas/app-container-cloud/index.html
Jersey User Guide （執筆時点の最新版は2.25）
https://jersey.java.net/documentation/latest/index.html
Apache Kafka

kafka 0.10.0.1 API
https://kafka.apache.org/0100/javadoc/index.html
Kafka 0.10.1 Documentation
https://kafka.apache.org/documentation/

第2部はまた後ほど。

[Cloud] Microservices messaging on Oracle Cloud using Apache Kafka

利用するコンポーネント

Oracle Cloud

オープンソース

Microservicesにおけるメッセージング

サンプルアプリケーションのアーキテクチャ

Apache Kafka

Producerマイクロサービス

サンプルデータ

Consumerマイクロサービス

Oracle Compute CloudへのApache Kafkaの構築

Oracle Cloud Marketplaceの利用

Oracle Compute CloudのVMを使う

VMへのSSHアクセスの有効化

VMへのKafkaのインストール

Kafkaのリスナーポートをオープン

Kafka brokerの構成

Compute CloudインスタンスのパブリックDNS

ソリューション概要

イベントフロー・シーケンス

ProducerがKafka brokerにイベントをPush

Consumer側では…

Server Sent Events (SSE)について

スケーラビリティ

コード

Producer マイクロサービス

Consumer マイクロサービス

Oracle Application Container Cloudへのデプロイ

Metadata files

manifest.json:

deployment.json

Build & zip

Producerアプリケーション

Consumerアプリケーション

アプリケーションのzipファイルをOracle Storage cloudへアップロード

Application Container Cloudへのデプロイ

デプロイ後

アプリケーションのテスト

Producer

Consumer

参考資料

0 件のコメント:

コメントを投稿