Deteksi dan tanggapi event gangguan spot instans menggunakan Simple Message Queue - Elastic Compute Service

Gunakan Simple Message Queue (formerly MNS) untuk menerima notifikasi gangguan spot instans dan memicu logika shutdown yang mulus sebelum reclaim.

Ikhtisar alur kerja

Prasyarat

Buat AccessKey
Gunakan AccessKey dari RAM user dengan izin minimum yang diperlukan, bukan akun root. Lihat Buat AccessKey.
Berikan izin kepada RAM user
Berikan izin berikut kepada RAM user untuk mengonsumsi pesan dari Simple Message Queue (MNS):
Cloud Products
Berikan izin
Simple Message Queue (SMQ, formerly MNS)
AliyunMNSFullAccess
Konfigurasikan kredensial akses dan titik akhir
Kode contoh membaca kredensial akses dan titik akhir dari variabel lingkungan.
- Kredensial akses: lihat Konfigurasikan variabel lingkungan di Linux, macOS, dan Windows.
- Titik akhir: lihat Konfigurasikan titik akhir.

Instal SDK SMQ

Instal SMQ Java SDK dengan menambahkan dependensi Maven. Untuk metode lainnya, lihat Instal SDK SMQ.

Tambahkan dependensi Maven berikut:

<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>

Prosedur

Membuat Antrian Pesan Sederhana
Buat antrian untuk menerima notifikasi gangguan spot instans.
1. Masuk ke Konsol Simple Message Queue (formerly MNS). Di panel navigasi kiri, pilih Queue Model > Queues.
2. Di bilah menu atas, pilih wilayah. Di halaman Queues, klik Create Queue.
3. Di panel Create Queue, konfigurasikan parameter dan klik OK.
  Parameter konfigurasi meliputi:
  - Name: Nama antrian, maksimal 120 karakter.
  - Maximum Message Size: Default: 64 KB.
  - Long Polling Wait Time: Default: 15 detik.
  - Message Visibility Timeout: Default: 30 detik.
  - Message Retention Period: Default: 4 hari.
  - Message Receive Delay: Default: 0 detik.
  - Enable Logging: Default: Disabled.
  - Dead-Letter Policy: Default: Disabled.
Buat kebijakan langganan
Cloud Monitor mengirim notifikasi gangguan spot instans melalui saluran push yang ditentukan dalam kebijakan langganan.
1. Masuk ke Konsol Cloud Monitor. Di panel navigasi kiri, pilih Event Hub > Event Subscriptions.
2. Di tab Subscription Policies, klik Create Subscription Policy. Klik Next dan konfigurasikan parameter.
  Hanya parameter utama yang dijelaskan di sini. Untuk parameter lainnya, lihat Kelola langganan event (disarankan).
  - Subscription Type: Pilih System Events.
  - Subscription Scope: Konfigurasikan parameter berikut:
    - Product: Pilih Elastic Compute Service.
    - Event Type: Pilih Status Notification.
    - Event Name: Pilih Spot Instance Interruption Notification.
    - Event Level: Pilih Warning.
  - Push and Integration: Klik + Add Channel > Add Channel, pilih antrian yang dibuat pada Langkah 1, lalu lengkapi kolom lainnya. Lihat Kelola Saluran Push.
Simulasikan event gangguan
Gunakan Test Event Subscription untuk mensimulasikan event gangguan, karena gangguan spot instans dipicu secara pasif dan sulit direproduksi selama pengembangan.
1. Di tab Subscription Policies, klik Test Event Subscription.
2. Di panel Create Event Debugging, atur Service menjadi ECS dan Name menjadi Spot Instance Interruption Notification.
  Sistem akan menghasilkan konten debugging JSON. Ganti bidang terkait resource dengan informasi spot instans Anda.
  - Ganti Alibaba Cloud account UID dengan UID akun Anda.
  - Ganti <resource-id> dan <instanceId> dengan ID spot instans Anda.
  - Ganti <region-id> dengan ID wilayah spot instans Anda.
```
{
    "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. Klik OK. Sistem menampilkan The operation is successful., dan Cloud Monitor mengirim notifikasi uji coba ke antrian Anda.

Ambil dan tanggapi pesan

Ambil notifikasi gangguan dari antrian dan tanggapi dengan logika kustom. Kode contoh berikut menunjukkan penanganan event gangguan menggunakan program konversi gambar ke skala abu-abu:

Gunakan tugas thread untuk mensimulasikan program konversi gambar.

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;

/**
 * Interruptible image processor that supports grayscale conversion.
 * Uses atomic variables and thread interruption flags for detection.
 * Features:
 * 1. Chunked processing with automatic progress saving
 * 2. Immediate response to spot instance interruption
 * 3. Partial result file generation after interruption
 */
public class InterruptibleImageProcessor implements Runnable {
    /**
     * Thread-safe state control using an atomic Boolean
     */
    private final AtomicBoolean running = new AtomicBoolean(true);
    /**
     * Stores image data being processed
     */
    private BufferedImage processedImage;
    /**
     * Processing progress percentage (0–100)
     */
    private int progress;
    /**
     * Thread entry point
     * **Interruption logic**:
     * 1. Save current progress after catching InterruptedException
     * 2. Restore thread interruption status (preserve semantics)
     */
    @Override
    public void run() {
        try {
            convertToGrayScale(new File("input.jpg"), new File("output.jpg"));
            // Simulate post-processing delay
            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(); // Restore interruption status
        } catch (Exception e) {
            System.err.println("Processing error: " + e.getMessage());
        }
    }

    /**
     * External interruption trigger method
     * **Coordination mechanism**:
     * Works with thread interruption flag for dual detection
     */
    public void stop() {
        running.set(false);
    }
}

Metode konversi gambar ke skala abu-abu.

/**
 * Convert input image to grayscale and save
 * @param inputFile Source image file object
 * @param outputFile Output file object
 * @throws Exception Includes I/O and interruption exceptions
 *
 * **Algorithm**:
 * Weighted average grayscale conversion matching human luminance perception:
 * Gray = 0.30*R + 0.59*G + 0.11*B
 * Reference: ITU-R BT.601 standard
 */
public void convertToGrayScale(File inputFile, File outputFile) throws Exception {
    // Read source image
    BufferedImage original = ImageIO.read(inputFile);
    int width = original.getWidth();
    int height = original.getHeight();
    // Create grayscale image buffer
    processedImage = new BufferedImage(width, height, BufferedImage.TYPE_BYTE_GRAY);

    // Process in chunks to support progress saving
    for (int y = 0; y < height && running.get(); y++) {
        // Process pixel by pixel
        for (int x = 0; x < width; x++) {
            // First interruption check: thread interruption flag
            if (Thread.interrupted()) {
                throw new InterruptedException("Image processing interrupted");
            }
            /* Core grayscale algorithm */
            int rgb = original.getRGB(x, y);
            // Extract RGB channels (ARGB format)
            // Red channel
            int r = (rgb >> 16) & 0xFF;
            // Green channel
            int g = (rgb >> 8) & 0xFF;
            // Blue channel
            int b = rgb & 0xFF;
            // Calculate grayscale value (weighted average)
            int gray = (int)(0.3 * r + 0.59 * g + 0.11 * b);
            // Reconstruct RGB (grayscale value replicated across channels)
            processedImage.setRGB(x, y, (gray << 16) | (gray << 8) | gray);
            // Update progress percentage (watch for integer division)
            progress = (y * width + x) * 100 / (width * height);
        }
        // Auto-save progress every 50 rows (checkpoint mechanism)
        if (y % 50 == 0) {
            saveProgress(outputFile);
        }
    }
    // Save final result
    ImageIO.write(processedImage, "jpg", outputFile);
}

Simpan progres pemrosesan gambar.

/**
 * Save processing progress to specified file
 * @param outputFile Output file object
 *
 * **Note**:
 * 1. Uses silent failure to avoid interrupting save process
 * 2. Generates temporary file named partial_output.jpg
 */
private void saveProgress(File outputFile) {
    try {
        // Use temporary filename to avoid overwriting final file
        ImageIO.write(processedImage, "jpg", new File("partial_output.jpg"));
    } catch (Exception e) {
        System.err.println("Auto-save failed: " + e.getMessage());
    }
}

Uji penanganan respons.

Uji prosesor gambar untuk mendeteksi dan merespons gangguan spot instans saat sedang berjalan.

/**
 * Main method (for testing)
 * **Test scenario**:
 * 1. Start processing thread
 * 2. Pull interruption notification message
 * 3. Wait for thread termination
 */
public static void main(String[] args) throws InterruptedException {
    // Initialize MNS client
    CloudAccount account = new CloudAccount(
            ServiceSettings.getMNSAccessKeyId(),
            ServiceSettings.getMNSAccessKeySecret(),
            ServiceSettings.getMNSAccountEndpoint());
    MNSClient client = account.getMNSClient();
    // Check if message matches spot instance interruption event
    boolean isMatch = false;
    // Start image processor
    InterruptibleImageProcessor processor = new InterruptibleImageProcessor();
    Thread processThread = new Thread(processor);
    processThread.start();
    try{
        // Pull message from queue
        CloudQueue queue = client.getQueueRef("spot-interruption");
        Message popMsg = queue.popMessage();
        if (popMsg != null){
            // Message body is Base64-encoded by default
            System.out.println("message body: " + popMsg.getMessageBodyAsRawString());
            // Base64 decode
            byte[] decodedBytes = Base64.getDecoder().decode(popMsg.getMessageBodyAsRawString());
            String decodedString = new String(decodedBytes);
            System.out.println("message content: " + decodedString);
            // Parse JSON string
            JSONObject json = new JSONObject(decodedString);
            // Get event name from "name" field
            String name = json.getString("name");
            isMatch = "Instance:PreemptibleInstanceInterruption".equals(name);
            // Handle spot instance interruption event
            if(isMatch){
                System.out.println("Spot instance will be interrupted and reclaimed");
                // Stop image processor
                processor.stop();
                processThread.interrupt();
                System.out.println("Program terminated");
                processThread.join();
                // Delete message
                queue.deleteMessage(popMsg.getReceiptHandle());
            }
        }
    }catch (Exception e){
        System.out.println("Unknown exception happened!");
        e.printStackTrace();
    }
    client.close();
}

Kode contoh lengkap:

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;

/**
 * Interruptible image processor that supports grayscale conversion.
 * Uses atomic variables and thread interruption flags for detection.
 * Features:
 * 1. Chunked processing with automatic progress saving
 * 2. Immediate response to spot instance interruption
 * 3. Partial result file generation after interruption
 */
public class InterruptibleImageProcessor implements Runnable {
    /**
     * Thread-safe state control using an atomic Boolean
     */
    private final AtomicBoolean running = new AtomicBoolean(true);
    /**
     * Stores image data being processed
     */
    private BufferedImage processedImage;
    /**
     * Processing progress percentage (0–100)
     */
    private int progress;

    /**
     * Convert input image to grayscale and save
     * @param inputFile Source image file object
     * @param outputFile Output file object
     * @throws Exception Includes I/O and interruption exceptions
     *
     * **Algorithm**:
     * Weighted average grayscale conversion matching human luminance perception:
     * Gray = 0.30*R + 0.59*G + 0.11*B
     * Reference: ITU-R BT.601 standard
     */
    public void convertToGrayScale(File inputFile, File outputFile) throws Exception {
        // Read source image
        BufferedImage original = ImageIO.read(inputFile);
        int width = original.getWidth();
        int height = original.getHeight();
        // Create grayscale image buffer
        processedImage = new BufferedImage(width, height, BufferedImage.TYPE_BYTE_GRAY);

        // Process in chunks to support progress saving
        for (int y = 0; y < height && running.get(); y++) {
            // Process pixel by pixel
            for (int x = 0; x < width; x++) {
                // First interruption check: thread interruption flag
                if (Thread.interrupted()) {
                    throw new InterruptedException("Image processing interrupted");
                }
                /* Core grayscale algorithm */
                int rgb = original.getRGB(x, y);
                // Extract RGB channels (ARGB format)
                // Red channel
                int r = (rgb >> 16) & 0xFF;
                // Green channel
                int g = (rgb >> 8) & 0xFF;
                // Blue channel
                int b = rgb & 0xFF;
                // Calculate grayscale value (weighted average)
                int gray = (int)(0.3 * r + 0.59 * g + 0.11 * b);
                // Reconstruct RGB (grayscale value replicated across channels)
                processedImage.setRGB(x, y, (gray << 16) | (gray << 8) | gray);
                // Update progress percentage (watch for integer division)
                progress = (y * width + x) * 100 / (width * height);
            }
            // Auto-save progress every 50 rows (checkpoint mechanism)
            if (y % 50 == 0) {
                saveProgress(outputFile);
            }
        }
        // Save final result
        ImageIO.write(processedImage, "jpg", outputFile);
    }

    /**
     * Save processing progress to specified file
     * @param outputFile Output file object
     *
     * **Note**:
     * 1. Uses silent failure to avoid interrupting save process
     * 2. Generates temporary file named partial_output.jpg
     */
    private void saveProgress(File outputFile) {
        try {
            // Use temporary filename to avoid overwriting final file
            ImageIO.write(processedImage, "jpg", new File("partial_output.jpg"));
        } catch (Exception e) {
            System.err.println("Auto-save failed: " + e.getMessage());
        }
    }

    /**
     * Thread entry point
     * **Interruption logic**:
     * 1. Save current progress after catching InterruptedException
     * 2. Restore thread interruption status (preserve semantics)
     */
    @Override
    public void run() {
        try {
            convertToGrayScale(new File("/Users/shaoberlin/Desktop/idea_workspace/aliyun/src/main/resources/input.jpg"), new File("output.jpg"));
            // Simulate post-processing delay
            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(); // Restore interruption status
        } catch (Exception e) {
            System.err.println("Processing error: " + e.getMessage());
        }
    }

    /**
     * External interruption trigger method
     * **Coordination mechanism**:
     * Works with thread interruption flag for dual detection
     */
    public void stop() {
        running.set(false);
    }

    /**
     * Main method (for testing)
     * **Test scenario**:
     * 1. Start processing thread
     * 2. Trigger interruption after 5000 ms
     * 3. Wait for thread termination
     */
    public static void main(String[] args) throws InterruptedException {
        // Initialize MNS client
        CloudAccount account = new CloudAccount(
                ServiceSettings.getMNSAccessKeyId(),
                ServiceSettings.getMNSAccessKeySecret(),
                ServiceSettings.getMNSAccountEndpoint());
        MNSClient client = account.getMNSClient();
        // Check if message matches spot instance interruption event
        boolean isMatch = false;
        // Start image processor
        InterruptibleImageProcessor processor = new InterruptibleImageProcessor();
        Thread processThread = new Thread(processor);
        processThread.start();
        try{
            // Pull message from queue
            CloudQueue queue = client.getQueueRef("spot-interruption");
            Message popMsg = queue.popMessage();
            if (popMsg != null){
                // Message body is Base64-encoded by default
                System.out.println("message body: " + popMsg.getMessageBodyAsRawString());
                // Base64 decode
                byte[] decodedBytes = Base64.getDecoder().decode(popMsg.getMessageBodyAsRawString());
                String decodedString = new String(decodedBytes);
                System.out.println("message content: " + decodedString);
                // Parse JSON string
                JSONObject json = new JSONObject(decodedString);
                // Get event name from "name" field
                String name = json.getString("name");
                isMatch = "Instance:PreemptibleInstanceInterruption".equals(name);
                // Handle spot instance interruption event
                if(isMatch){
                    System.out.println("Spot instance will be interrupted and reclaimed");
                    // Stop image processor
                    processor.stop();
                    processThread.interrupt();
                    System.out.println("Program terminated");
                    processThread.join();
                    // Delete message
                    queue.deleteMessage(popMsg.getReceiptHandle());
                }
            }
        }catch (Exception e){
            System.out.println("Unknown exception happened!");
            e.printStackTrace();
        }
        client.close();
    }
}

Catatan

Untuk membuat snapshot, lihat CreateSnapshot.
Untuk membuat gambar kustom, lihat CreateImage.

Referensi

Jika spot instans Anda menyimpan data atau konfigurasi penting, lihat Pertahankan dan pulihkan data untuk spot instans guna mencegah kehilangan data.

Cloud Products	Berikan izin
Simple Message Queue (SMQ, formerly MNS)	AliyunMNSFullAccess