# 创建事务生产者配置
$ cat > /tmp/transaction-producer.properties << 'EOF' bootstrap.servers=192.168.1.51:9092,192.168.1.52:9092,192.168.1.53:9092 acks=all enable.idempotence=true transactional.id=fgedu-tx-producer-1 transaction.timeout.ms=900000 key.serializer=org.apache.kafka.common.serialization.StringSerializer value.serializer=org.apache.kafka.common.serialization.StringSerializer EOF # 事务消费者配置 $ cat > /tmp/transaction-consumer.properties << 'EOF' bootstrap.servers=192.168.1.51:9092,192.168.1.52:9092,192.168.1.53:9092 group.id=fgedu-tx-consumer-group isolation.level=read_committed enable.auto.commit=false auto.offset.reset=earliest key.deserializer=org.apache.kafka.common.serialization.StringDeserializer value.deserializer=org.apache.kafka.common.serialization.StringDeserializer EOF

# 事务操作流程（Java代码示例）

// 1. 初始化事务
producer.initTransactions();

try {
// 2. 开始事务
producer.beginTransaction();

// 3. 发送消息到多个分区
producer.send(new ProducerRecord<>(“fgedu-orders”, “key1”, “value1”));
producer.send(new ProducerRecord<>(“fgedu-orders”, “key2”, “value2”));
producer.send(new ProducerRecord<>(“fgedu-payments”, “key1”, “value1”));

// 4. 发送Offset到事务（消费-处理-生产模式）
producer.sendOffsetsToTransaction(
Collections.singletonMap(
new TopicPartition(“fgedu-input”, 0),
new OffsetAndMetadata(1000)
),
“fgedu-consumer-group”
);

// 5. 提交事务
producer.commitTransaction();
} catch (Exception e) {
// 6. 回滚事务
producer.abortTransaction();
}

# 查看事务状态
$ /bigdata/app/kafka/bin/kafka-transactions.sh \
–bootstrap-server 192.168.1.51:9092 \
–list

TransactionalId: fgedu-tx-producer-1
State: Complete
ProducerId: 1001

# 数据丢失常见原因

1. 生产者配置问题
– acks=0或acks=1
– 未启用重试
– 缓冲区满时丢弃消息

2. Broker配置问题
– min.insync.replicas=1
– unclean.leader.election.enable=true
– 副本数不足

3. 消费者配置问题
– 自动提交Offset但处理失败
– 消费异常后跳过消息

# 排查步骤
# 1. 检查生产者配置
$ grep -E “acks|retries|enable.idempotence” /tmp/producer.properties

acks=all
retries=3
enable.idempotence=true

# 2. 检查Broker配置
$ grep -E “min.insync.replicas|unclean.leader.election” /bigdata/app/kafka/config/server.properties

min.insync.replicas=2
unclean.leader.election.enable=false

# 3. 检查消费者配置
$ grep -E “enable.auto.commit” /tmp/consumer.properties

enable.auto.commit=false

# 生产者预防方案
# 配置文件
acks=all
retries=3
enable.idempotence=true
delivery.timeout.ms=120000

# 添加回调处理
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception != null) {
// 记录失败消息
// 重试或存储到死信队列
}
}
});

# Broker预防方案
min.insync.replicas=2
unclean.leader.election.enable=false
replica.lag.time.max.ms=30000

# 消费者预防方案
enable.auto.commit=false
# 处理成功后再提交Offset
consumer.commitSync();

# 数据重复常见原因

1. 生产者重试
– 网络超时后重试
– Broker已写入但未返回确认

2. 消费者重复消费
– Offset提交失败
– 消费者Rebalance
– 处理成功但Offset未提交

# 解决方案

# 1. 生产者启用幂等性
enable.idempotence=true

# 2. 消费者实现幂等性消费
# 方案A：使用唯一Key去重
# 方案B：使用数据库唯一约束
# 方案C：使用Redis去重

# 消费者幂等性示例
public void consume(ConsumerRecord record) {
String key = record.key();
String value = record.value();

// 检查是否已处理
if (redis.get(“processed:” + key) != null) {
return; // 跳过已处理的消息
}

// 处理消息
processMessage(value);

// 标记为已处理
redis.set(“processed:” + key, “1”);

// 提交Offset
consumer.commitSync();
}

# 数据顺序保证方案

# 1. 单分区顺序
# 相同Key的消息发送到同一分区
# 分区内消息有序

# 2. 生产者配置
enable.idempotence=true
max.in.flight.requests.per.connection=5 # 幂等性启用后可为5

# 3. 如果未启用幂等性
max.in.flight.requests.per.connection=1 # 必须为1

# 4. 消费者顺序消费
# 单线程消费或使用分区锁

# 顺序消费示例
public void consume(ConsumerRecords records) {
for (ConsumerRecord record : records) {
String key = record.key();

// 按Key加锁，保证相同Key顺序处理
synchronized (getLock(key)) {
processMessage(record.value());
}
}

// 批量提交Offset
consumer.commitSync();
}

# 生产者最佳实践
1. 启用acks=all
2. 启用enable.idempotence=true
3. 设置合理的retries
4. 实现发送回调处理
5. 使用事务保证跨分区原子性

# Broker最佳实践
1. min.insync.replicas=2
2. unclean.leader.election.enable=false
3. replication.factor=3
4. 监控ISR状态

# 消费者最佳实践
1. enable.auto.commit=false
2. 处理成功后再提交Offset
3. 实现幂等性消费
4. 合理设置max.poll.records