SMQ を使用してプリエンプティブルインスタンスの割り込みイベントを検出し、対応する

ワークフロー

準備

1. AccessKey ペアを作成します。

   Resource Access Management（RAM）ユーザーの AccessKey ペアを作成します。Alibaba Cloud アカウントはリソースに対するすべての権限を持っています。Alibaba Cloud アカウントの AccessKey ペアが漏洩した場合、リソースはリスクにさらされます。RAM ユーザーの AccessKey ペアを使用することをお勧めします。AccessKey ペアの作成方法については、「AccessKey ペアを作成する」をご参照ください。

2. RAM ユーザーに権限を付与します。

   RAM ユーザーに、SMQ に関連するリソースに対する操作を実行するための権限を付与します。このトピックで提供されているサンプルコードは、SMQ からメッセージを消費します。次の表に、付与できる SMQ に対する権限を示します。

   クラウドサービス	ポリシー

   クラウドサービス	ポリシー
   SMQ（旧称 MNS）	AliyunMNSFullAccess

3. アクセス認証情報とエンドポイントを構成します。

   このトピックのサンプルコードは、システム環境変数からアクセス認証情報とエンドポイントを読み取ります。

   • 環境変数を構成します。詳細については、「Linux、macOS、および Windows で環境変数を構成する」をご参照ください。

   • エンドポイントを構成します。詳細については、「エンドポイントを構成する」をご参照ください。

4. SMQ SDK をインストールします。

   Java 用 SMQ SDK を取得します。この例では、Maven の依存関係を追加することで Java 用 SMQ SDK をインストールします。その他のインストール方法については、「Install SMQ SDK for Java」をご参照ください。

   次のサンプルコードは、Maven の依存関係を追加する方法の例を示しています。

   <dependencies>
       <! -- Alibaba Cloud SMQ SDK --> 

手順

1. SMQ を作成します。

   CloudMonitor が送信するプリエンプティブルインスタンスの割り込み通知を受信するための SMQ を作成します。

   1. SMQ コンソールにログインします。左側のナビゲーションウィンドウで、キューモデル > キュー を選択します。

   2. 上部のナビゲーションバーで、リージョンを選択します。キュー ページで、キューを作成 をクリックします。

   3. キューを作成 パネルで、画面の指示に従って必須パラメーターを構成し、OK をクリックします。

      

2. サブスクリプションポリシーを作成します。

   CloudMonitor は、プリエンプティブルインスタンスの割り込みイベントをリアルタイムで監視します。イベントアラートが発生した場合、CloudMonitor はサブスクリプションポリシーで指定されたプッシュチャネルを介して割り込み通知をプッシュします。

   1. CloudMonitor コンソールにログインします。 左側のナビゲーションウィンドウで、[イベントサブスクリプション] を選択します。

   2. [サブスクリプションポリシー] タブで、[サブスクリプションポリシーの作成] をクリックし、必須パラメーターを構成します。

      この例では、プリエンプティブルインスタンスの割り込みイベントをサブスクライブするために使用される主要なパラメーターのみを説明します。ビジネス要件に基づいてパラメーターを指定できます。詳細については、「イベントサブスクリプションを管理する」をご参照ください。

      • [サブスクリプションタイプ]：[システムイベント] を選択します。

        

      • [サブスクリプションスコープ]：次の図に示すようにパラメーターを指定します。

        

      • [プッシュと統合]：[チャネルを追加] をクリックします。表示されるダイアログボックスで、[チャネルを増やす] をクリックします。表示されるダイアログボックスで、手順 1 で作成した SMQ を選択し、画面の指示に従って他のパラメーターを構成します。プッシュチャネルについては、「プッシュチャネルを管理する」をご参照ください。

3. 割り込みイベントをシミュレートします。

   プリエンプティブルインスタンスの割り込みイベントはトリガーイベントです。プリエンプティブルインスタンスの割り込みイベントハンドラを開発する場合、コードをデバッグすることはできません。[デバッグイベントサブスクリプション] を使用して、プリエンプティブルインスタンスの割り込みイベントをシミュレートできます。

   1. [サブスクリプションポリシー] タブで、[デバッグイベントサブスクリプション] をクリックします。

   2. [イベントデバッグの作成] パネルで、[プロダクト] を [elastic Compute Service (ECS)] に、[名前] を [instance:preemptibleinstanceinterruption] に設定します。

      システムは JSON 形式でデバッグコンテンツを自動的に生成します。JSON ファイルのリソース情報を、割り込みイベントをシミュレートするプリエンプティブルインスタンスの情報に置き換えます。

      • Alibaba Cloud account ID を Alibaba Cloud アカウント ID に置き換えます。

      • <resource-id> と <instanceId> をプリエンプティブルインスタンスの ID に置き換えます。

      • <region ID> をプリエンプティブルインスタンスのリージョン ID に置き換えます。

        {
            "product": "ECS",
            "resourceId": "acs:ecs:cn-shanghai:<Alibaba Cloud account ID>UID:instance/<resource-id>", //"resourceId": "acs:ecs:cn-shanghai:<Alibaba Cloud アカウント ID>UID:instance/<resource-id>",
            "level": "WARN",
            "instanceName": "instanceName",
            "regionId": "<Region ID>", //"regionId": "<リージョン ID>",
            "groupId": "0",
            "name": "Instance:PreemptibleInstanceInterruption",
            "content": {
                "instanceId": "<instanceId>", //"instanceId": "<instanceId>",
                "instanceName": "wor***b73",
                "action": "delete" //"action": "削除"
            },
            "status": "Normal" //"status": "正常"
        }

   3. [OK] をクリックします。操作が成功したことを示すメッセージが表示されます。CloudMonitor は SMQ にアラート通知を自動的に送信します。

4. メッセージをプルして対応します。

   割り込みイベントハンドラをシミュレートして、プリエンプティブルインスタンスの割り込み通知メッセージを SMQ からプルします。ビジネス要件に基づいて、対応するビジネス処理ロジックを追加することもできます。次のサンプルコードは、グレースケール画像変換ハンドラを使用して割り込みイベントに対応および処理する方法の例を示しています。

   1. スレッドタスクを使用して、グレースケール画像変換ハンドラをシミュレートします。

      import com.aliyun.mns.client.CloudAccount;
      import com.aliyun.mns.client.CloudQueue;
      import com.aliyun.mns.client.MNSClient;
      import com.aliyun.mns.common.utils.ServiceSettings;
      import com.aliyun.mns.model.Message;
      import org.json.JSONObject;
      
      import javax.imageio.ImageIO;
      import java.awt.image.BufferedImage;
      import java.io.File;
      import java.util.Base64;
      import java.util.concurrent.atomic.AtomicBoolean;
      
      /**
       * グレースケール変換をサポートする割り込み可能な画像プロセッサの実装クラス。
       * アトミック変数とスレッド割り込みフラグに基づく割り込み検出メカニズム。
       * 特徴：
       * 1. チャンクごとに処理を実行して、進捗状況を自動的に保存します。
       * 2. プリエンプティブルインスタンスの割り込みイベントに即座に対応します。
       * 3. プリエンプティブルインスタンスの割り込みイベント発生後に部分的な結果を含むファイルを生成します。
       */
      public class InterruptibleImageProcessor implements Runnable {
          /**
            * スレッドセーフな状態制御にアトミックブール値を使用します。
           */
          private final AtomicBoolean running = new AtomicBoolean(true);
          /**
            * 処理中の画像データを格納します。
           */
          private BufferedImage processedImage;
          /**
           * 処理の進捗状況（パーセント）。有効な値：0～100。
           */
          private int progress;
          /**
           * スレッド実行エントリ。
           * **割り込み処理ロジック**：
           * 1. システムが割り込み例外をキャプチャした後、現在の進捗状況を保存します。
           * 2. 割り込みセマンティクスを保持するために、スレッドを割り込みが発生した状態に戻します。
           */
          @Override
          public void run() {
              try {
                  convertToGrayScale(new File("input.jpg"), new File("output.jpg"));
                  // プリエンプティブルインスタンスの実行中に割り込みイベントをシミュレートします。
                  Thread.sleep(5000); 
                  System.out.println("画像処理が完了しました");
              } catch (InterruptedException e) {
                  System.out.println("処理が中断され、進捗状況は " + progress + "% に保存されました");
                  saveProgress(new File("partial_output.jpg"));
                  Thread.currentThread().interrupt(); // スレッドが割り込まれた状態に戻します。
              } catch (Exception e) {
                  System.err.println("エラー処理: " + e.getMessage());
              }
          }
      
          /**
           * 外部割り込みをトリガーするメソッド
           * **コラボレーションメカニズム**：
           * スレッド割り込みフラグに基づいて割り込みを二重チェックします。
           */
          public void stop() {
              running.set(false);
          }
      }
      

   2. グレースケール画像変換メソッド。

      /**
       * 入力した画像をグレースケール画像に変換して保存します。
       * @param inputFile 元の画像ファイルのオブジェクト。
       * @param outputFile 出力ファイルオブジェクト。
       * @throws Exception I/O 異常と割り込み例外が含まれます。
       *
       * **アルゴリズムの説明**：
       * 次の数式で赤、緑、青の光の知覚される輝度に基づいて設計された加重平均値と係数を使用して、画像をグレースケールに変換します。
       * Gray = 0.30*R + 0.59*G + 0.11*B
       * 参考文献：ITU-R 勧告 BT. 601。
       */
      public void convertToGrayScale(File inputFile, File outputFile) throws Exception {
          // ソース画像データを読み取ります。
          BufferedImage original = ImageIO.read(inputFile);
          int width = original.getWidth();
          int height = original.getHeight();
          // グレースケール画像バッファを作成します。
          processedImage = new BufferedImage(width, height, BufferedImage.TYPE_BYTE_GRAY);
      
          // チャンクごとに処理を実行して、進捗状況を保存します。
          for (int y = 0; y < height && running.get(); y++) {
              // 画像のピクセル単位の処理を実行します。
              for (int x = 0; x < width; x++) {
                  // 最初の割り込み検出で、スレッド割り込みフラグを確認します。
                  if (Thread.interrupted()) {
                      throw new InterruptedException("画像処理が中断されました");
                  }
                  /* グレースケール画像変換のコアアルゴリズム * /  /*グレースケール画像変換のコアアルゴリズム*/
                  int rgb = original.getRGB(x, y);
                  // ARGB 形式で RGB チャネルを分解します。
                  // 赤チャネル。
                  int r = (rgb >> 16) & 0xFF;
                  // 緑チャネル。
                  int g = (rgb >> 8) & 0xFF;
                  // 青チャネル。
                  int b = rgb & 0xFF;
                  // 加重平均法を使用してグレースケール値を計算します。
                  int gray = (int)(0.3 * r + 0.59 * g + 0.11 * b);
                  // グレースケール値を赤、緑、青のチャネルにコピーして RGB 値を再構築します。
                  processedImage.setRGB(x, y, (gray << 16) | (gray << 8) | gray);
                  // 進捗状況をパーセントで更新します。小数値ではなく整数値が返される整数除算の問題に注意してください。
                  progress = (y * width + x) * 100 / (width * height);
              }
              // チェックポイントメカニズムに基づいて、50 行ごとに処理した後、進捗状況を自動的に保存します。
              if (y % 50 == 0) {
                  saveProgress(outputFile);
              }
          }
          // 完全な結果を保存します。
          ImageIO.write(processedImage, "jpg", outputFile);
      }

   3. 画像処理の進捗状況が保存されます。

      /**
       * 処理の進捗状況を指定されたファイルに保存します。
       * @param outputFile 出力ファイルオブジェクト。
       *
       * **注**：
       * 1. 保存処理の中断を防ぐために、フェイルサイレントメカニズムが実装されています。
       * 2. partial_output.jpg という名前の一時ファイルが生成されます。
       */
      private void saveProgress(File outputFile) {
          try {
              // 最終的なファイルが上書きされるのを防ぐために、一時ファイル名を使用します。
              ImageIO.write(processedImage, "jpg", new File("partial_output.jpg"));
          } catch (Exception e) {
              System.err.println("自動保存に失敗しました: " + e.getMessage());
          }
      }

   4. 応答処理をテストします。

      テスト画像処理プログラムの実行中に、プリエンプティブルインスタンスが中断されて再利用されようとしているイベントを受信し、そのイベントに対応します。

      /**
       * メインテストメソッド
       * **テストシナリオ**：
       * 1. スレッドを処理します。
       * 2. 割り込みイベント通知メッセージをプルします。
       * 3. スレッドが終了するのを待ちます。
       */
      public static void main(String[] args) throws InterruptedException {
          // MNSClient アカウントを初期化します。
          CloudAccount account = new CloudAccount(
                  ServiceSettings.getMNSAccessKeyId(),
                  ServiceSettings.getMNSAccessKeySecret(),
                  ServiceSettings.getMNSAccountEndpoint());
          MNSClient client = account.getMNSClient();
          // プリエンプティブルインスタンスで割り込みイベントが発生するかどうかを確認します。
          boolean isMatch = false;
          // 画像処理プログラムを開始します。
          InterruptibleImageProcessor processor = new InterruptibleImageProcessor();
          Thread processThread = new Thread(processor);
          processThread.start();
          try{
              // SMQ からメッセージを取得します。
              CloudQueue queue = client.getQueueRef("spot-interruption");
              Message popMsg = queue.popMessage();
              if (popMsg != null){
                  // デフォルトでは、メッセージ本文は Base64 でエンコードされています。
                  System.out.println("message body: " + popMsg.getMessageBodyAsRawString());
                  // Base64 デコードを実行します。
                  byte[] decodedBytes = Base64.getDecoder().decode(popMsg.getMessageBodyAsRawString());
                  String decodedString = new String(decodedBytes);
                  System.out.println("message content: " + decodedString);
                  // JSON 文字列を解析します。
                  JSONObject json = new JSONObject(decodedString);
                  // イベント名フィールドの値を取得します。
                  String name = json.getString("name");
                  isMatch = "Instance:PreemptibleInstanceInterruption".equals(name);
                  // プリエンプティブルインスタンスの割り込みイベントに対応します。
                  if(isMatch){
                      System.out.println("プリエンプティブルインスタンスが中断されて再利用されようとしています。");
                      // 画像処理プログラムを終了します。
                      processor.stop();
                      processThread.interrupt();
                      System.out.println("プログラムが終了しました");
                      processThread.join();
                      // メッセージを削除します。
                      queue.deleteMessage(popMsg.getReceiptHandle());
                  }
              }
          }catch (Exception e){
              System.out.println("不明な例外が発生しました！");
              e.printStackTrace();
          }
          client.close();
      }

   サンプルコード:

   /*  サンプルコードは上記と同じため省略 */