如何让Java @SqsListener与EventBridge管道接收同个AWS SQS队列的相同消息？

1. 是否可通过配置实现双消费者接收同一条消息？ #

不行。AWS SQS的核心设计是点对点消息模型，每条消息只会被分配给一个可用的消费者——不管是使用@SqsListener的Java应用还是EventBridge管道，本质都是该队列的消费者，SQS会自动完成消息的负载均衡，无法通过任何配置让两个消费者同时接收同一条原始消息。

2. 仅用一个SQS队列的最小改动方案 #

如果必须保留单个SQS队列，最可靠且改动最小的方案是让其中一个消费者处理完消息后，将原消息复制并重新投递回队列，同时通过自定义消息属性区分消息的处理状态，避免死循环和重复处理：

具体实现步骤： #

• 步骤1：修改Java消费者逻辑
  在@SqsListener处理完消息后，添加消息重投逻辑，给重投的消息添加自定义属性（比如processed-by-java: true），标记该消息已被Java消费者处理。同时确保消费者具备幂等性（比如通过消息ID做幂等校验）。

  @SqsListener("${app.consumer.shopper-order.queue}")
  public void listen(String message, @Header("MessageId") String messageId, AmazonSQS sqsClient) throws JsonProcessingException {
      // 1. 执行原始消息处理逻辑
      // do stuff
  
      // 2. 构造重投消息，添加自定义属性
      SendMessageRequest request = new SendMessageRequest()
              .withQueueUrl(System.getenv("app.consumer.shopper-order.queue"))
              .withMessageBody(message)
              .addMessageAttributesEntry("processed-by-java", new MessageAttributeValue()
                      .withDataType("String")
                      .withStringValue("true"));
      
      // 避免死循环：仅重投未被Java处理过的消息
      Map<String, MessageAttributeValue> currentAttrs = // 获取当前消息的属性集合
      if (currentAttrs == null || !"true".equals(currentAttrs.get("processed-by-java")?.getStringValue())) {
          sqsClient.sendMessage(request);
      }
  }
  

• 步骤2：配置EventBridge管道的消息过滤
  在Terraform中配置EventBridge管道的SQS数据源时，添加消息过滤规则，只处理带有processed-by-java: true属性的消息，确保EventBridge只消费Java消费者重投的消息：

  resource "aws_eventbridge_pipe" "sqs_to_target" {
      name     = "sqs-event-pipe"
      role_arn = aws_iam_role.pipe_role.arn
  
      source = aws_sqs_queue.shopper_order_queue.arn
      source_parameters {
          sqs_queue_parameters {
              batch_size = 10
              // 添加消息过滤规则
              filter_criteria {
                  filter {
                      pattern = "{\"MessageAttributes\": {\"processed-by-java\": {\"StringValue\": [\"true\"]}}}"
                  }
              }
          }
      }
  
      // 配置你的业务目标（如Lambda、外部服务等）
      target = aws_lambda_function.your_target.arn
  }
  

• 步骤3：确保幂等性
  两个消费者都需要实现幂等逻辑，比如通过消息ID或者业务唯一标识（如订单ID）判断消息是否已处理，避免重复执行业务操作。

备选方案（不推荐） #

如果不想修改代码，可以调整SQS的消息可见性超时，设置一个较短的时间（比如10秒）。当第一个消费者拿到消息后，若在超时时间内未删除消息，消息会重新回到队列，第二个消费者就有机会获取。但该方案不可靠：消息可能被重复投递多次，且依赖超时时间的精准设置，容易导致业务逻辑混乱，仅适用于非核心业务场景。

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