Python 用 SDK - Platform For AI - Alibaba Cloud ドキュメントセンター

呼び出しロジックの定義に必要な時間を短縮し、呼び出しの安定性を向上させるために、公式の Elastic Algorithm Service (EAS) SDK を使用することをお勧めします。このトピックでは、EAS SDK for Python について説明し、一般的に使用される入力と出力、および EAS SDK for Python を使用してサービスを呼び出す方法を示すデモを提供します。

SDK のインストール

pip install -U eas-prediction --user

メソッド

共通パラメーターの説明

エンドポイント: サーバーのエンドポイント。

標準モードでサービスを呼び出す場合は、このパラメーターをデフォルトゲートウェイのエンドポイントに設定します。例: 182848887922***.cn-shanghai.pai-eas.aliyuncs.com。
Virtual Private Cloud (VPC) 直接接続経由でサービスを呼び出す場合は、このパラメーターをリージョンの共通エンドポイントに設定します。たとえば、中国 (上海) リージョンでは、このパラメーターを pai-eas-vpc.cn-shanghai.aliyuncs.com に設定します。

PredictClient

メソッド	説明
`PredictClient(endpoint, service_name, custom_url)`	説明: PredictClient クラスのクライアントオブジェクトを作成します。パラメーター: エンドポイント: サーバーのエンドポイント。詳細については、「共通パラメーターの説明」をご参照ください。サービス名: サービス名。 custom_url: サービスの URL。このパラメーターはオプションであり、`<uid>.<region>.pai-eas.aliyuncs.com` 形式ではないエンドポイントを持つサービス (Web UI サービスなど) にのみ必要です。このパラメーターを設定して、クライアントを作成できます。例: `client = PredictClient(custom_url='<url>')`。
`set_endpoint(endpoint)`	説明: サーバーのエンドポイントを指定します。詳細については、「共通パラメーターの説明」をご参照ください。パラメーター: エンドポイント: サーバーのエンドポイント。
`set_service_name(service_name)`	説明: サービス名を指定します。パラメーター: サービス名: サービス名。
`set_endpoint_type(endpoint_type)`	説明: サーバーのゲートウェイの種類を指定します。パラメーター: エンドポイントタイプ: 使用するゲートウェイの種類。次のゲートウェイタイプがサポートされています。 ENDPOINT_TYPE_GATEWAY: デフォルトゲートウェイ。 ENDPOINT_TYPE_DIRECT: VPC 直接接続チャネル。このパラメーターを設定しない場合、デフォルトゲートウェイを使用してサービスにアクセスします。
`set_token(token)`	説明: サービスへのアクセスに使用するトークンを指定します。パラメーター: トークン: サービスアクセスのトークン。
`set_retry_count(max_retry_count)`	説明: リクエストの失敗後に許可される最大再試行回数を設定します。パラメーター: max_retry_count: リクエストの失敗後に許可される最大再試行回数。デフォルト値: 5。重要サーバーでプロセスエラーが発生した場合、サーバーエラーが発生した場合、またはゲートウェイへの持続的接続が閉じられた場合、クライアントはリクエストを再送信する必要があります。したがって、このパラメーターを 0 に設定しないことをお勧めします。
`set_max_connection_count(max_connection_count)`	説明: クライアントの接続プールで許可される持続的接続の最大数を指定します。パフォーマンスを向上させるために、クライアントはサーバーへの持続的接続を確立し、持続的接続を接続プールに保存します。リクエストを開始するたびに、クライアントは接続プール内のアイドル接続を使用して、必要なサービスにアクセスします。パラメーター: max_connection_count: 接続プールで許可される持続的接続の最大数。デフォルト値: 100。
`set_timeout(timeout)`	説明: リクエストのタイムアウト期間を指定します。パラメーター: タイムアウト: リクエストのタイムアウト期間。デフォルト値: 5000。単位: ミリ秒。
`init()`	説明: クライアントオブジェクトを初期化します。上記のすべてのメソッドが呼び出された後、`Init()` メソッドを呼び出して、パラメーターを有効にする必要があります。
`predict(request)`	説明: オンライン予測サービスに予測リクエストを送信します。パラメーター: リクエスト: 抽象クラス。StringRequest クラスや TFRequest クラスなど、さまざまなクラスのリクエストにすることができます。戻り値: 予測リクエストへの応答。

StringRequest

メソッド	説明
`StringRequest(request_data)`	説明: StringRequest クラスのオブジェクトを作成します。パラメーター: request_data: 送信するリクエスト文字列。

StringResponse

メソッド	説明
`to_string()`	説明: StringResponse クラスのレスポンスを文字列に変換します。戻り値: リクエストのレスポンス本文。

TFRequest

メソッド	説明
`TFRequest(signature_name)`	説明: TFRequest クラスのオブジェクトを作成します。パラメーター: signature_name: 呼び出すサービスのモデルのシグネチャ名。
`add_feed(self, input_name, shape, data_type, content)`	説明: 呼び出す TensorFlow オンライン予測サービスの入力テンソルを指定します。パラメーター: input_name: 入力テンソルのエイリアス。 shape: 入力テンソルの形状。 data_type: 入力テンソルのデータ型。次のデータ型がサポートされています。 TFRequest.DT_FLOAT TFRequest.DT_DOUBLE TFRequest.DT_INT8 TFRequest.DT_INT16 TFRequest.DT_INT32 TFRequest.DT_INT64 TFRequest.DT_STRING TFRequest.TF_BOOL content: 入力テンソルの内容。内容は 1 次元配列の形式です。
`add_fetch(self, output_name)`	説明: TensorFlow オンライン予測サービスの呼び出し時に返す出力テンソルのエイリアスを指定します。パラメーター: output_name: 返す出力テンソルのエイリアス。 TensorFlow モデルが SavedModel 形式の場合、このパラメーターはオプションです。このパラメーターを指定しない場合、すべての出力テンソルが返されます。 TensorFlow モデルがフリーズモデルの場合、このパラメーターは必須です。
`to_string()`	説明: Protocol Buffer (PB) オブジェクトを文字列にシリアル化します。PB オブジェクトは TFRequest クラスを使用して作成され、リクエストの送信に使用されます。戻り値: TFRequest ベースのシリアル化から取得された文字列。

TFResponse

メソッド	説明
`get_tensor_shape(output_name)`	説明: 指定されたエイリアスで識別される出力テンソルの形状をクエリします。パラメーター: output_name: 形状をクエリする出力テンソルのエイリアス。戻り値: 出力テンソルの形状。
`get_values(output_name)`	説明: 指定された出力テンソルのデータベクトルをクエリします。パラメーター: output_name: データをクエリする出力テンソルのエイリアス。戻り値: 1 次元配列。このメソッドを `get_tensor_shape()` メソッドと一緒に呼び出して、出力テンソルの形状をクエリし、1 次元配列を必要な多次元配列に復元できます。出力テンソルのデータ型によって、返される 1 次元配列のデータ型が決まります。

TorchRequest

メソッド	説明
`TorchRequest()`	説明: TorchRequest クラスのオブジェクトを作成します。
`add_feed(self, index, shape, data_type, content)`	説明: 呼び出す PyTorch オンライン予測サービスの入力テンソルを指定します。パラメーター: index: 入力テンソルのインデックス。 shape: 入力テンソルの形状。 data_type: 入力テンソルのデータ型。次のデータ型がサポートされています。 TFRequest.DT_FLOAT TFRequest.DT_DOUBLE TFRequest.DT_INT8 TFRequest.DT_INT16 TFRequest.DT_INT32 TFRequest.DT_INT64 TFRequest.DT_STRING TFRequest.TF_BOOL content: 入力テンソルの内容。内容は 1 次元配列の形式です。
`add_fetch(self, output_index)`	説明: PyTorch オンライン予測サービスの呼び出し時に返す出力テンソルのインデックスを指定します。このメソッドはオプションです。このメソッドを呼び出して出力テンソルのインデックスを指定しない場合、すべての出力テンソルが返されます。パラメーター: output_index: 返す出力テンソルのインデックス。
`to_string()`	説明: PB オブジェクトを文字列にシリアル化します。PB オブジェクトは TorchRequest クラスを使用して作成され、リクエストの送信に使用されます。戻り値: TorchRequest ベースのシリアル化から取得された文字列。

TorchResponse

メソッド	説明
`get_tensor_shape(output_index)`	説明: 指定されたインデックスで識別される出力テンソルの形状をクエリします。パラメーター: output_index: 形状をクエリする出力テンソルのインデックス。戻り値: 指定されたインデックスで識別される出力テンソルの形状。
`get_values(output_index)`	説明: 指定された出力テンソルのデータベクトルをクエリします。戻り値は 1 次元配列です。このメソッドを `get_tensor_shape()` メソッドと一緒に呼び出して、出力テンソルの形状をクエリし、1 次元配列を必要な多次元配列に復元できます。出力テンソルのデータ型によって、返される 1 次元配列のデータ型が決まります。パラメーター: output_index: クエリする出力テンソルのインデックス。戻り値: 1 次元配列。

QueueClient

メソッド	説明
`QueueClient(endpoint, queue_name)`	説明: QueueClient クラスのクライアントオブジェクトを作成します。パラメーター: エンドポイント: サーバーのエンドポイント。 queue_name: 作成するキューの名前。戻り値: 作成されたクライアントオブジェクト。
`set_token(token)`	説明: 認証のために QueueClient クラスのクライアントオブジェクトのトークンを作成します。パラメーター: トークン: 作成するキューのトークン。
`init(uid=None,gid='eas')`	説明: QueueClient クラスのオブジェクトを初期化します。パラメーター: uid: クライアントのユーザー ID。クライアントはサーバーへの登録に使用されます。各クライアントインスタンスのユーザー ID は一意である必要があり、各ユーザー ID は 1 回だけ登録できます。サーバーによってプッシュされたデータは、ユーザー ID 間で均等に分散されます。 gid: クライアントのグループ ID。クライアントはサーバーへの登録に使用されます。デフォルトでは、同じグループ ID を持つクライアントは同じグループに属します。異なるグループが存在する場合、同じデータレコードのコピーがすべてのグループにプッシュされます。
`set_logger(logger=None)`	説明: キューのロガーを設定します。デフォルトでは、ロガーは警告情報を通常の出力として表示します。ロギングを無効にするには、logger を None に設定します。パラメーター: logger: 設定するロガー。
`truncate(index)`	説明: 特定のインデックスより前のキュー内のデータを切り捨て、インデックスより後のデータのみを保持します。パラメーター: index: キュー内のデータの切り捨てに使用するインデックス。
`put(data,tags:dict={})`	説明: データレコードをキューに書き込みます。パラメーター: data: キューに書き込むデータレコード。 tags: オプション。キューに書き込むデータレコードのタグ。戻り値: index: キューに書き込まれたデータレコードのインデックス。この値を使用して、キュー内のデータをクエリできます。 requestId: キューに書き込まれたデータレコードに対して自動的に生成されるリクエスト ID。reuqestId を特別なタグとして使用して、キュー内のデータをクエリできます。
`get(request_id=None, index=0, length=1, timeout='5s', auto_delete=True, tags={})`	説明: 指定された条件に基づいてキュー内のデータをクエリします。パラメーター: request_id: クエリするデータレコードのリクエスト ID。このパラメーターを指定すると、システムは index から始まる length で指定された数のデータレコードをクエリします。クエリ結果にリクエスト ID に一致するレコードが含まれている場合、そのレコードが返されます。それ以外の場合、null が返されます。 index: クエリを開始するインデックス。デフォルト値: 0。最初のデータレコードからクエリが開始されることを示します。 length: クエリするデータレコードの数。このパラメーターを指定すると、システムは index から始まる最大 length のデータレコードを返します。index で指定されたデータレコードも返されます。 timeout: クエリのタイムアウト期間。タイムアウト期間中に、キューに length データレコードが含まれている場合、データレコードはすぐに返されます。それ以外の場合、指定されたタイムアウト期間が経過するとクエリは停止します。 auto_delete: 取得したデータレコードをキューから自動的に削除するかどうかを指定します。auto_delete を False に設定すると、データレコードが繰り返しクエリされます。この場合、`Del()` メソッドを使用してデータを手動で削除できます。 tags: データレコードのクエリに使用するタグ。タグのデータ型は dict です。このパラメーターを指定すると、システムは指定された index から始まる length データレコードをクエリし、指定されたタグを持つデータレコードを返します。戻り値: DataFrame 形式で取得されたデータレコード。
`attributes()`	説明: キューの属性をクエリします。属性には、キューの全長とキュー内の現在のデータ長が含まれます。戻り値: attrs: キューの属性。属性のデータ型は dict です。
`delete(indexes)`	説明: インデックス値で指定されたデータレコードをキューから削除します。パラメーター: indexes: データレコードの削除に使用する指定されたインデックス値。単一のインデックス値を文字列として指定するか、複数のインデックス値をリストとして指定できます。
`search(index)`	説明: データレコードのキュー情報をクエリします。パラメーター: インデックス: データレコードのインデックス。戻り値: データレコードのキュー情報。キュー情報は JSONObject 型で、次のフィールドが含まれています。 ConsumerId: データレコードを処理するインスタンスの ID です。処理中: データレコードが処理中かどうかを示します。有効な値: True: データレコードは処理中です。 False: データレコードはキューで処理されるのを待機しています。 WaitCount: キュー内の現在のデータレコードの前に処理待ちのデータレコード数を示します。このパラメーターは、IsPending が False に設定されている場合にのみ有効です。 IsPending が True に設定されている場合、このパラメーターの値は 0 です。サンプルレスポンス: `{'ConsumerId': 'eas.****', 'IsPending': False, 'WaitCount':2}` が返された場合、データレコードはキューで処理されるのを待機しています。返されたログに `search error:Code 404, Message: b'no data in stream'` と `{}` が表示された場合、データレコードはキューに見つかりません。これは、データレコードがサーバーによって既に処理され、結果がクライアントに返されているか、index パラメーターが正しく設定されていないことが原因である可能性があります。
`watch(index, window, index_only=False, auto_commit=False)`	説明: キュー内のデータレコードをサブスクライブします。次に、キューイングサービスは指定された条件に基づいてデータをクライアントにプッシュします。パラメーター: index: データレコードのサブスクリプションを開始するインデックス。 window: キューイングサービスによって単一のクライアントにプッシュされることが許可されるデータレコードの最大数。説明プッシュされたデータレコードがコミットされていない場合、サーバーは他のデータレコードをクライアントにプッシュしません。N 個のデータレコードがコミットされた後、サーバーは N 個のデータレコードをクライアントにプッシュして、クライアントによって処理されるデータレコードの数が window で指定された値を超えないようにします。これにより、データ処理におけるクライアント側の同時実行性が制御されます。 index_only: インデックス値のみをプッシュするかどうかを指定します。 auto_commit: データレコードがプッシュされた後に自動的にコミットするかどうかを指定します。auto_commit を False に設定することをお勧めします。この場合、プッシュされたデータレコードを受信して計算した後、手動でコミットする必要があります。計算が完了する前にインスタンスで例外が発生した場合、インスタンス上のコミットされていないデータレコードは、キューイングサービスによって他のインスタンスにプッシュされます。戻り値: プッシュされたデータを読み取るために使用されるウォッチャー。
`commit(index)`	説明: index で指定されたデータレコードをコミットします。説明データレコードが処理され、他のインスタンスにプッシュする必要がなくなった場合は、コミットされます。その後、データレコードはキューから削除できます。パラメーター: index: コミットするデータレコードを指定するために使用するインデックス値。単一のインデックス値を文字列として指定するか、複数のインデックス値をリストとして指定できます。

Watcher

メソッド	説明
`run()`	説明: ウォッチャーを実行してサーバーへの WebSocket 接続を確立し、サーバーからプッシュされたデータを受信し、リアルタイムで呼び出し元にデータを返します。戻り値: DataFrame 形式で呼び出し元にプッシュされたリアルタイムデータ。
`close()`	説明: ウォッチャーを閉じてバックエンド接続を終了します。説明単一のクライアントに対して開始できるウォッチャーは 1 つだけです。別のウォッチャーを開始する前に、ウォッチャーを閉じる必要があります。

デモ

文字列としての入力と出力

カスタムプロセッサを使用してサービスをデプロイする場合、Predictive Model Markup Language (PMML) モデルに基づいてデプロイされたサービスなど、文字列を使用してサービスを呼び出すことがよくあります。デモコード:

#!/usr/bin/env python

# EAS 予測用ライブラリのインポート
from eas_prediction import PredictClient
from eas_prediction import StringRequest

if __name__ == '__main__':
    # エンドポイントとサービス名を使用してクライアントを作成します
    client = PredictClient('http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'scorecard_pmml_example')
    # サービスアクセストークンを設定します
    client.set_token('YWFlMDYyZDNmNTc3M2I3MzMwYmY0MmYwM2Y2MTYxMTY4NzBkNzdj****')
    # クライアントを初期化します
    client.init()

    # リクエスト文字列を作成します
    request = StringRequest('[{"fea1": 1, "fea2": 2}]')
    # サービスを 1000000 回呼び出します
    for x in range(0, 1000000):
        resp = client.predict(request)
        print(resp)

テンソルとしての入力と出力

TensorFlow を使用してサービスをデプロイする場合、TFRequest クラスと TFResponse クラスを使用してサービスを呼び出す必要があります。デモコード:

#!/usr/bin/env python

# EAS 予測用ライブラリのインポート
from eas_prediction import PredictClient
from eas_prediction import StringRequest
from eas_prediction import TFRequest

if __name__ == '__main__':
    # エンドポイントとサービス名を使用してクライアントを作成します
    client = PredictClient('http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'mnist_saved_model_example')
    # サービスアクセストークンを設定します
    client.set_token('YTg2ZjE0ZjM4ZmE3OTc0NzYxZDMyNmYzMTJjZTQ1YmU0N2FjMTAy****')
    # クライアントを初期化します
    client.init()

    #request = StringRequest('[{}]')
    # 予測メソッド名と入力テンソルを設定します
    req = TFRequest('predict_images')
    req.add_feed('images', [1, 784], TFRequest.DT_FLOAT, [1] * 784)
    # サービスを 1000000 回呼び出します
    for x in range(0, 1000000):
        resp = client.predict(req)
        print(resp)

VPC 直接接続チャネルを使用してサービスを呼び出す

VPC 直接接続チャネルを使用して、EAS 専用リソースグループにデプロイされたサービスのみにアクセスできます。VPC 直接接続チャネルを使用するには、EAS 専用リソースグループと指定された vSwitch が同じ VPC 内に存在する必要があります。EAS 専用リソースグループの購入方法とネットワーク接続の構成方法の詳細については、「専用リソースグループを操作する」および「ネットワーク接続を構成する」をご参照ください。通常モードと比較して、このモードには追加のコード行 client.set_endpoint_type(ENDPOINT_TYPE_DIRECT) が含まれています。このモードは、高並列性と高トラフィックのシナリオで使用できます。デモコード：

#!/usr/bin/env python

# EAS 予測用ライブラリのインポート
from eas_prediction import PredictClient
from eas_prediction import StringRequest
from eas_prediction import TFRequest
from eas_prediction import ENDPOINT_TYPE_DIRECT

if __name__ == '__main__':
    # エンドポイントとサービス名を使用してクライアントを作成します。VPC アクセスを使用する場合は、リージョンの共通エンドポイントを使用します
    client = PredictClient('http://pai-eas-vpc.cn-hangzhou.aliyuncs.com', 'mnist_saved_model_example')
    # サービスアクセストークンを設定します
    client.set_token('M2FhNjJlZDBmMzBmMzE4NjFiNzZhMmUxY2IxZjkyMDczNzAzYjFi****')
    # VPC 直接接続チャネルを使用するようにクライアントを設定します
    client.set_endpoint_type(ENDPOINT_TYPE_DIRECT)
    # クライアントを初期化します
    client.init()

    # 予測メソッド名と入力テンソルを設定します
    request = TFRequest('predict_images')
    request.add_feed('images', [1, 784], TFRequest.DT_FLOAT, [1] * 784)
    # サービスを 1000000 回呼び出します
    for x in range(0, 1000000):
        resp = client.predict(request)
        print(resp)

PyTorch モデルの呼び出し

PyTorch を使用してサービスをデプロイする場合、TorchRequest クラスと TorchResponse クラスを使用してサービスを呼び出す必要があります。デモコード:

#!/usr/bin/env python

# EAS 予測用ライブラリのインポート
from eas_prediction import PredictClient
from eas_prediction import TorchRequest

if __name__ == '__main__':
    # エンドポイントとサービス名を使用してクライアントを作成します
    client = PredictClient('http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'pytorch_gpu_wl')
    # クライアントを初期化します
    client.init()


    req = TorchRequest()
    # 入力テンソルを設定します
    req.add_feed(0, [1, 3, 224, 224], TorchRequest.DT_FLOAT, [1] * 150528)
    # req.add_fetch(0)
    import time
    st = time.time()
    timer = 0
    # サービスを 10 回呼び出します
    for x in range(0, 10):
        resp = client.predict(req)
        timer += (time.time() - st)
        st = time.time()
        print(resp.get_tensor_shape(0))
        # print(resp)
    # 平均応答時間を計算します
    print("平均応答時間: %s 秒" % (timer / 10) )

Blade プロセッサベースのモデルの呼び出し

Blade プロセッサを使用してサービスをデプロイする場合、BladeRequest クラスと BladeResponse クラスを使用してサービスを呼び出す必要があります。デモコード:

#!/usr/bin/env python

# EAS 予測用ライブラリのインポート
from eas_prediction import PredictClient
from eas_prediction import BladeRequest 

if __name__ == '__main__':
    # エンドポイントとサービス名を使用してクライアントを作成します
    client = PredictClient('http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'nlp_model_example')
    # クライアントを初期化します
    client.init()

    # 入力テンソルを設定します
    req = BladeRequest()

    req.add_feed('input_data', 1, [1, 360, 128], BladeRequest.DT_FLOAT, [0.8] * 85680)
    req.add_feed('input_length', 1, [1], BladeRequest.DT_INT32, [187])
    req.add_feed('start_token', 1, [1], BladeRequest.DT_INT32, [104])
    req.add_fetch('output', BladeRequest.DT_FLOAT)
    import time
    st = time.time()
    timer = 0
    # サービスを 10 回呼び出します
    for x in range(0, 10):
        resp = client.predict(req)
        timer += (time.time() - st)
        st = time.time()
        # print(resp)
        # print(resp.get_values('output'))
        print(resp.get_tensor_shape('output'))
    # 平均応答時間を計算します
    print("平均応答時間: %s 秒" % (timer / 10) )

デフォルトの TensorFlow メソッドと互換性のある EAS Blade プロセッサベースのモデルの呼び出し

TFRequest クラスと TFResponse クラスを使用して、EAS でサポートされているデフォルトの TensorFlow メソッドと互換性のある Blade プロセッサベースのモデルを呼び出すことができます。デモコード:

#!/usr/bin/env python

# EAS 予測用ライブラリのインポート
from eas_prediction import PredictClient
# Blade TFRequest のインポートが必要
from eas_prediction.blade_tf_request import TFRequest 

if __name__ == '__main__':
    # エンドポイントとサービス名を使用してクライアントを作成します
    client = PredictClient('http://182848887922****.cn-shanghai.pai-eas.aliyuncs.com', 'nlp_model_example')
    # クライアントを初期化します
    client.init()

    # 入力テンソルを設定します
    req = TFRequest(signature_name='predict_words')

    req.add_feed('input_data', [1, 360, 128], TFRequest.DT_FLOAT, [0.8] * 85680)
    req.add_feed('input_length', [1], TFRequest.DT_INT32, [187])
    req.add_feed('start_token', [1], TFRequest.DT_INT32, [104])
    req.add_fetch('output')
    import time
    st = time.time()
    timer = 0
    # サービスを 10 回呼び出します
    for x in range(0, 10):
        resp = client.predict(req)
        timer += (time.time() - st)
        st = time.time()
        # print(resp)
        # print(resp.get_values('output'))
        print(resp.get_tensor_shape('output'))
    # 平均応答時間を計算します
    print("平均応答時間: %s 秒" % (timer / 10) )

キューイングサービスを使用してデータを送信およびサブスクライブする

キュー内のデータの送受信、キューの状態のクエリ、およびキューによってプッシュされたデータのサブスクライブを行うことができます。次のデモでは、1 つのスレッドがキューにデータをプッシュし、別のスレッドがウォッチャーを使用してプッシュされたデータをサブスクライブします。デモコード:

#!/usr/bin/env python

# EAS 予測用ライブラリのインポート
from eas_prediction import QueueClient
import threading

if __name__ == '__main__':
    endpoint = '182848887922****.cn-shanghai.pai-eas.aliyuncs.com'
    queue_name = 'test_group.qservice/sink'
    token = 'YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MTUx****'

    queue = QueueClient(endpoint, queue_name)
    queue.set_token(token)
    queue.init()
    queue.set_timeout(30000)

    # キュー内のすべてのメッセージを切り詰めます
    attributes = queue.attributes()
    if 'stream.lastEntry' in attributes:
        queue.truncate(int(attributes['stream.lastEntry']) + 1)

    count = 100
    # キューにメッセージを送信するスレッドを作成します
    def send_thread():
        for i in range(count):
            index, request_id = queue.put('[{}]')
            print('送信: ', i, index, request_id)

    # キューからのメッセージを監視するスレッドを作成します
    def watch_thread():
        watcher = queue.watch(0, 5, auto_commit=True)
        i = 0
        for x in watcher.run():
            print('受信: ', i, x.index, x.tags['requestId'])
            i += 1
            if i == count:
                break
        watcher.close()

    thread1 = threading.Thread(target=watch_thread)
    thread2 = threading.Thread(target=send_thread)

    thread1.start()
    thread2.start()

    thread1.join()
    thread2.join()