如何配置基于SpringBoot的Kafka连接器实现SQS到Kafka的消息同步并获取样板代码

基础依赖配置 #

在现有Spring Boot项目中引入以下核心依赖即可满足开发要求，无需引入重型的Kafka Connect框架：

• spring-cloud-starter-aws-sqs：自动装配AWS SQS客户端，支持凭证自动发现（EC2实例角色、环境变量、配置文件多渠道适配）
• spring-kafka：提供生产端到Kafka的消息发送能力
• spring-boot-starter-actuator（可选）：提供应用健康检查、自定义指标埋点能力

外置配置项示例 #

所有动态参数全部外置到application.yml，避免硬编码：

# AWS SQS 配置
aws:
  region: cn-north-1
  credentials:
    access-key: ${AWS_ACCESS_KEY:}
    secret-key: ${AWS_SECRET_KEY:}
sqs:
  queue-url: https://sqs.xxx.amazonaws.com.cn/your-queue-url
  # 拉取参数：长轮询20秒（AWS允许的最大长轮询时长，可减少空请求成本）
  receive-wait-seconds: 20
  max-batch-size: 10
# Kafka 配置
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      acks: all
      retries: 3
      enable-idempotence: true
kafka:
  target-topic: sqs-sync-target-topic
# 连接器总开关
connector:
  enabled: true

注意：AWS凭证严禁硬编码在配置文件中，优先通过ECS/EC2实例角色、环境变量、配置中心密钥管理服务注入。

核心实现代码（启动自动触发） #

通过实现ApplicationRunner接口，在Spring应用完全启动、健康检查通过后自动启动消息拉取流程，避免启动阶段空跑；拉取线程设置为守护线程，支持优雅停机：

@Component
@RequiredArgsConstructor
@Slf4j
public class SqsToKafkaConnectorRunner implements ApplicationRunner {
    private final SqsClient sqsClient;
    private final KafkaTemplate<String, String> kafkaTemplate;

    @Value("${sqs.queue-url}")
    private String queueUrl;
    @Value("${sqs.max-batch-size:10}")
    private int maxBatchSize;
    @Value("${sqs.receive-wait-seconds:20}")
    private int receiveWaitSeconds;
    @Value("${kafka.target-topic}")
    private String targetTopic;
    @Value("${connector.enabled:true}")
    private boolean connectorEnabled;

    private ExecutorService pullExecutor;

    @Override
    public void run(ApplicationArguments args) {
        if (!connectorEnabled) {
            log.info("SQS to Kafka 连接器已关闭，跳过启动");
            return;
        }
        // 初始化单线程拉取池，设置为守护线程不阻塞应用关闭
        pullExecutor = Executors.newSingleThreadExecutor(threadFactory -> {
            Thread pullThread = new Thread(threadFactory, "sqs-kafka-pull-worker");
            pullThread.setDaemon(true);
            return pullThread;
        });
        pullExecutor.submit(this::pullMessageLoop);
    }

    /**
     * 核心消息拉取循环：拉取→发送Kafka→确认成功后删除SQS消息
     */
    private void pullMessageLoop() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                // 1. 长轮询拉取SQS消息
                ReceiveMessageRequest receiveRequest = ReceiveMessageRequest.builder()
                        .queueUrl(queueUrl)
                        .maxNumberOfMessages(maxBatchSize)
                        .waitTimeSeconds(receiveWaitSeconds)
                        .build();
                List<Message> sqsMessages = sqsClient.receiveMessage(receiveRequest).messages();
                if (sqsMessages.isEmpty()) {
                    continue;
                }
                log.debug("从SQS拉取到{}条消息", sqsMessages.size());

                // 2. 批量发送到Kafka，key设为SQS的receiptHandle
                List<CompletableFuture<SendResult<String, String>>> sendFutures = sqsMessages.stream()
                        .map(msg -> kafkaTemplate.send(targetTopic, msg.receiptHandle(), msg.body()))
                        .toList();

                // 3. 等待所有Kafka消息确认写入成功
                List<SendResult<String, String>> sendResults = sendFutures.stream()
                        .map(CompletableFuture::join)
                        .toList();

                // 4. 批量删除SQS中已确认发送成功的消息
                List<DeleteMessageBatchRequestEntry> deleteEntries = sendResults.stream()
                        .map(result -> DeleteMessageBatchRequestEntry.builder()
                                .id(UUID.randomUUID().toString())
                                .receiptHandle(result.getProducerRecord().key())
                                .build())
                        .toList();
                sqsClient.deleteMessageBatch(DeleteMessageBatchRequest.builder()
                        .queueUrl(queueUrl)
                        .entries(deleteEntries)
                        .build());
                log.debug("本轮{}条消息已成功同步到Kafka", deleteEntries.size());
            } catch (Exception e) {
                log.error("消息流转流程异常", e);
                // 异常时休眠1秒，避免无限刷错造成资源浪费
                LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));
            }
        }
    }

    /**
     * 应用关闭时优雅停止拉取线程
     */
    @PreDestroy
    public void stopPullWorker() {
        if (pullExecutor == null) {
            return;
        }
        pullExecutor.shutdown();
        try {
            if (!pullExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
                pullExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            pullExecutor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

核心设计符合行业最佳实践的说明 #

• 可靠性兜底：严格遵循「先等Kafka写入成功，再删除SQS消息」的流程，结合Kafka Producer的幂等配置、acks=all策略，实现至少一次的消息投递语义，不会因为流程中断导致消息丢失
• 启动适配：拉取逻辑在应用完全初始化后触发，线程为守护线程，支持优雅停机，不会因为连接器阻塞Spring容器启动
• 成本优化：采用AWS推荐的20秒长轮询机制，减少无效空请求，降低SQS调用成本
• 故障隔离：流程异常时增加退避休眠，避免异常风暴打满线程池；发送失败的消息不会被从SQS删除，会在可见超时后自动重投

后续功能迭代指导方向 #

基础迭代阶段（上线前必须补齐） #

• 增加监控埋点：统计每分钟拉取消息数、发送成功数、失败数、消息端到端延迟，接入监控系统配置告警
• 增加死信队列逻辑：对连续3次发送Kafka失败的消息，自动路由到SQS关联的死信队列，避免异常消息无限阻塞拉取流程
• 增加SQS消息的元数据透传：把SQS的MessageId、发送时间戳写入Kafka消息头，方便下游系统排重和链路追踪

能力迭代阶段（吞吐量增长后优化） #

• 扩展拉取线程池：从单线程拉取扩展为可配置的多线程拉取，根据队列积压量动态调整并发度，支撑更高的消息吞吐量
• 增加消息转换能力：支持配置化的字段映射规则，从SQS消息中提取指定字段作为Kafka消息key，实现基于业务逻辑的Kafka分区路由
• 增加缓存层去重：把SQS的MessageId写入Redis短时间缓存，拦截重复投递的消息，实现近似恰好一次的投递语义

长期演进方向 #

• 扩展连接器适配能力：支持对接更多源端（RabbitMQ、HTTP接口、数据库变更）和目标端（Elasticsearch、数仓），形成统一的消息流转平台
• 增加动态配置能力：支持不重启应用的前提下，动态调整拉取并发度、开关连接器、修改目标Topic地址

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