如何从AWS SQS响应队列获取对应请求的消息（Java+Step Functions）

核心思路 #

给每个请求生成全局唯一的RequestID，让这个ID贯穿整个流程：Step Functions发起请求时携带该ID，Lambda处理完成后将响应与RequestID绑定写入响应队列，最后Step Functions从响应队列中筛选出与当前执行上下文一致的RequestID对应的消息，确保并发场景下不会拿错响应。

具体实现步骤 #

1. 生成并传递唯一RequestID #

在Step Functions的起始步骤生成RequestID（比如UUID），并将其存入执行上下文，后续所有步骤都携带这个ID：

• 可以用Step Functions的Pass状态结合内置函数生成：

  {
    "Type": "Pass",
    "Parameters": {
      "RequestID.$": "States.UUID()",
      "BusinessData.$": "$.BusinessData"
    },
    "Next": "SendToABCQueue"
  }
  

• 或者用Java Lambda生成ID后传入Step Functions执行上下文。

2. 发送请求到ABCQueue #

在Step Functions的SendToABCQueue步骤，将RequestID和业务数据一起封装成消息体发送到SQS：

import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.SendMessageRequest;

public class SendToQueueHandler {
    private final SqsClient sqsClient = SqsClient.create();
    private static final String ABC_QUEUE_URL = "https://sqs.region.amazonaws.com/123456789012/ABCQueue";

    public void sendMessage(String requestID, String businessData) {
        String messageBody = String.format("{\"RequestID\":\"%s\",\"BusinessData\":\"%s\"}", requestID, businessData);
        
        SendMessageRequest request = SendMessageRequest.builder()
                .queueUrl(ABC_QUEUE_URL)
                .messageBody(messageBody)
                .build();
        
        sqsClient.sendMessage(request);
    }
}

3. Lambda处理请求并写入XYZQueue #

Lambda从ABCQueue接收消息，提取RequestID，处理业务逻辑后，将RequestID和响应数据绑定写入XYZQueue：

import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.SendMessageRequest;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.SQSEvent;

public class ProcessRequestHandler implements RequestHandler<SQSEvent, Void> {
    private final SqsClient sqsClient = SqsClient.create();
    private static final String XYZ_QUEUE_URL = "https://sqs.region.amazonaws.com/123456789012/XYZQueue";

    @Override
    public Void handleRequest(SQSEvent event, Context context) {
        for (SQSEvent.SQSMessage msg : event.getRecords()) {
            // 解析消息体获取RequestID和业务数据
            String messageBody = msg.getBody();
            // 实际开发建议用Jackson等JSON库解析，示例简化处理
            String requestID = extractRequestID(messageBody);
            String businessData = extractBusinessData(messageBody);
            
            // 执行业务处理逻辑
            String responseData = processBusinessLogic(businessData);
            
            // 将RequestID和响应数据写入XYZQueue
            String responseMessage = String.format("{\"RequestID\":\"%s\",\"ResponseData\":\"%s\"}", requestID, responseData);
            SendMessageRequest sendRequest = SendMessageRequest.builder()
                    .queueUrl(XYZ_QUEUE_URL)
                    .messageBody(responseMessage)
                    .build();
            sqsClient.sendMessage(sendRequest);
        }
        return null;
    }

    private String extractRequestID(String body) {
        return body.split("\"RequestID\":\"")[1].split("\"")[0];
    }

    private String extractBusinessData(String body) {
        return body.split("\"BusinessData\":\"")[1].split("\"")[0];
    }

    private String processBusinessLogic(String data) {
        // 替换为实际业务处理代码
        return "Processed_" + data;
    }
}

4. Step Functions从XYZQueue筛选匹配的响应 #

Step Functions通过轮询XYZQueue，每次获取消息后检查RequestID是否与当前执行上下文的ID一致：

1. 用Step Functions的Task状态调用SQS的ReceiveMessage接口获取消息；
2. 用Choice状态判断消息中的RequestID是否等于上下文的RequestID：
   • 匹配：处理响应，流程继续；
   • 不匹配：将消息重新放回队列（设置合适的可见性超时），然后重新轮询。

轮询任务的Java实现（也可直接用Step Functions集成SQS）

import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
import software.amazon.awssdk.services.sqs.model.DeleteMessageRequest;
import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityRequest;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import java.util.Map;

public class PollXYZQueueHandler implements RequestHandler<Map<String, String>, Map<String, String>> {
    private final SqsClient sqsClient = SqsClient.create();
    private static final String XYZ_QUEUE_URL = "https://sqs.region.amazonaws.com/123456789012/XYZQueue";
    private static final int VISIBILITY_TIMEOUT = 30; // 消息可见性超时30秒

    @Override
    public Map<String, String> handleRequest(Map<String, String> input, Context context) {
        String targetRequestID = input.get("RequestID");
        
        ReceiveMessageRequest receiveRequest = ReceiveMessageRequest.builder()
                .queueUrl(XYZ_QUEUE_URL)
                .maxNumberOfMessages(1)
                .visibilityTimeout(VISIBILITY_TIMEOUT)
                .build();
        
        ReceiveMessageResponse response = sqsClient.receiveMessage(receiveRequest);
        
        for (software.amazon.awssdk.services.sqs.model.Message msg : response.messages()) {
            String messageBody = msg.body();
            String msgRequestID = extractRequestID(messageBody);
            
            if (targetRequestID.equals(msgRequestID)) {
                // 匹配成功，删除队列中的消息并返回响应
                DeleteMessageRequest deleteRequest = DeleteMessageRequest.builder()
                        .queueUrl(XYZ_QUEUE_URL)
                        .receiptHandle(msg.receiptHandle())
                        .build();
                sqsClient.deleteMessage(deleteRequest);
                
                String responseData = extractResponseData(messageBody);
                return Map.of("Success", "true", "ResponseData", responseData);
            } else {
                // 不匹配，立即释放消息可见性，让其他请求能读取
                sqsClient.changeMessageVisibility(
                        ChangeMessageVisibilityRequest.builder()
                                .queueUrl(XYZ_QUEUE_URL)
                                .receiptHandle(msg.receiptHandle())
                                .visibilityTimeout(0)
                                .build()
                );
            }
        }
        
        // 未获取到匹配消息，返回失败让Step Functions重试
        return Map.of("Success", "false");
    }

    private String extractRequestID(String body) {
        return body.split("\"RequestID\":\"")[1].split("\"")[0];
    }

    private String extractResponseData(String body) {
        return body.split("\"ResponseData\":\"")[1].split("\"")[0];
    }
}

关键注意事项 #

• 可见性超时设置：轮询时设置合适的可见性超时，避免不匹配的消息长时间被锁定；
• Step Functions重试机制：配置轮询任务的重试策略，设置合理的重试间隔和次数；
• 消息去重：给SQS开启内容重复检测，避免重复写入响应消息；
• RequestID传递：确保RequestID在整个流程中不丢失，Step Functions上下文要正确传递该变量；
• 异常处理：处理Lambda执行失败、SQS消息丢失等异常场景，避免流程卡死。

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