Elastic Compute Service:Simple Message Queue を使用したスポットインスタンス中断の処理

ワークフローの概要

前提条件

1. AccessKey の作成

   ルートアカウントの代わりに、最小限の権限を持つ RAM ユーザーの AccessKey を使用します。 詳細については、「AccessKey の作成」をご参照ください。

2. RAM ユーザーへの権限付与

   SMQ からメッセージを消費するために、RAM ユーザーに次の権限を付与します。

   クラウド製品	権限ポリシー
   SMQ	AliyunMNSFullAccess

3. アクセス認証情報とエンドポイントの設定

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

   • アクセス認証情報：詳細については、「Linux、macOS、および Windows で環境変数を設定する」をご参照ください。

   • エンドポイント：詳細については、「エンドポイントの設定」をご参照ください。

4. SMQ SDK のインストール

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

   次の Maven 依存関係を追加します。

   <dependencies>
       <!-- Alibaba Cloud Simple Message Queue の SDK -->
       <dependency>
         <groupId>com.aliyun.mns</groupId>
         <artifactId>aliyun-sdk-mns</artifactId>
         <version>1.2.0</version>
       </dependency>
   </dependencies>

操作手順

1. SMQ キューの作成

   スポットインスタンスの中断通知を受信するためのキューを作成します。

   1. SMQ コンソールにログインします。 左側メニューで、キューモデル > キュー を選択します。

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

   3. キューを作成 パネルでパラメーターを設定し、OK をクリックします。

      

2. サブスクリプションポリシーの作成

   Cloud Monitor は、サブスクリプションポリシーで指定されたプッシュチャネルを通じてスポットインスタンスの中断通知を送信します。

   1. Cloud Monitor コンソールにログインします。 左側メニューで、[イベントハブ] > [イベントサブスクリプション] を選択します。

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

      ここでは主要なパラメーターのみを説明します。 その他のパラメーターについては、「イベントサブスクリプションの管理 (推奨)」をご参照ください。

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

        

      • [サブスクリプション範囲]：以下のように設定します。

        

      • [プッシュと統合]: [+ Add Channel] > [チャネルの追加] をクリックし、ステップ 1 で作成したキューを選択し、残りのフィールドに入力します。 詳細については、「プッシュチャネルの管理」をご参照ください。

3. 中断イベントのシミュレーション

   スポットインスタンスの中断は受動的にトリガーされ、開発中に再現することが困難なため、[イベントサブスクリプションのテスト] を使用して中断イベントをシミュレートします。

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

   2. [イベントデバッグの作成] パネルで、サービス を ECS に、名前 を [スポットインスタンス中断通知] に設定します。

      システムは JSON デバッグコンテンツを生成します。 リソース関連のフィールドをお使いのスポットインスタンスの情報で置き換えます。

      • Alibaba Cloud account UID をお使いのアカウント UID に置き換えます。

      • <resource-id> と <instanceId> をお使いのスポットインスタンス ID に置き換えます。

      • <region-id> をお使いのスポットインスタンスのリージョン ID に置き換えます。

        {
            "product": "ECS",
            "resourceId": "acs:ecs:cn-shanghai:Alibaba Cloud account UID:instance/<resource-id>",
            "level": "WARN",
            "instanceName": "instanceName",
            "regionId": "<region-id>",
            "groupId": "0",
            "name": "Instance:PreemptibleInstanceInterruption",
            "content": {
                "instanceId": "<instanceId>",
                "instanceName": "wor***b73",
                "action": "delete"
            },
            "status": "Normal"
        }

   3. OK をクリックします。 システムに[操作が完了しました。]と表示され、Cloud Monitor はテスト通知をキューに送信します。

4. メッセージのプルと応答

   キューから中断通知をプルし、カスタムロジックで対応します。 次のサンプルコードは、画像のグレースケール変換プログラムで中断イベントを処理する方法を示しています。

   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. InterruptedException をキャッチした後、現在の進行状況を保存
           * 2. スレッドの中断ステータスを復元 (セマンティクスを維持)
           */
          @Override
          public void run() {
              try {
                  convertToGrayScale(new File("input.jpg"), new File("output.jpg"));
                  // 後処理の遅延をシミュレート
                  Thread.sleep(5000); 
                  System.out.println("Image processing completed");
              } catch (InterruptedException e) {
                  System.out.println("Processing interrupted. Progress saved at " + progress + "%");
                  saveProgress(new File("partial_output.jpg"));
                  Thread.currentThread().interrupt(); // 中断ステータスを復元
              } catch (Exception e) {
                  System.err.println("Processing error: " + 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("Image processing interrupted");
                  }
                  /* コアグレースケールアルゴリズム */
                  int rgb = original.getRGB(x, y);
                  // RGB チャネルを抽出 (ARGB フォーマット)
                  // 赤チャネル
                  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("Auto-save failed: " + e.getMessage());
          }
      }

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

      画像プロセッサを実行中にスポットインスタンスの中断を検出して応答をテストします。

      /**
       * main メソッド (テスト用)
       * **テストシナリオ**：
       * 1. 処理スレッドを開始
       * 2. 中断通知メッセージをプル
       * 3. スレッドの終了を待機
       */
      public static void main(String[] args) throws InterruptedException {
          // MNS クライアントを初期化
          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{
              // キューからメッセージをプル
              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);
                  // "name" フィールドからイベント名を取得
                  String name = json.getString("name");
                  isMatch = "Instance:PreemptibleInstanceInterruption".equals(name);
                  // スポットインスタンスの中断イベントを処理
                  if(isMatch){
                      System.out.println("Spot instance will be interrupted and reclaimed");
                      // 画像プロセッサを停止
                      processor.stop();
                      processThread.interrupt();
                      System.out.println("Program terminated");
                      processThread.join();
                      // メッセージを削除
                      queue.deleteMessage(popMsg.getReceiptHandle());
                  }
              }
          }catch (Exception e){
              System.out.println("Unknown exception happened!");
              e.printStackTrace();
          }
          client.close();
      }

   完全なサンプルコード：

   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;
   
       /**
        * 入力画像をグレースケールに変換して保存
        * @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("Image processing interrupted");
                   }
                   /* コアグレースケールアルゴリズム */
                   int rgb = original.getRGB(x, y);
                   // RGB チャネルを抽出 (ARGB フォーマット)
                   // 赤チャネル
                   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);
       }
   
       /**
        * 処理の進行状況を指定されたファイルに保存
        * @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("Auto-save failed: " + e.getMessage());
           }
       }
   
       /**
        * スレッドのエントリポイント
        * **中断ロジック**：
        * 1. InterruptedException をキャッチした後、現在の進行状況を保存
        * 2. スレッドの中断ステータスを復元 (セマンティクスを維持)
        */
       @Override
       public void run() {
           try {
               convertToGrayScale(new File("input.jpg"), new File("output.jpg"));
               // 後処理の遅延をシミュレート
               Thread.sleep(5000); 
               System.out.println("Image processing completed");
           } catch (InterruptedException e) {
               System.out.println("Processing interrupted. Progress saved at " + progress + "%");
               saveProgress(new File("partial_output.jpg"));
               Thread.currentThread().interrupt(); // 中断ステータスを復元
           } catch (Exception e) {
               System.err.println("Processing error: " + e.getMessage());
           }
       }
   
       /**
        * 外部からの中断トリガーメソッド
        * **協調メカニズム**：
        * スレッド中断フラグと連携して二重検出を行う
        */
       public void stop() {
           running.set(false);
       }
   
       /**
        * main メソッド (テスト用)
        * **テストシナリオ**：
        * 1. 処理スレッドを開始
        * 2. キューから中断通知メッセージをプル
        * 3. スレッドの終了を待機
        */
       public static void main(String[] args) throws InterruptedException {
           // MNS クライアントを初期化
           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{
               // キューからメッセージをプル
               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);
                   // "name" フィールドからイベント名を取得
                   String name = json.getString("name");
                   isMatch = "Instance:PreemptibleInstanceInterruption".equals(name);
                   // スポットインスタンスの中断イベントを処理
                   if(isMatch){
                       System.out.println("Spot instance will be interrupted and reclaimed");
                       // 画像プロセッサを停止
                       processor.stop();
                       processThread.interrupt();
                       System.out.println("Program terminated");
                       processThread.join();
                       // メッセージを削除
                       queue.deleteMessage(popMsg.getReceiptHandle());
                   }
               }
           }catch (Exception e){
               System.out.println("Unknown exception happened!");
               e.printStackTrace();
           }
           client.close();
       }
   }