本文介绍一种健壮、符合 aws 最佳实践的方式：先调用 describestream 检查流是否存在及状态，若不存在则创建，并轮询等待其变为 active 状态，最后安全写入数据。避免依赖错误码或字符串匹配，提升代码可维护性与可靠性。
本文介绍一种健壮、符合 aws 最佳实践的方式：先调用 describestream 检查流是否存在及状态，若不存在则创建，并轮询等待其变为 active 状态，最后安全写入数据。避免依赖错误码或字符串匹配，提升代码可维护性与可靠性。
在 AWS Kinesis 中，判断一个命名流（stream）是否已存在，不应依赖 CreateStream 的异常响应（如 HTTP 400）进行推断——因为该错误可能由多种原因触发（如权限不足、配额超限、参数非法），且错误消息格式不保证向后兼容。更可靠、语义明确的方式是使用 DescribeStream API：它会直接返回流的元数据和当前状态（如 CREATING、ACTIVE、DELETING 等），若流不存在，则返回标准的 ResourceNotFoundException 错误。
以下是推荐的 Go 实现流程（基于 AWS SDK for Go v1）：
- 尝试描述流：调用 DescribeStream；
-
处理结果：
- 若成功返回且 StreamDescription.StreamStatus == "ACTIVE" → 直接写入；
- 若返回 ResourceNotFoundException → 调用 CreateStream 创建；
- 若返回 StreamStatus == "CREATING" → 启动轮询（polling），等待其变为 ACTIVE；
- 轮询机制：使用 DescribeStream 定期重试（建议间隔 2–5 秒），配合最大重试次数或超时控制，防止无限等待；
- 写入数据：仅当流处于 ACTIVE 状态后，才执行 PutRecord 或 PutRecords。
import (
"log"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/awserr"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/kinesis"
)
func ensureStreamActive(streamName string, shardCount int64) error {
sess := session.Must(session.NewSession())
k := kinesis.New(sess)
// Step 1: Describe stream
descInput := &kinesis.DescribeStreamInput{
StreamName: aws.String(streamName),
}
descResult, err := k.DescribeStream(descInput)
if err != nil {
if aerr, ok := err.(awserr.Error); ok && aerr.Code() == kinesis.ErrCodeResourceNotFoundException {
// Stream does not exist — create it
log.Printf("Stream %s not found, creating...", streamName)
_, err = k.CreateStream(&kinesis.CreateStreamInput{
StreamName: aws.String(streamName),
ShardCount: aws.Int64(shardCount),
})
if err != nil {
return err
}
// Fall through to polling
} else {
return err // Other errors (e.g., permission denied) are fatal
}
} else {
// Stream exists; check status
status := *descResult.StreamDescription.StreamStatus
if status == "ACTIVE" {
log.Printf("Stream %s is already ACTIVE", streamName)
return nil
}
if status != "CREATING" {
return fmt.Errorf("unexpected stream status: %s", status)
}
log.Printf("Stream %s is CREATING, waiting for ACTIVE...", streamName)
}
// Step 2: Poll until ACTIVE (with timeout)
const maxWait = 5 * time.Minute
const pollInterval = 3 * time.Second
deadline := time.Now().Add(maxWait)
for time.Now().Before(deadline) {
descResult, err := k.DescribeStream(descInput)
if err != nil {
if aerr, ok := err.(awserr.Error); ok && aerr.Code() == kinesis.ErrCodeResourceNotFoundException {
// Should not happen after CreateStream succeeded, but handle gracefully
time.Sleep(pollInterval)
continue
}
return err
}
status := *descResult.StreamDescription.StreamStatus
if status == "ACTIVE" {
log.Printf("Stream %s is now ACTIVE", streamName)
return nil
}
if status == "CREATING" || status == "UPDATING" {
time.Sleep(pollInterval)
continue
}
return fmt.Errorf("stream %s entered terminal state: %s", streamName, status)
}
return fmt.Errorf("timeout waiting for stream %s to become ACTIVE", streamName)
}
// Usage example
func main() {
streamName := "my-kinesis-stream"
if err := ensureStreamActive(streamName, 2); err != nil {
log.Fatal("Failed to ensure stream:", err)
}
// Now safely write
k := kinesis.New(session.Must(session.NewSession()))
_, err := k.PutRecord(&kinesis.PutRecordInput{
StreamName: aws.String(streamName),
Data: []byte("Hello Kinesis"),
PartitionKey: aws.String("pk-001"),
})
if err != nil {
log.Fatal("Failed to write record:", err)
}
}
⚠️ 注意事项：
- DescribeStream 是低开销、高可用的只读操作，比 ListStreams（需分页遍历）更高效，尤其适用于单流检查场景；
- 创建流后必须等待 ACTIVE 状态——Kinesis 不允许向 CREATING 或 UPDATING 状态的流写入，否则将返回 ResourceInUseException；
- 生产环境建议封装轮询逻辑为可配置的工具函数（支持自定义超时、重试策略、上下文取消）；
- 若需更高并发或批量管理多流，可结合 ListStreams + 并行 DescribeStream，但单流场景下 DescribeStream 始终是首选。
通过该方案，你获得的是确定性、可测试、可监控的流初始化流程，完全规避了“错误驱动逻辑”的反模式，也与 AWS 官方文档推荐方式一致。
