客户端 SDK 开发指南

# https://kafka.apache.org/documentation/#producerconfigs
# 批量大小，一次性最多积累 1MiB 的数据。
batch.size=1048576
# 攒批时间，对于 Producer，如果在指定的时间内没有达到攒批的上限，也会触发发送。这个时间代表了发送的最大延迟。
linger.ms=1000

# https://kafka.apache.org/21/documentation.html#producerconfigs
# 最大重试次数
retries=2147483647

# 初始退避延迟时间
retry.backoff.ms=100
# 最大退避延迟时间
retry.backoff.max.ms=1000
# 每次发送请求的 RPC 超时时间
request.timeout.ms=30000
# 整个发送调用整体超时时间，超过这个时间后即不再重试，返回调用侧失败异常。
delivery.timeout.ms=120000
# 批量大小，一次性最多积累 1MiB 的数据。
batch.size=1048576
# 攒批时间，对于 Producer，如果在指定的时间内没有达到攒批的上限，也会触发发送。这个时间代表了发送的最大延迟。
linger.ms=1000

package main

import (
	"time"

	"github.com/twmb/franz-go/pkg/kgo"
)

func main() {
	client, err := kgo.NewClient(
		kgo.SeedBrokers("localhost:9092"),
		// 对应 Kafka 配置中的 linger.ms，例如 1000 毫秒。
		kgo.ProducerLinger(1*time.Second),
	)
	if err != nil {
		panic(err)
	}
	defer client.Close()
}

// https://pkg.go.dev/github.com/segmentio/kafka-go#Writer
// 推荐配置示例：
w := &kafka.Writer{
	// 消息发送最大重试次数。
	// 参考 Apache Kafka Java Client 的 retries 参数，设置为 Integer.MAX_VALUE。
	MaxAttempts: 2147483647,
	// 触发一次发送的最大消息条数。
	BatchSize: 4096,
	// 需要的分区副本确认数。
	RequiredAcks: kafka.RequireAll,
}

package main

import (
	"time"

	"github.com/IBM/sarama"
)

func main() {
	// https://pkg.go.dev/github.com/IBM/sarama#NewConfig
	config := sarama.NewConfig()

	// 触发一次 flush 的字节数阈值。
	config.Producer.Flush.Bytes = 1048576
	// 触发一次 flush 的消息条数阈值。
	config.Producer.Flush.Messages = 4096
	// flush 的最大等待时间，等同于 Kafka Java Producer 的 linger.ms。
	config.Producer.Flush.Frequency = 1 * time.Second

	// 消息发送最大重试次数（默认 3），等同于 Kafka Java Producer 的 retries。
	config.Producer.Retry.Max = 2147483647
	// 重试之间的退避等待时间（默认 100ms），等同于 Kafka Java Producer 的 retry.backoff.ms。
	config.Producer.Retry.Backoff = 1 * time.Second

	// TODO: 使用 config 创建 producer，例如 sarama.NewSyncProducer(brokers, config)
	_ = config
}

# https://www.elastic.co/docs/reference/beats/filebeat/kafka-output
# Leader 选举期间的重试退避等待时间。
retry.backoff: 3s
# 批量发送 Kafka 请求的等待时间。
bulk_flush_frequency: 1s