Detect and respond to spot instance interruption events using Simple Message Queue - Elastic Compute Service: Central de Documentação do Alibaba Cloud

Use o Simple Message Queue (anteriormente MNS) para receber notificações de interrupção de instâncias spot e acionar a lógica de desligamento graceful antes da recuperação do recurso.

Visão geral do fluxo de trabalho

Pré-requisitos

Crie uma AccessKey

Use a AccessKey de um usuário RAM com as permissões mínimas necessárias em vez da conta raiz. Consulte Criar uma AccessKey.

Conceda permissões ao usuário RAM

Conceda ao usuário RAM a seguinte permissão para consumir mensagens do Simple Message Queue (MNS):

Produtos Cloud	Conceder permissões
Simple Message Queue (SMQ, anteriormente MNS)	AliyunMNSFullAccess

Configure as credenciais de acesso e o endpoint

O código de exemplo lê as credenciais de acesso e o endpoint das variáveis de ambiente.
- Credenciais de acesso: consulte Configurar variáveis de ambiente no Linux, macOS e Windows.
- Endpoint: consulte Configurar o endpoint.

Instale o SDK do SMQ

Instale o SDK Java do SMQ adicionando uma dependência Maven. Para outros métodos, consulte Instalar o SDK do SMQ.

Adicione a seguinte dependência 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>

Procedimento

Crie uma fila no Simple Message Queue

Crie uma fila para receber as notificações de interrupção das instâncias spot.
1. Faça login no console do Simple Message Queue (anteriormente MNS). No painel de navegação à esquerda, escolha Queue Model > Queues.
2. Na barra de menu superior, selecione uma região. Na página Queues, clique em Create Queue.
3. No painel Create Queue, configure os parâmetros e clique em OK.
Crie uma política de assinatura

O Cloud Monitor envia notificações de interrupção de instâncias spot pelo canal de push especificado na política de assinatura.
1. Faça login no console do Cloud Monitor. No painel de navegação à esquerda, escolha .
2. Na aba Subscription Policies, clique em Create Subscription Policy. Clique em Next e configure os parâmetros.
  
  Esta seção descreve apenas os parâmetros principais. Para outros parâmetros, consulte Gerenciar assinaturas de eventos (recomendado).
  - Subscription Type: Selecione System Events.
  - Subscription Scope: Configure conforme mostrado abaixo.
  - Push and Integration: Clique em + Add Channel > Add Channel, selecione a fila criada na Etapa 1 e preencha os campos restantes. Consulte Gerenciar Canais de Push.
Simule um evento de interrupção

Use o recurso Test Event Subscription para simular um evento de interrupção, pois as interrupções de instâncias spot ocorrem passivamente e são difíceis de reproduzir durante o desenvolvimento.
1. Na aba Subscription Policies, clique em Test Event Subscription.
2. No painel Create Event Debugging, defina Service como ECS e Name como Spot Instance Interruption Notification.
  
  O sistema gera um conteúdo JSON de depuração. Substitua os campos relacionados ao recurso pelas informações da sua instância spot.
  - Substitua Alibaba Cloud account UID pelo UID da sua conta.
  - Substitua <resource-id> e <instanceId> pelo ID da sua instância spot.
  - Substitua <region-id> pelo ID da região da sua instância spot.
```
{
    "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. Clique em OK. O sistema exibe The operation is successful. e o Cloud Monitor envia uma notificação de teste para a sua fila.

Consuma e responda às mensagens

Consuma as notificações de interrupção da fila e responda com uma lógica personalizada. O código de exemplo a seguir demonstra o tratamento de um evento de interrupção com um programa de conversão de imagens para escala de cinza:

Use uma tarefa de thread para simular o programa de conversão de imagem.

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

Método de conversão de imagem para escala de cinza.

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

Salve o progresso do processamento da imagem.

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

Teste o tratamento da resposta.

Teste o processador de imagens para detectar e responder a uma interrupção de instância spot durante a execução.

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

Código de exemplo completo:

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

Nota

Para criar snapshots, consulte CreateSnapshot.
Para criar imagens personalizadas, consulte CreateImage.

Referências

Se a sua instância spot armazena dados ou configurações críticas, consulte Preservar e recuperar dados de instâncias spot para evitar perda de dados.