Detect and respond to spot instance interruption events using Simple Message Queue - Elastic Compute Service

Spot örneklerin geri alınmasından önce zarif kapatma mantığını tetiklemek üzere kesinti bildirimlerini almak için Simple Message Queue'yu (önceki adıyla MNS) kullanın.

İş akışı genel bakış

Ön koşullar

Bir AccessKey oluşturun

Kök hesap yerine minimum gerekli izinlere sahip bir RAM kullanıcısının AccessKey'ini kullanın. Bkz. AccessKey oluşturma.

RAM kullanıcısına izinler verin

RAM kullanıcısına Simple Message Queue'dan (MNS) mesajları tüketmek için aşağıdaki izni verin:

Cloud Ürünleri	İzinleri verin
Simple Message Queue (SMQ, önceki adıyla MNS)	AliyunMNSFullAccess

Erişim kimlik bilgilerini ve uç noktayı yapılandırın

Örnek kod, erişim kimlik bilgilerini ve uç noktayı ortam değişkenlerinden okur.
- Erişim kimlik bilgileri: bkz. Linux, macOS ve Windows'ta ortam değişkenlerini yapılandırma.
- Uç nokta: bkz. Uç noktayı yapılandırma.

SMQ SDK'yı yükleyin

Bir Maven bağımlılığı ekleyerek SMQ Java SDK'yı yükleyin. Diğer yöntemler için bkz. SMQ SDK'yı yükleme.

Aşağıdaki Maven bağımlılığını ekleyin:

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

Yordam

Bir Simple Message Queue oluşturun

Spot örnek kesinti bildirimlerini almak için bir kuyruk oluşturun.
1. Simple Message Queue (önceki adıyla MNS) konsoluna giriş yapın. Sol gezinme bölmesinde Queue Model > Queues seçeneğini belirleyin.
2. Üst menü çubuğunda bir bölge seçin. Queues sayfasında Create Queue öğesini tıklayın.
3. Create Queue panelinde parametreleri yapılandırın ve OK öğesini tıklayın.
  
  Yapılandırma parametreleri şunları içerir:
  - Name: Kuyruk adı, en fazla 120 karakter.
  - Maximum Message Size: Varsayılan: 64 KB.
  - Long Polling Wait Time: Varsayılan: 15 saniye.
  - Message Visibility Timeout: Varsayılan: 30 saniye.
  - Message Retention Period: Varsayılan: 4 gün.
  - Message Receive Delay: Varsayılan: 0 saniye.
  - Enable Logging: Varsayılan: Devre dışı.
  - Dead-Letter Policy: Varsayılan: Devre dışı.
Bir abonelik ilkesi oluşturun

Cloud Monitor, abonelik ilkesinde belirtilen gönderim kanalı aracılığıyla spot örnek kesinti bildirimlerini gönderir.
1. Cloud Monitor konsoluna giriş yapın. Sol gezinme bölmesinde Event Hub > Event Subscriptions seçeneğini belirleyin.
2. Subscription Policies sekmesinde Create Subscription Policy öğesini tıklayın. İleri'yi tıklayın ve parametreleri yapılandırın.
  
  Burada yalnızca anahtar parametreler açıklanmaktadır. Diğer parametreler için bkz. Olay aboneliklerini yönetme (önerilen).
  - Subscription Type: System Events seçeneğini belirleyin.
  - Subscription Scope: Aşağıdaki parametreleri yapılandırın:
    - Product: Elastic Compute Service seçeneğini belirleyin.
    - Event Type: Status Notification seçeneğini belirleyin.
    - Event Name: Spot Instance Interruption Notification seçeneğini belirleyin.
    - Event Level: Warning seçeneğini belirleyin.
  - Push and Integration: + Add Channel > Add Channel öğelerini tıklayın, 1. Adımda oluşturulan kuyruğu seçin ve kalan alanları doldurun. Bkz. Gönderim Kanallarını Yönetme.
Bir kesinti olayını simüle edin

Spot örnek kesintileri pasif olarak tetiklendiğinden ve geliştirme sırasında çoğaltılması zor olduğundan, bir kesinti olayını simüle etmek için Test Event Subscription özelliğini kullanın.
1. Subscription Policies sekmesinde Test Event Subscription öğesini tıklayın.
2. Create Event Debugging panelinde Service değerini ECS ve Name değerini Spot Instance Interruption Notification olarak ayarlayın.
  
  Sistem JSON hata ayıklama içeriği oluşturur. Kaynakla ilgili alanları spot örnek bilgilerinizle değiştirin.
  - Alibaba Cloud account UID değerini hesap UID'nizle değiştirin.
  - <resource-id> ve <instanceId> değerlerini spot örnek ID'nizle değiştirin.
  - <region-id> değerini spot örnek bölge ID'nizle değiştirin.
```
{
    "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 öğesini tıklayın. Sistem The operation is successful. mesajını görüntüler ve Cloud Monitor kuyruğunuza bir test bildirimi gönderir.

Mesajları çekin ve yanıt verin

Kesinti bildirimlerini kuyruktan çekin ve özel mantıkla yanıt verin. Aşağıdaki örnek kod, bir görüntü gri tonlama dönüştürme programı ile bir kesinti olayının nasıl ele alındığını göstermektedir:

Görüntü dönüştürme programını simüle etmek için bir iş parçacığı görevi kullanın.

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);
    }
}

Görüntü gri tonlama dönüştürme yöntemi.

/**
 * 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);
}

Görüntü işleme ilerlemesini kaydedin.

/**
 * 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());
    }
}

Yanıt işlemeyi test edin.

Çalışırken spot örnek kesintisini algılayıp buna yanıt verdiğini görmek için görüntü işleyiciyi test edin.

/**
 * 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();
}

Tam örnek kod:

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();
    }
}

Not

Anlık görüntü oluşturmak için bkz. CreateSnapshot.
Özel görüntüler oluşturmak için bkz. CreateImage.

Referanslar

Spot örneğiniz kritik veriler veya yapılandırmalar depoluyorsa veri kaybını önlemek için bkz. Öncelikli örnekler için verileri koruma ve kurtarma.