从KinesisProducerLibrary迁移到KinesisAsyncClient后包装类延迟升高求助

你遇到的核心问题是默认配置的Kinesis Async Client在高负载下存在客户端侧的请求排队瓶颈，导致包装类函数延迟升高；而Kinesis服务端的发布-接收延迟改善，是因为高流量下服务端资源被更充分利用。下面是针对性的优化方案：

1. 调整maxConcurrency参数（核心优化点） #

默认的Kinesis Async ClientmaxConcurrency值较低（通常为50），高流量下会导致请求在客户端内部排队，直接拉高包装类函数的延迟。这个参数控制客户端同时处理的异步请求数量，调高它可以提升并发处理能力，减少排队。

修改配置代码：

@Provides
@Singleton
fun kinesisAsyncClient(): KinesisAsyncClient {
    return KinesisAsyncClient.builder()
        .region(software.amazon.awssdk.regions.Region.of(getAwsRegion()))
        .asyncConfiguration { config ->
            config.maxConcurrency(200) // 根据实际流量调整，建议从200开始测试
        }
        .build()
}

2. 替换单条putRecord为批量请求 #

KPL的核心优势之一是自动批量处理请求，而你当前使用的单条putRecord会产生大量独立的网络请求，增大客户端开销。改用putRecords批量提交请求，能显著降低客户端侧的处理延迟和网络开销：

修改使用代码示例：

try {
    // 假设已收集一批eventRecord
    val records = batchEventRecords.map { eventRecord ->
        val eventBytes = JSON.writeValueAsBytes(eventRecord)
        PutRecordsRequestEntry.builder()
            .partitionKey(eventRecord.entityId)
            .data(SdkBytes.fromByteArray(eventBytes))
            .build()
    }

    val request = PutRecordsRequest.builder()
        .streamName(streamName)
        .records(records)
        .build()

    kinesisAsyncClient.putRecords(request)
} catch (e: Exception) { 
    // 异常处理逻辑
}

3. 优化HTTP客户端连接池 #

默认的Netty HTTP连接池大小可能无法支撑高并发请求，调整连接池参数可以减少连接建立的开销：

@Provides
@Singleton
fun kinesisAsyncClient(): KinesisAsyncClient {
    return KinesisAsyncClient.builder()
        .region(software.amazon.awssdk.regions.Region.of(getAwsRegion()))
        .asyncConfiguration { asyncConfig ->
            asyncConfig.maxConcurrency(200)
        }
        .httpClientConfiguration { httpConfig ->
            httpConfig.maxConnections(100) // 调整最大连接数
                .connectionTimeout(Duration.ofSeconds(3))
                .socketTimeout(Duration.ofSeconds(10))
        }
        .build()
}

4. 调整重试策略 #

默认的重试策略可能在遇到临时错误时无限制重试，导致请求堆积。可以调整重试次数和退避策略，避免不必要的延迟：

@Provides
@Singleton
fun kinesisAsyncClient(): KinesisAsyncClient {
    val retryPolicy = RetryPolicy.builder()
        .numRetries(3)
        .backoffStrategy(BackoffStrategies.exponentialBackoff(Duration.ofMillis(100), Duration.ofSeconds(2)))
        .build()

    return KinesisAsyncClient.builder()
        .region(software.amazon.awssdk.regions.Region.of(getAwsRegion()))
        .asyncConfiguration { asyncConfig ->
            asyncConfig.maxConcurrency(200)
                .retryPolicy(retryPolicy)
        }
        .build()
}

总结 #

• maxConcurrency是解决你当前延迟问题的关键配置，调高它能直接减少客户端侧的请求排队。
• 批量请求是对齐KPL性能的核心手段，务必替换单条putRecord。
• 连接池和重试策略的优化可以进一步提升高负载下的稳定性。

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