Java如何获取AWS Elemental MediaConvert作业状态进度并实现完成监听

• EventBridge（原CloudWatch Events）：MediaConvert作业状态发生变更时会自动推送事件到EventBridge，你可以配置规则过滤作业完成、失败的状态事件，再将符合条件的事件投递到下游组件
• SQS（简单消息队列）：作为EventBridge的目标端缓存作业状态事件，Java代码可以轮询SQS获取通知，适合异步解耦的生产场景
• SNS（简单通知服务）：也可作为EventBridge的目标端，支持主动推送通知到Java服务的HTTP端点、邮件、短信等，适合需要实时推送的场景
• 轻量测试场景下也可以直接调用MediaConvert SDK轮询作业状态，不推荐生产使用，会造成资源浪费且通知延迟高

方案1：生产级方案（EventBridge + SQS） #

前置AWS资源配置 #

1. 创建标准SQS队列，配置权限允许EventBridge向该队列发送消息
2. 在EventBridge创建事件规则，匹配MediaConvert作业状态变更事件，事件模式参考：

{
  "source": ["aws.mediaconvert"],
  "detail-type": ["MediaConvert Job State Change"],
  "detail": {
    "status": ["COMPLETE", "ERROR"]
  }
}

3. 将之前创建的SQS队列设置为该事件规则的目标

Java监听代码实现 #

首先引入AWS SDK for Java 2.x依赖（1.x版本逻辑类似，可对应调整API）：

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>sqs</artifactId>
    <version>2.20.0</version> <!-- 替换为最新稳定版本即可 -->
</dependency>
<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>mediaconvert</artifactId>
    <version>2.20.0</version>
</dependency>

编写SQS轮询监听逻辑，拿到事件后匹配你提交作业时存储的jobID即可触发业务逻辑：

import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.*;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.List;

public class MediaConvertJobListener {
    // 替换为你自己的SQS队列URL、资源区域
    private static final String SQS_QUEUE_URL = "https://sqs.xxx.amazonaws.com/xxx/xxx";
    private static final Region REGION = Region.US_EAST_1;
    private static final ObjectMapper objectMapper = new ObjectMapper();

    public static void main(String[] args) {
        SqsClient sqsClient = SqsClient.builder()
                .region(REGION)
                .build();

        // 持续轮询SQS消息
        while (!Thread.currentThread().isInterrupted()) {
            ReceiveMessageRequest receiveRequest = ReceiveMessageRequest.builder()
                    .queueUrl(SQS_QUEUE_URL)
                    .waitTimeSeconds(20) // 开启长轮询减少空请求消耗
                    .maxNumberOfMessages(10)
                    .build();

            List<Message> messages = sqsClient.receiveMessage(receiveRequest).messages();
            for (Message message : messages) {
                try {
                    // 解析事件内容
                    JsonNode eventBody = objectMapper.readTree(message.body());
                    JsonNode detail = eventBody.get("detail");
                    String jobId = detail.get("jobId").asText();
                    String jobStatus = detail.get("status").asText();

                    // 匹配目标jobID后处理业务逻辑
                    if (jobId.equals("你提交作业时保存的jobID")) {
                        if ("COMPLETE".equals(jobStatus)) {
                            System.out.println("转码作业完成，jobID：" + jobId);
                            // 此处添加作业完成后的业务逻辑
                        } else if ("ERROR".equals(jobStatus)) {
                            String errorMsg = detail.get("errorMessage").asText();
                            System.out.println("转码作业失败，jobID：" + jobId + "，错误信息：" + errorMsg);
                            // 此处添加作业失败后的处理逻辑
                        }

                        // 处理完成后删除消息，避免重复消费
                        DeleteMessageRequest deleteRequest = DeleteMessageRequest.builder()
                                .queueUrl(SQS_QUEUE_URL)
                                .receiptHandle(message.receiptHandle())
                                .build();
                        sqsClient.deleteMessage(deleteRequest);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
        sqsClient.close();
    }
}

方案2：轻量测试方案（直接轮询MediaConvert作业状态） #

如果是小范围测试，不想额外配置其他AWS组件，可以直接调用MediaConvert的getJob接口轮询作业状态：

import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.mediaconvert.MediaConvertClient;
import software.amazon.awssdk.services.mediaconvert.model.GetJobRequest;
import software.amazon.awssdk.services.mediaconvert.model.Job;
import software.amazon.awssdk.services.mediaconvert.model.JobStatus;

public class MediaConvertPollingListener {
    // 替换为你的资源区域、MediaConvert专属端点（可在MediaConvert控制台查询）
    private static final Region REGION = Region.US_EAST_1;
    private static final String MEDIACONVERT_ENDPOINT = "https://xxx.mediaconvert.xxx.amazonaws.com";

    public void listenJobStatus(String targetJobId) throws InterruptedException {
        MediaConvertClient mediaConvertClient = MediaConvertClient.builder()
                .region(REGION)
                .endpointOverride(java.net.URI.create(MEDIACONVERT_ENDPOINT))
                .build();

        GetJobRequest jobRequest = GetJobRequest.builder()
                .id(targetJobId)
                .build();

        while (true) {
            Job job = mediaConvertClient.getJob(jobRequest).job();
            JobStatus status = job.status();

            if (status == JobStatus.COMPLETE) {
                System.out.println("转码作业完成：" + targetJobId);
                // 作业完成业务逻辑
                break;
            } else if (status == JobStatus.ERROR || status == JobStatus.CANCELED) {
                System.out.println("转码作业失败/取消：" + targetJobId + "，状态：" + status);
                // 错误处理逻辑
                break;
            }

            // 轮询间隔可自行调整，此处设置为5秒
            Thread.sleep(5000);
        }
        mediaConvertClient.close();
    }
}

内容的提问来源于stack exchange，提问作者robberson