Alibaba Cloud Model Studio:HTTP コールバックまたは MQ を介したタスク完了通知の受信

通常、非同期イベント通知を受信するために、業務システムに HTTP コールバックエンドポイントをデプロイします。以下の要件を満たす必要があります：

• リクエスト URL：パブリックネットワークまたは Alibaba Cloud VPC 経由でアクセス可能な HTTP URL。
• リクエストメソッド：POST。
• リクエストボディ：非同期タスク完了イベントデータを含む JSON フォーマット。正確なイベント構造は EventBridge コンソールで確認できます。例：

{
    "datacontenttype": "application/json;charset=utf-8",
    "aliyunaccountid": "xxxxx",
    "aliyunpublishtime": "2023-10-25T01:45:16.993Z",
    "data": {
        "start_time": "2023-10-25 09:45:09",
        "user_api_unique_key": "apikey:v1:audio:asr:transcription:paraformer-8k-v1",
        "task_status": "SUCCEEDED",
        "contain_result": false,
        "end_time": "2023-10-25 09:45:16",
        "task_id": "a154c328-xxxx-xxxx-xxxx-e52a9a7e9a35",
        "region": "ap-southeast-1",
        "request_id": "108f38f5-xxxx-xxxx-xxxx-6504db9080b3",
        "api_key_id": "1250"
    },
    "aliyunoriginalaccountid": "xxxxxxxx",
    "specversion": "1.0",
    "aliyuneventbusname": "default",
    "id": "81765e5b-xxxx-xxxx-xxxx-bbad8dde2bd9",
    "source": "acs.dashscope",
    "time": "2023-1-25T01:45:16.969Z",
    "aliyunregionid": "ap-southeast-1",
    "type": "dashscope:System:AsyncTaskFinish"
}

EventBridge コンソールでは、Alibaba Cloud Model Studio によって配信されたイベントをクエリできます。

1. Alibaba Cloud アカウントにログインし、EventBridge コンソールに移動します。 [シンガポールリージョン]に切り替えます。 左側のナビゲーションウィンドウで、[イベントバス]を選択し、[デフォルト]をクリックしてクラウドサービスのデフォルトイベントバスに入ります。

   Alibaba Cloud Model Studio は、デフォルトでデフォルトイベントバスを使用します。

2. [イベント追跡] をクリックし、クエリ条件を入力して、Alibaba Cloud Model Studio からの非同期タスク完了イベントを検索します。

   • イベントソース：acs.dashscope を検索して選択します。これは、イベントが DashScope (Alibaba Cloud Model Studio の基盤となるモデルサービス) から発信されたことを示します。
   • イベントタイプ：dashscope:System:AsyncTaskFinish を検索して選択します。これは、非同期タスク完了イベントを示します。

3. [詳細] をクリックすると、報告されたイベントの詳細が表示されます。

{
    "datacontenttype": "application/json;charset=utf-8",
    "aliyunaccountid": "xxxxx",
    "aliyunpublishtime": "2023-10-25T01:45:16.993Z",
    "data": {
        "start_time": "2023-10-25 09:45:09",
        "user_api_unique_key": "apikey:v1:audio:asr:transcription:paraformer-8k-v1",
        "task_status": "SUCCEEDED",
        "contain_result": false,
        "end_time": "2023-10-25 09:45:16",
        "task_id": "a154c328-xxxx-xxxx-xxxx-e52a9a7e9a35",
        "region": "ap-southeast-1",
        "request_id": "108f38f5-xxxx-xxxx-xxxx-6504db9080b3",
        "api_key_id": "1250"
    },
    "aliyunoriginalaccountid": "xxxxxxxx",
    "specversion": "1.0",
    "aliyuneventbusname": "default",
    "id": "81765e5b-xxxx-xxxx-xxxx-bbad8dde2bd9",
    "source": "acs.dashscope",
    "time": "2023-1-25T01:45:16.969Z",
    "aliyunregionid": "ap-southeast-1",
    "type": "dashscope:System:AsyncTaskFinish"
}

1. 左側のナビゲーションウィンドウで、[イベントルール] を選択し、[ルールを作成] をクリックします。

2. 基本情報の設定: カスタムルール名と説明を入力します。

3. イベントパターン設定：転送するイベントを指定します。

   • イベントソース: acs.dashscope を検索して選択します。これは、イベントが Alibaba Cloud Model Studio から発生したことを示します。

   • イベントタイプ: 非同期タスク完了イベントを示す dashscope:System:AsyncTaskFinish を検索して選択します。

   • パターン内容: 特定のイベントに一致するようにフィルター条件を設定します。 ステップ 2 で照会したイベント詳細のフィールドを使用します。 パターン構文については、「イベントパターン」をご参照ください。 例：

     • デフォルト：イベントソースとタイプを選択すると、パターンはデフォルトで以下のように設定され、すべての dashscope:System:AsyncTaskFinish イベントが転送されます。

{
  "source": ["acs.dashscope"],
  "type": ["dashscope:System:AsyncTaskFinish"]
}

• フィールドによるフィルタリング：user_api_unique_key フィールドが :paraformer-8k-v1 で終わるイベントのみを転送します。つまり、paraformer-8k-v1 モデルのイベントのみです。イベントタイプは dashscope:System:AsyncTaskFinish です。

{
  "source": ["acs.dashscope"],
  "type": ["dashscope:System:AsyncTaskFinish"],
  "data": {
    "user_api_unique_key": [
      {"suffix": ":paraformer-8k-v1"}
    ]
  }
}

4. イベントターゲットの設定：HTTP コールバック URL や RocketMQ メッセージキューなど、複数のイベントターゲットタイプをサポートしています。詳細については、ステップ 4 をご参照ください。

1. イベントターゲットの設定: イベントを HTTP コールバック URL に転送します。

   • サービスタイプ：「HTTP」を選択します。

   • URL：HTTP サービスエンドポイントを入力します。

   • 本文：「完全なイベント」を選択します。

   • ネットワークタイプ：サービスエンドポイントに基づいて選択します。

     • HTTP はパブリックネットワークとプライベートネットワークの両方をサポートしています。プライベートネットワークを選択した場合は、VPC、vSwitch、および SecurityGroup を設定します。

2. OK をクリックしてルールを保存します。イベントターゲットが HTTP 形式で表示された場合、構成は正常に完了しました。

   ルールの [イベントターゲット] 列に [HTTP (1)] と表示されます。

RocketMQ 経由でイベントを受信するには、まず RocketMQ キューを準備します。

1. RocketMQ コンソールに移動します。左側のナビゲーションウィンドウで、[インスタンス] を選択し、[インスタンスの作成] をクリックします。

   インスタンス ID の例：rmq-cn-nwy*******。

2. インスタンスの Topic を作成し、カスタムのトピック名を入力します。

3. インスタンスの Group を作成し、カスタムのグループ名を入力します。

EventBridge コンソールでは、Alibaba Cloud Model Studio によって配信されたイベントをクエリできます。

1. Alibaba Cloud アカウントにログインしてEventBridge コンソールに移動し、[シンガポールリージョン]に切り替えます。左側のナビゲーションウィンドウで、[イベントバス] を選択し、[default] をクリックして、クラウドサービスのデフォルトイベントバスに移動します。

   Alibaba Cloud Model Studio は、デフォルトでデフォルトイベントバスを使用します。

2. [イベント追跡] をクリックし、クエリ条件を入力して、Alibaba Cloud Model Studio からの非同期タスク完了イベントを検索します。

   • イベントソース：acs.dashscope を検索して選択します。これは、イベントが DashScope (Alibaba Cloud Model Studio の基盤となるモデルサービス) から発信されたことを示します。
   • イベントタイプ：dashscope:System:AsyncTaskFinish を検索して選択します。これは、非同期タスク完了イベントを示します。

3. [詳細] をクリックすると、報告されたイベントの詳細を表示できます。

{
    "datacontenttype": "application/json;charset=utf-8",
    "aliyunaccountid": "xxxxx",
    "aliyunpublishtime": "2023-10-25T01:45:16.993Z",
    "data": {
        "start_time": "2023-10-25 09:45:09",
        "user_api_unique_key": "apikey:v1:audio:asr:transcription:paraformer-8k-v1",
        "task_status": "SUCCEEDED",
        "contain_result": false,
        "end_time": "2023-10-25 09:45:16",
        "task_id": "a154c328-xxxx-xxxx-xxxx-e52a9a7e9a35",
        "region": "ap-southeast-1",
        "request_id": "108f38f5-xxxx-xxxx-xxxx-6504db9080b3",
        "api_key_id": "1250"
    },
    "aliyunoriginalaccountid": "xxxxxxxx",
    "specversion": "1.0",
    "aliyuneventbusname": "default",
    "id": "81765e5b-xxxx-xxxx-xxxx-bbad8dde2bd9",
    "source": "acs.dashscope",
    "time": "2023-1-25T01:45:16.969Z",
    "aliyunregionid": "ap-southeast-1",
    "type": "dashscope:System:AsyncTaskFinish"
}

1. 左側のナビゲーションウィンドウで、[イベントルール] を選択し、[ルールを作成] をクリックします。

2. 基本情報の設定: カスタムルール名と説明を入力します。

3. イベントパターンの設定：転送するイベントを指定します。

   • イベントソース: イベントが Alibaba Cloud Model Studio から発生したことを示す acs.dashscope を検索して選択します。

   • イベントタイプ: 非同期タスクの完了イベントを示す dashscope:System:AsyncTaskFinish を検索して選択します。

   • パターンの内容：特定のイベントに一致するようにフィルター条件を設定します。ステップ 2 で取得したイベント詳細のフィールドを使用します。パターン構文については、「イベントパターン」をご参照ください。例：

     • デフォルト：イベントソースとタイプを選択すると、パターンはデフォルトで以下のように設定され、すべての dashscope:System:AsyncTaskFinish イベントが転送されます。

{
  "source": ["acs.dashscope"],
  "type": ["dashscope:System:AsyncTaskFinish"]
}

• フィールドによるフィルタリング：user_api_unique_key フィールドが :paraformer-8k-v1 で終わるイベントのみを転送します。つまり、paraformer-8k-v1 モデルのイベントのみです。イベントタイプは dashscope:System:AsyncTaskFinish です。

{
  "source": ["acs.dashscope"],
  "type": ["dashscope:System:AsyncTaskFinish"],
  "data": {
    "user_api_unique_key": [
      {"suffix": ":paraformer-8k-v1"}
    ]
  }
}

4. イベントターゲットの設定: HTTP コールバック URL や RocketMQ メッセージキューなど、複数のイベントターゲットタイプをサポートしています。 詳細については、ステップ 4 をご参照ください。

RocketMQ を作成した後、イベントターゲット設定インターフェイスを開き、RocketMQ インスタンスを選択します。

• サービスタイプ：「Message Queue for RocketMQ」を選択します。
• バージョン：作成した RocketMQ のバージョン (例：RocketMQ 5.x)。
• インスタンス ID：RocketMQ インスタンスのインスタンス ID。詳細はステップ 1をご参照ください。
• トピック：作成したトピック名。詳細はステップ 1をご参照ください。

設定後、非同期タスクを送信します。完了後、設定した RocketMQ トピックでメッセージを表示できます。

• RocketMQ でオンラインでメッセージを表示するには、ワンクリック送受信体験機能を有効にします。
• この機能は Function Compute を使用します。無料試用クォータを超えた場合、最小限の料金が発生します。詳細は、「Function Compute の課金ルール」をご参照ください。

[Topics] ページで、対象のトピックを見つけ、[アクション] 列の [詳細] をクリックします。

トピック詳細ページで、[ワンクリック送受信体験] タブに切り替えます。[基本機能] セクションで、[PushConsumer] などの消費方法を選択します。

[トピック名] や [グループ ID] などのパラメーターを設定した後、[実行] をクリックしてメッセージの受信を開始します。[実行結果] タブには、メッセージ ID や受信時刻など、受信したメッセージが表示されます。

RocketMQ Java SDK を使用して、関連するトピックをサブスクライブし、メッセージリスナーを実装するロジックを実装します。メッセージ受信時にそれを処理します。

以下の例は、RocketMQ 5.0 の Java クライアントコードを示しています。

• Maven プロジェクトに、この依存関係を追加します：

<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-client-java</artifactId>
  <version>5.0.4</version>
</dependency>

• MQ メッセージを消費するためのサンプルコード：

import com.alibaba.fastjson2.JSON;
import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collections;

public class ConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerExample.class);

    private ConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        /*
          コンソールのインスタンス詳細ページの [Endpoints] タブから取得したエンドポイント。
          Alibaba Cloud ECS インスタンスから内部ネットワーク経由でアクセスする場合は、VPC エンドポイントを使用します。
          ローカルのパブリックネットワークまたはオンプレミスの IDC からアクセスする場合は、パブリックエンドポイントを使用します。パブリックエンドポイントアクセスには、インスタンスのパブリックネットワークアクセスを有効にする必要があります。
         */
        String endpoints = "xxxx";
        // サブスクライブするターゲットトピックを指定します。トピックは事前にコンソールで作成しておく必要があります。そうしないとエラーが発生します。
        String topic = "xxxx";
        // コンシューマーのコンシューマーグループを指定します。グループは事前にコンソールで作成しておく必要があります。そうしないとエラーが発生します。
        String consumerGroup = "xxxx";
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        /*
          パブリックエンドポイントを使用する場合は、インスタンスのユーザー名とパスワードを設定します。これらはコンソールのインスタンス詳細ページで確認できます。
          Alibaba Cloud ECS インスタンスから内部ネットワーク経由でアクセスする場合、この設定はスキップします。サーバーが VPC 情報をインテリジェントに取得します。
         */
        builder.setCredentialProvider(new StaticSessionCredentialsProvider("xxxx", "xxxx"));
        ClientConfiguration clientConfiguration = builder.build();
        // サブスクリプションフィルターのルール—任意のタグを持つメッセージをサブスクライブします。
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // PushConsumer をコンシューマーグループ、通信パラメーター、およびサブスクリプション関係で初期化します。
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
        .setClientConfiguration(clientConfiguration)
        // コンシューマーグループを設定します。
        .setConsumerGroup(consumerGroup)
        // 事前にバインドされたサブスクリプション関係を設定します。
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        // メッセージリスナーを設定します。
        .setMessageListener(messageView -> {
            try {
                // メッセージを処理し、消費結果を返します。
                ByteBuffer buffer = messageView.getBody();
                ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity());
                for (int i = 0; i < buffer.capacity(); i++) {
                    newBuffer.put(buffer.get(i));
                }
                String result = new String(newBuffer.array());
                LOGGER.info("Consume message={}", JSON.toJSONString(result));
                System.out.println(result);
                return ConsumeResult.SUCCESS;
            } catch (Exception e) {
                LOGGER.error("deal message has error", e);
                return ConsumeResult.FAILURE;
            }
        })
        .build();
        Thread.sleep(Long.MAX_VALUE);

        // PushConsumer が不要になったら閉じます。
        pushConsumer.close();
    }
}