Spring Cloud Bus の紹介

[[399945]]

Spring エコシステムで RocketMQ を試すシリーズの記事:

Spring エコシステムで RocketMQ を使用するにはどうすればよいでしょうか? 》「Luo Meiqi と Spring Porter の物語...」「RocketMQ-Spring が卒業してから 2 年が経ちました。なぜ、Spring エコシステムで最も人気のあるメッセージング実装になったのでしょうか?」「rocketmq-spring-boot-starter を使用して RocketMQ メッセージを構成、送信、および消費する」「Spring Cloud Stream システムと原則の紹介」

この記事に付属するインタラクティブ チュートリアルは、Alibaba Cloud Zhixing Mobile Lab にログインされています。 PC で start.aliyun.com にログインすると、ブラウザですぐに体験できます。

Spring Cloud Bus は、Spring Cloud システム内のメッセージ バスとして位置付けられ、メッセージ ブローカーを使用して分散システム内のすべてのノードを接続します。バスの公式リファレンスドキュメントは比較的シンプルで、写真さえないほどシンプルです。

これは、Spring Cloud Bus コード構造の最新バージョンです (コードの量は比較的少ないです)。

バスの例のデモ

Bus の実装を分析する前に、Spring Cloud Bus を使用する 2 つの簡単な例を見てみましょう。

1. すべてのノードの設定を追加

バスの例は、バスの AutoConfiguration レイヤーにデフォルト構成があるため、比較的単純です。メッセージ ミドルウェアに対応する Spring Cloud Stream および Spring Cloud Bus の依存関係を導入するだけで済みます。その後、起動されたすべてのアプリケーションは同じトピックを使用してメッセージを受信および送信します。

バスに対応したデモをgithubに載せてあります。デモでは 5 つのノードの起動をシミュレートします。いずれかのインスタンスに構成項目を追加するだけで、すべてのノードに構成項目が追加されます。

デモアドレス: https://github.com/fangjian0423/rocketmq-binder-demo/tree/master/rocketmq-bus-demo

任意のノードのコントローラーによって提供される構成アドレスにアクセスします (キーは hangzhou です)。

curl -X GET 'http://localhost:10001/bus/env?key=hangzhou'

すべてのノードの構成に Hangzhou キーがないため、すべてのノードによって返される結果は不明です。

バスは、EnvironmentBusEndpoint と呼ばれるエンドポイントを提供します。これは、メッセージ ブローカーを介して構成を追加/更新するために使用されます。

任意のノードのエンドポイントに対応する URL (/actuator/bus-env?name=hangzhou&value=alibaba) にアクセスして、構成項目を追加します (たとえば、node1 の URL にアクセスします)。

curl -X POST 'http://localhost:10001/actuator/bus-env?name=hangzhou&value=alibaba' -H 'コンテンツタイプ: application/json'

次に、すべてのノード/バス/env に再度アクセスして構成を取得します。

$ curl -X GET 'http://localhost:10001/bus/env?key=hangzhou'unknown%~ $ curl -X GET 'http://localhost:10002/bus/env?key=hangzhou'unknown%~ $ curl -X GET 'http://localhost:10003/bus/env?key=hangzhou'unknown%~ $ curl -X GET 'http://localhost:10004/bus/env?key=hangzhou'unknown%~ $ curl -X GET 'http://localhost:10005/bus/env?key=hangzhou'unknown%~ $ curl -X POST 'http://localhost:10001/actuator/bus-env?name=hangzhou&value=alibaba' -H 'content-type: application/json'~ $ curl -X GET 'http://localhost:10005/bus/env?key=hangzhou'alibaba%~ $ curl -X GET 'http://localhost:10004/bus/env?key=hangzhou'alibaba%~ $ curl -X GET 'http://localhost:10003/bus/env?key=hangzhou'alibaba%~ $ curl -X GET 'http://localhost:10002/bus/env?key=hangzhou'alibaba%~ $ curl -X GET 'http://localhost:10001/bus/env?key=hangzhou'alibaba%

ご覧のとおり、すべてのノードには、キー hangzhou と対応する値 alibaba を持つ新しい構成があります。この構成項目は、Bus によって提供される EnvironmentBusEndpoint を通じて完了します。

これはプログラマー DD が描いた図で、Spring Cloud Config が Bus と連携してすべてのノード構成の更新を完了し、前の例を説明します (この記事の例は更新ではなく新しい構成ですが、プロセスは同じです)。

2. 一部ノードの構成変更

たとえば、宛先を node1 の rocketmq-bus-node2 として指定します (node2 は spring.cloud.bus.id で rocketmq-bus-node2:10002 に設定されており、これに一致します)。

curl -X POST 'http://localhost:10001/actuator/bus-env/rocketmq-bus-node2?name=hangzhou&value=xihu' -H 'コンテンツタイプ: application/json'

/bus/env にアクセスして設定を取得します (メッセージは node1 で送信されるため、Bus は送信者のノード node1 の設定も変更します)。

~ $ curl -X POST 'http://localhost:10001/actuator/bus-env/rocketmq-bus-node2?name=hangzhou&value=xihu' -H 'content-type: application/json'~ $ curl -X GET 'http://localhost:10005/bus/env?key=hangzhou'alibaba%~ $ curl -X GET 'http://localhost:10004/bus/env?key=hangzhou'alibaba%~ $ curl -X GET 'http://localhost:10003/bus/env?key=hangzhou'alibaba%~ $ curl -X GET 'http://localhost:10002/bus/env?key=hangzhou'xihu%~ $ curl -X GET 'http://localhost:10001/bus/env?key=hangzhou'xihu%

node1 と node2 のみの構成が変更され、他の 3 つのノードの構成は変更されていないことがわかります。

バス実装

1. バスコンセプトの紹介

1) イベント

リモート イベント RemoteApplicationEvent は Bus で定義されています。このイベントは Spring のイベント ApplicationEvent を継承し、現在 4 つの特定の実装があります。

EnvironmentChangeRemoteApplicationEvent: リモート環境変更イベント。これは主に、Spring コンテキストで Map<String,String> 型のデータを受信し、Environment 内のイベントを更新するために使用されます。この記事の例は、このイベントを EnvironmentBusEndpoint および EnvironmentChangeListener と組み合わせて使用​​して実装されています。 AckRemoteApplicationEvent: リモート確認イベント。バスはリモート イベントを正常に受信すると、確認のために AckRemoteApplicationEvent 確認イベントを送り返します。 RefreshRemoteApplicationEvent: リモート構成更新イベント。 @RefreshScope およびすべての @ConfigurationProperties アノテーションで変更された構成クラスの動的な更新。 UnknownRemoteApplicationEvent: リモートの不明なイベント。バスの内部メッセージ本体がリモート イベントに変換されるときに例外が発生した場合、このイベントにパッケージ化されます。

バス内には、RemoteApplicationEvent 以外のイベント、つまり Trace と連携してリモート メッセージの送信を記録する SentApplicationEvent メッセージ送信イベントもあります。

これらのイベントは ApplicationListener で操作されます。たとえば、EnvironmentChangeRemoteApplicationEvent には、構成を追加/変更するための EnvironmentChangeListener が装備されています。

パブリック クラス EnvironmentChangeListener は ApplicationListener<EnvironmentChangeRemoteApplicationEvent> を実装します { プライベート 静的 Log log = LogFactory.getLog(EnvironmentChangeListener.class); @Autowired プライベート EnvironmentManager env; @Override public void onApplicationEvent(EnvironmentChangeRemoteApplicationEvent イベント) { Map<String, String> values ​​= event.getValues(); log.info("リモート環境変更要求を受信しました。更新するキー/値" + 値); Map.Entry<String, String> エントリ: values.entrySet() の場合 { env.setProperty(entry.getKey(), entry.getValue()); } }}

他のノードから送信された EnvironmentChangeRemoteApplicationEven イベントを受信した後、EnvironmentManager#setProperty が呼び出され、構成が設定されます。このメソッドは、各構成項目に対して EnvironmentChangeEvent イベントを送信し、ConfigurationPropertiesRebinder によってリッスンされて、構成を追加/更新するための再バインド操作が実行されます。

2) アクチュエータエンドポイント

Bus は、構成の追加/変更とグローバル構成の更新のために、EnvironmentBusEndpoint と RefreshBusEndpoint の 2 つのエンドポイントを公開します。対応するエンドポイント ID または URL は bus-env と bus-refresh です。

3) 構成

バスによるメッセージの送信には、トピックやグループなどの情報を含める必要があります。これらのコンテンツは BusProperties にカプセル化されており、デフォルトの構成プレフィックスは spring.cloud.bus です。次に例を示します。

spring.cloud.bus.refresh.enabled は、グローバル リフレッシュ リスナーを有効/無効にするために使用されます。 spring.cloud.bus.env.enabled は、新しく追加/変更されたエンドポイントの構成を有効/無効にするために使用されます。 spring.cloud.bus.ack.enabled は、AckRemoteApplicationEvent イベントの送信をオン/オフにするために使用されます。 spring.cloud.bus.trace.enabled は、トレースのログ記録用のリスナーをオン/オフにするために使用されます。

メッセージ送信に使用されるデフォルトのトピックは springCloudBus ですが、構成によって変更できます。グループはブロードキャスト モードに設定するか、最新のオフセット付きの UUID を使用することができます。

各バス アプリケーションには対応するバス ID があり、公式の値の選択方法はより複雑です。

${vcap.application.name:${spring.application.name:application}}:${vcap.application.instance_index:${spring.application.index:${local.server.port:${server.port:0}}}}:${vcap.application.instance_id:${ランダム値}}

バス リモート イベントの宛先はバス ID に基づいて照合されるため、バス ID を手動で構成することをお勧めします。

spring.cloud.bus.id=${spring.application.name}-${server.port}

2. バスの基礎分析

バスの基礎分析には、次の側面が含まれます。

メッセージはどのように送信されますか?メッセージはどのように受信されますか?目的地はどのようにマッチングされますか?リモートイベントはどのように受信されますか?次のアクションはどのようにトリガーされますか?

BusAutoConfiguration 自動構成クラスは、@EnableBinding(SpringCloudBusClient.class) によって変更されます。

@EnableBinding の使用方法は、「Spring Cloud Stream システムと原則の概要」の記事で説明されており、その値は SpringCloudBusClient.class であり、SpringCloudBusClient のプロキシに基づいて入力および出力 DirectChannels を作成します。

パブリック インターフェイス SpringCloudBusClient { 文字列 INPUT = "springCloudBusInput";文字列 OUTPUT = "springCloudBusOutput"; SpringCloudBusClient のメッセージ チャネルを出力します。 SpringCloudBusClient のサブスクライブ可能なチャネルを SpringCloudBusInput() に追加します。

2 つのバインディング、springCloudBusInput と springCloudBusOutput のプロパティは、構成ファイルを通じて変更できます (たとえば、トピックを変更します)。

spring.cloud.stream.bindings: springCloudBusInput: 宛先: my-bus-topic springCloudBusOutput: 宛先: my-bus-topic

メッセージの受信と送信:

// BusAutoConfiguration@EventListener(classes = RemoteApplicationEvent.class) // 1public void acceptLocal(RemoteApplicationEvent event) { if (this.serviceMatcher.isFromSelf(event) && !(event instanceof AckRemoteApplicationEvent)) { // 2 this.cloudBusOutboundChannel.send(MessageBuilder.withPayload(event).build()); // 3 }}@StreamListener(SpringCloudBusClient.INPUT) // 4public void acceptRemote(RemoteApplicationEvent event) { if (event instanceof AckRemoteApplicationEvent) { if (this.bus.getTrace().isEnabled() && !this.serviceMatcher.isFromSelf(event) && this.applicationEventPublisher != null) { // 5 this.applicationEventPublisher.publishEvent(event); } // ACK の場合は、この時点で処理は完了です。 return; } this.serviceMatcher.isForSelf(event) の場合、 this.applicationEventPublisher != null になります。 // 6 this.serviceMatcher.isFromSelf(event) の場合、 // 7 this.applicationEventPublisher.publishEvent(event); } if (this.bus.getAck().isEnabled()) { // 8 AckRemoteApplicationEvent ack = new AckRemoteApplicationEvent(this, this.serviceMatcher.getServiceId(), this.bus.getAck().getDestinationService(), event.getDestinationService(), event.getId(), event.getClass()); this.cloudBusOutboundChannel .send(MessageBuilder.withPayload(ack).build()); this.applicationEventPublisher.publishEvent(ack); } } if (this.bus.getTrace().isEnabled() && this.applicationEventPublisher != null) { // 9 // 送信されたイベントを登録するように設定されているので、オリジンに関係なく、ローカルで消費するために公開します this.applicationEventPublisher.publishEvent(new SentApplicationEvent(this, event.getOriginService(), event.getDestinationService(), event.getId(), event.getClass())); }}

Spring イベント監視メカニズムを使用して、すべてのローカル RemoteApplicationEvent リモート イベントを監視します (たとえば、bus-env は EnvironmentChangeRemoteApplicationEvent をローカルに送信し、bus-refresh は RefreshRemoteApplicationEvent をローカルに送信します。これらのイベントはここで監視されます)。ローカルで受信したイベントが AckRemoteApplicationEvent リモート確認イベントではないかどうか (そうでない場合、メッセージの受信、メッセージの送信を繰り返す無限ループになります)、およびイベントがアプリケーション自体によって送信されたかどうか (イベントの送信者はアプリケーション自体です) を判別します。両方の条件が満たされている場合は、手順 3 に進みます。リモート イベントをペイロードとしてメッセージを作成し、Spring Cloud Stream によって作成された、バインディング名 springCloudBusOutput の MessageChannel を使用して、ブローカーにメッセージを送信します。

4.@StreamListener アノテーションは、Spring Cloud Stream によって構築された Binding 名が springCloudBusInput である MessageChannel を消費し、受信メッセージはリモート メッセージです。

リモート イベントが AckRemoteApplicationEvent リモート確認イベントであり、アプリケーションがメッセージ トレース スイッチをオンにしていて、リモート イベントがアプリケーション自体によって送信されていない場合 (イベントの送信者がアプリケーション自体ではなく、イベントが他のアプリケーションによって送信されていることを示します)、AckRemoteApplicationEvent リモート確認イベントのローカル送信は、アプリケーションが他のアプリケーションによって送信されたリモート イベントを受信したことを確認したことを示し、プロセスは終了します。リモート イベントが他のアプリケーションからアプリケーション自体に送信された場合 (イベントの受信者がアプリケーション自体である場合)、手順 7 と 8 を実行します。それ以外の場合は、手順 9 を実行します。リモート イベントがアプリケーション自体によって送信されない場合 (イベントの送信者がアプリケーション自体ではない場合)、イベントはローカル方式で送信されます。アプリケーション自体は、対応するメッセージ受信者によって最初からローカルで処理されているため、再度送信する必要はありません。 AckRemoteApplicationEvent リモート確認イベントのスイッチがオンになっている場合は、AckRemoteApplicationEvent イベントを構築し、リモートとローカルの両方に送信します (ローカルに送信するのは、ステップ 5 でローカルの AckRemoteApplicationEvent イベントが送信されないため、つまり、アプリケーションが自身のアプリケーションを確認するためです。リモートに送信するのは、アプリケーションがメッセージを受信したことを他のアプリケーションに通知するためです)。メッセージ トレース スイッチがオンになっている場合、SentApplicationEvent イベントが構築され、ローカルに送信されます。

bus-env がトリガーされると、すべてのノードの EnvironmentChangeListener が構成の変更をリッスンし、コンソールに次の情報が出力されます。

oscbevent.EnvironmentChangeListener: リモート環境変更要求を受信しました。更新するキー/値 {hangzhou=alibaba}

リモート確認イベント AckRemoteApplicationEvent をローカルでリッスンすると、すべてのノードから情報を受信します。たとえば、node5 のコンソールでリッスンされる AckRemoteApplicationEvent イベントは次のようになります。

ServiceId [rocketmq-bus-node5:10005] リスナー {"type":"AckRemoteApplicationEvent","timestamp":1554124670484,"originService":"rocketmq-bus-node5:10005","destinationService":"**","id":"375f0426-c24e-4904-bce1-5e09371fc9bc ","ackId":"750d033f-356a-4aad-8cf0-3481ace8698c","ackDestinationService":"**","event":"org.springframework.cloud.bus.event.EnvironmentChangeRemoteApplicationEvent"}ServiceId [rocketmq-bus-node5:10005] リスナー{"type":"AckRemoteApplicationEvent","timestamp":1554124670184,"originService":"rocketmq-bus-node1:10001","destinationService":"**","id":"91f06cf1-4bd9-4dd8-9526-9299a35bb7cc ","ackId":"750d033f-356a-4aad-8cf0-3481ace8698c","ackDestinationService":"**","event":"org.springframework.cloud.bus.event.EnvironmentChangeRemoteApplicationEvent"}ServiceId [rocketmq-bus-node5:10005] リスナーオン{"type":"AckRemoteApplicationEvent","timestamp":1554124670402,"originService":"rocketmq-bus-node2:10002","destinationService":"**","id":"7df3963c-7c3e-4549-9a22-a23fa90a6b85 ","ackId":"750d033f-356a-4aad-8cf0-3481ace8698c","ackDestinationService":"**","event":"org.springframework.cloud.bus.event.EnvironmentChangeRemoteApplicationEvent"}ServiceId [rocketmq-bus-node5:10005] リスナーオン{"type":"AckRemoteApplicationEvent","timestamp":1554124670406,"originService":"rocketmq-bus-node3:10003","destinationService":"**","id":"728b45ee-5e26-46c2-af1a-e8d1571e5d3a ","ackId":"750d033f-356a-4aad-8cf0-3481ace8698c","ackDestinationService":"**","event":"org.springframework.cloud.bus.event.EnvironmentChangeRemoteApplicationEvent"}ServiceId [rocketmq-bus-node5:10005] リスナーオン{"type":"AckRemoteApplicationEvent","timestamp":1554124670427,"originService":"rocketmq-bus-node4:10004","destinationService":"**","id":"1812fd6d-6f98-4e5b-a38a-4b11aee 08aeb","ackId":"750d033f-356a-4aad-8cf0-3481ace8698c","ackDestinationService":"**","event":"org.springframework.cloud.bus.event.EnvironmentChangeRemoteApplicationEvent"}

それでは、この章の冒頭で述べた 4 つの質問に戻って、1 つずつ答えてみましょう。

メッセージの送信方法: イベントは、BusAutoConfiguration#acceptLocal メソッドの Spring Cloud Stream を介して springCloudBustopic に送信されます。メッセージの受信方法: BusAutoConfiguration#acceptRemote メソッドで Spring Cloud Stream を介して springCloudBustopic メッセージを受信します。宛先の一致方法: 宛先は、BusAutoConfiguration#acceptRemote メソッドのリモート イベント受信メソッドで一致されます。リモート イベントを受信した後に次のアクションをトリガーする方法: Bus は、Spring のイベント メカニズムを通じてローカル RemoteApplicationEvent を受信し、次のアクションを実行します (たとえば、EnvironmentChangeListener は EnvironmentChangeRemoteApplicationEvent を受信し、RefreshListener は RefreshRemoteApplicationEvent を受信します)。

要約する

Spring Cloud Bus 自体の内容は比較的少ないですが、それでも Spring Cloud Stream システムと Spring 独自のイベント メカニズムを事前に理解しておく必要があります。これを基に、ローカル イベントとリモート イベントに対する Spring Cloud Bus の処理ロジックをより深く理解できるようになります。

現在、Bus に組み込まれているリモート イベントは少なく、そのほとんどは構成関連のイベントです。 RemoteApplicationEvent を継承し、@RemoteApplicationEventScan アノテーションを使用して独自のマイクロサービス メッセージ システムを構築できます。