フィードバックを送信
Message Bus
ご覧のページは、お客様の利便性のために一部機械翻訳されています。また、ドキュメントは頻繁に更新が加えられており、翻訳は未完成の部分が含まれることをご了承ください。最新情報は都度公開されておりますため、必ず英語版をご参照ください。翻訳に問題がある場合は、 こちら までご連絡ください。

以前のバージョンでデフォルトの同期メッセージングを使用する

Liferay DXP 7.4 U48/Portal GA49 以前 サポート対象外

重要

このチュートリアルで説明するメッセージングAPIは、Liferay DXP 7.4 U48/Portal GA49以降では利用できません。

デフォルトの同期メッセージングでは、メッセージバススレッドが登録済みのメッセージリスナーにメッセージを配信する間、送信側はブロックされます。 送信側は、応答メッセージを受信するか、送信スレッドがタイムアウトした時点でブロックを解除します。

注記

送信者は、最初に受信した応答メッセージを受け取ると、ブロックを解除します。

サンプルプロジェクトを使用して、デフォルトの同期メッセージを送信します。 次に、メッセージにタイムアウトが発生するように例を修正します。

デフォルトの同期メッセージを送信する

サンプルプロジェクトでは、デフォルトモードで SynchronousMessageSender を使用してメッセージを送信し、応答を待ちます。

サンプルプロジェクトでは、 SynchronousMessageSender を使用して、メッセージを 2 つのリスナーに直接送信します。

Liferay の新しいインスタンスを起動するには、次のコマンドを実行します。

docker run -it -m 8g -p 8080:8080 liferay/portal:7.4.3.48-ga48

http://localhost:8080 で Liferay にサインインしてください。 メールアドレス test@liferay.com とパスワード test を使用してください。 プロンプトが表示されたら、パスワードを learn に変更してください。

次に、以下の手順に従ってください。

  1. サンプルをダウンロードして解凍してください。

    curl https://resources.learn.liferay.com/examples/liferay-m4q7.zip -O

    unzip liferay-m4q7.zip

  2. サンプルプロジェクトのモジュールをビルドしてデプロイします。

    cd liferay-m4q7

    ./gradlew deploy -Ddeploy.docker.container.id=$(docker ps -lq)
    注記

    このコマンドは、モジュール JAR を Docker コンテナ上の /opt/liferay/osgi/modules にコピーすることと同じです。

  3. Dockerコンテナのコンソールにモジュールの起動状況が表示されます。

    com.acme.m4q7.able.impl_1.0.0 が起動しました
com.acme.m4q7.baker.impl_1.0.0 が起動しました
com.acme.m4q7.charlie.impl_1.0.0 が起動しました

  4. ブラウザで http://localhost:8080 にアクセスして Liferay インスタンスを開き、認証情報を使用してサインインしてください。

  5. Gogoシェル を開きます。

  6. Gogoシェルコマンドフィールドに、 m4q7:sendMessage と入力し、続けてメッセージを入力します。 例えば、

    m4q7:sendMessage foo

  7. 出力が以下のようになっていることを確認してください。

    INFO [acme/m4q7_able-2][M4Q7CharlieMessageListener:23] メッセージペイロード foo を受信しました
INFO [acme/m4q7_baker-2][M4Q7BakerMessageListener:21] メッセージペイロード M4Q7CharlieMessageListener を受信しました
INFO [pipe-m4q7:sendMessage foo][M4Q7BakerOSGiCommands:28] レスポンス: M4Q7CharlieMessageListener

acme/m4q7_able 宛先で、 M4Q7CharlieMessageListener が Gogo シェル メッセージを受信しました。 acme/m4q7_baker 宛先で、 M4Q7BakerMessageListener は M4Q7CharlieMessageListener からの応答メッセージを受信しました。 最後に、 M4Q7BakerOSGiCommandsの sendMessage メソッドは、メッセージ送信者から返された応答オブジェクトをログに記録しました。

プロジェクト概要

これら3つのサンプルモジュールクラスは、宛先の管理、メッセージの受信、およびメッセージの送信を行います。

m4q7-able-impl モジュール: M4Q7AbleMessagingConfigurator は acme/m4q7_able という名前のメッセージ宛先を作成し、それをメッセージバスに登録します。

m4q7-baker-impl モジュール:

  • M4Q7BakerOSGiCommandsは、acme/m4q7_able宛先にメッセージを送信し、応答をログに記録します。
  • M4Q7BakerMessagingConfiguratorは、acme/m4q7_bakerという名前のメッセージ宛先を作成し、それをメッセージバスに登録します。
  • M4Q7BakerMessageListenerは、acme/m4q7_baker宛先に送信されたメッセージをリッスンし、メッセージペイロードをログに記録します。

m4q7-charlie-implモジュール： M4Q7CharlieMessageListener は、acme/m4q7_able宛てに送信されたメッセージをリッスンし、メッセージのペイロードをログに記録し、元のメッセージの応答先宛てに応答メッセージを送信します。

イベントの流れは以下のとおりです。

  1. m4q7:sendMessage Gogoシェルコマンドを呼び出し、メッセージを渡します。
  2. M4Q7BakerOSGiCommandsの sendMessage(String) メソッドは Gogo シェル コマンドでトリガーされ、コマンド引数をメッセージとして acme/m4q7_able 宛先に送信します。
  3. メッセージバススレッドはメッセージを M4Q7CharlieMessageListener に配信します。
  4. M4Q7CharlieMessageListener メッセージペイロードをログに記録し、応答メッセージで自身のクラス名を元のメッセージの応答先に送信します acme/m4q7_baker。
  5. M4Q7BakerMessageListener は応答メッセージを受信し、そのペイロードをログに記録します。
  6. 処理は M4Q7BakerOSGiCommandsに戻り、そこで元のメッセージに対する応答がログに記録されます。

それでは、各クラスを詳しく見ていきましょう。まずは、宛先コンフィギュレータから始めます。

目的地コンフィギュレーターを確認する

m4q7-able-impl モジュールと m4q7-baker-impl モジュールは、それぞれ宛先コンフィギュレータクラス M4Q7AbleMessagingConfigurator と M4Q7BakerMessagingConfiguratorを持っています。 それぞれが宛先を作成および構成します。

M4Q7AbleMessagingConfigurator クラスは、 acme/m4q7_able 宛先を設定します。

@Component
public class M4Q7AbleMessagingConfigurator {

	@Activate
	private void _activate(BundleContext bundleContext) {
		DestinationConfiguration destinationConfiguration =
			DestinationConfiguration.createSerialDestinationConfiguration(
				"acme/m4q7_able");

		Destination destination = _destinationFactory.createDestination(
			destinationConfiguration);

		_serviceRegistration = bundleContext.registerService(
			Destination.class, destination,
			MapUtil.singletonDictionary(
				"destination.name", destination.getName()));
	}

	@Deactivate
	private void _deactivate() {
		if (_serviceRegistration != null) {
			_serviceRegistration.unregister();
		}
	}

	@Reference
	private DestinationFactory _destinationFactory;

	private ServiceRegistration<Destination> _serviceRegistration;

}

M4Q7BakerMessagingConfigurator クラスは、 acme/m4q7_baker 宛先を設定します。

@Component
public class M4Q7BakerMessagingConfigurator {

	@Activate
	private void _activate(BundleContext bundleContext) {
		DestinationConfiguration destinationConfiguration =
			DestinationConfiguration.createSerialDestinationConfiguration(
				"acme/m4q7_baker");

		Destination destination = _destinationFactory.createDestination(
			destinationConfiguration);

		_serviceRegistration = bundleContext.registerService(
			Destination.class, destination,
			MapUtil.singletonDictionary(
				"destination.name", destination.getName()));
	}

	@Deactivate
	private void _deactivate() {
		if (_serviceRegistration != null) {
			_serviceRegistration.unregister();
		}
	}

	@Reference
	private DestinationFactory _destinationFactory;

	private ServiceRegistration<Destination> _serviceRegistration;

}

どちらのコンフィギュレータも コンポーネント クラスです。 @Reference アノテーションを使用して、 DestinationFactory インスタンスを注入します。

_activate(BundleContext) メソッドは、 DestinationFactory と DestinationConfiguration を使用して、 シリアル 宛先を作成します。 最後に、 _activate(BundleContext) メソッドは、 Destination を BundleContext を使用して OSGi サービスに登録します。

警告

デフォルトの同期メッセージングでは、シリアルまたはパラレル宛先のみを使用してください。 DestinationConfigurationの createSerialDestinationConfiguration(String) および createParallelDestinationConfiguration(String) メソッドを呼び出すことで作成できます。

デフォルトの同期メッセージングでは同期宛先を使用しないでください。同期宛先を使用すると、メッセージ送信者のタイムアウトが無効になります。

コンフィギュレータが非アクティブ化されると、その _deactivate() メソッドは宛先サービスを登録解除します。

聞き手を調査する

m4q7-charlie-impl モジュールの M4Q7CharlieMessageListener クラスは、 acme/m4q7_able Destination に送信されるメッセージをリッスンします。 「メッセージの受信」で示されているのと同じ方法で登録されます。

M4Q7CharlieMessageListener クラス:

@Component(
	property = "destination.name=acme/m4q7_able",
	service = MessageListener.class
)
public class M4Q7CharlieMessageListener implements MessageListener {

	@Override
	public void receive(Message message) {
		if (_log.isInfoEnabled()) {
			_log.info("メッセージペイロードを受信しました " + message.getPayload());
		}

		_messageBus.sendMessage(
			message.getResponseDestinationName(),
			new Message() {
				{
					setPayload("M4Q7CharlieMessageListener");
					setResponseId(message.getResponseId());
				}
			});
	}

	private static final Log _log = LogFactoryUtil.getLog(
		M4Q7CharlieMessageListener.class);

	@Reference
	private MessageBus _messageBus;

}

M4Q7CharlieMessageListener がメッセージを受信すると、その receive(Message) メソッドはメッセージペイロードをログに記録し、元のメッセージの応答先に応答メッセージを送信します。 このメソッドは、応答メッセージのペイロードをリスナークラス名に設定し、応答メッセージIDを元のメッセージの応答IDに設定します。

重要

デフォルトの同期メッセージングでは、応答メッセージは元のメッセージの応答 ID を使用し、 を応答宛先に送信する必要があります。

m4q7-baker-impl モジュールの M4Q7BakerMessageListener クラスは、 acme/m4q7_bakerに送信されるメッセージをリッスンします。これは、 M4Q7BakerOSGiCommandsのメッセージの応答先です。

M4Q7BakerMessageListener クラス:

@Component(
	property = "destination.name=acme/m4q7_baker",
	service = MessageListener.class
)
public class M4Q7BakerMessageListener implements MessageListener {

	@Override
	public void receive(Message message) {
		if (_log.isInfoEnabled()) {
			_log.info("受信したメッセージペイロード " + message.getPayload());
		}
	}

	private static final Log _log = LogFactoryUtil.getLog(
		M4Q7BakerMessageListener.class);

}

M4Q7BakerMessageListener がメッセージを受信すると、その receive(Message) メソッドがメッセージペイロードをログに記録します。

送信者を調べる

m4q7-baker-impl モジュールの M4Q7BakerOSGiCommands クラスは、コマンド引数をメッセージとして "acme/m4q7_able" 宛先に送信する OSGi コマンドを提供します。

@Component(
	property = {"osgi.command.function=sendMessage", "osgi.command.scope=m4q7"},
	service = M4Q7BakerOSGiCommands.class
)
public class M4Q7BakerOSGiCommands {

	public void sendMessage(String payload) throws MessageBusException {
		Message message = new Message();

		message.setPayload(payload);
		message.setResponseDestinationName("acme/m4q7_baker");

		Object response = _synchronousMessageSender.send(
			"acme/m4q7_able", message, 10000);

		if (_log.isInfoEnabled()) {
			_log.info("レスポンス: " + response);
		}
	}

	private static final Log _log = LogFactoryUtil.getLog(
		M4Q7BakerOSGiCommands.class);

	@Reference(target = "(mode=DEFAULT)")
	private SynchronousMessageSender _synchronousMessageSender;

}

M4Q7BakerOSGiCommands は、独自のクラスタイプのサービス コンポーネント です。 これは、 @Reference アノテーションを使用して、 SynchronousMessageSender を挿入します。このモードは、 デフォルト モードに設定されています (アノテーションの target = "(mode=DEFAULT)" 属性によって指定されます)。

注記

デフォルト モードでは、 SynchronousMessageSenderの send メソッドは、応答メッセージが受信されるか、送信者がタイムアウトするまで呼び出し元のクラスをブロックします。

M4Q7BakerOSGiCommandsの @Component プロパティは、 m4q7 スコープで sendMessage という Gogo シェル コマンド関数を定義します。 このコマンドは入力文字列 を受け取り、 M4Q7BakerOSGiCommandsの sendMessage(String) メソッドにマッピングされます。

sendMessage(String) メソッドは、ペイロードとして Gogo シェル コマンドの String を、レスポンスとして "acme/m4q7_baker" を持つ メッセージ を作成します。 行き先。

sendMessage(String) メソッドは、 SynchronousMessageSenderの send(String, Message, long) メソッドを呼び出し、 "acme/m4q7_able" 宛先名、メッセージ インスタンス、および を渡してメッセージを送信します。 10000 ミリ秒のタイムアウト。 デフォルトモードでは、 SynchronousMessageSender はメッセージバススレッドを使用してメッセージをメッセージリスナーに配信します。 M4Q7BakerOSGiCommands クラスでの実行は、 "acme/m4q7_baker" 応答先で元のメッセージの応答 ID を持つメッセージが受信されるまでブロックされます。 応答が受信されると、 M4Q7BakerOSGiCommands sendMessage(String) メソッドで実行が続行され、そこでメッセージ応答がログに記録されます。 タイムアウトが一致する応答メッセージを受信する前に期限切れになった場合、 SynchronousMessageSenderの send(String, Message, long) メソッドは MessageBusException をスローします。

重要

デフォルトの同期メッセージングでは、応答メッセージは元のメッセージの応答 ID を使用し、 を応答宛先に送信する必要があります。

メッセージリスナーが応答メッセージを返すのを確認できたので、応答タイムアウトをテストできます。

応答タイムアウトを実演する

タイムアウトを強制するために、メッセージ応答ロジックを無効にする方法は次のとおりです。

  1. M4Q7CharlieMessageListenerの receive(Message) メソッドで、 _messageBus.sendMessage(...) の呼び出しをコメントアウトします。

    @Override
public void receive(Message message) {
    if (_log.isInfoEnabled()) {
        Object payload = message.getPayload();

        _log.info("受信したメッセージペイロード " + payload.toString());
    }

    // _messageBus.sendMessage(
    // message.getResponseDestinationName(),
    // new Message() {
    // {
    // setPayload("M4Q7CharlieMessageListener");
    // setResponseId(message.getResponseId());
    // }
    // });
}

  2. サンプルプロジェクトを再デプロイしてください。

    ./gradlew deploy -Ddeploy.docker.container.id=$(docker ps -lq)

  3. Gogoシェルコマンドフィールドに、 m4q7:sendMessage と入力し、続けてメッセージを入力します。 例えば、

    m4q7:sendMessage foo

  4. Gogoシェルページが以下のようになっていることを確認してください。

    エラー：メッセージに対する返信がありませんでした。

  5. Dockerコンソールに表示されるメッセージが以下のようになっていることを確認してください。

    INFO [acme/m4q7_able-2][M4Q7CharlieMessageListener:23] メッセージペイロード foo を受信しました

M4Q7CharlieMessageListener はメッセージを受信しましたが、返信しませんでした。 SynchronousMessageSender は MessageBusException をスローし、それが Gogo シェル ページに表示されました。

タイムアウトを設定してメッセージを同期的に送信できたことをお祝い申し上げます。

次は？

ダイレクト モードを使用して同期メッセージングを検証したい場合は、 以前のバージョンでのダイレクト同期メッセージングの使用 を参照してください。

メッセージ送信後すぐに処理を続行したい場合は、 非同期メッセージングの使用 を参照してください。

Capability:
Development and Tooling
Resource Type:
Official Documentation
Feature:
Core Frameworks
Deployment Approach:
Liferay PaaS
Liferay Self-Hosted