Kafka数据实战

发表于 | 更新于 | 分类于 | 阅读次数：
本文字数： 2.7k | 阅读时长 ≈ 2 分钟

Kafka保证数据不丢失

1.生产者数据的不丢失

kafka的ack机制：在kafka发送数据的时候，每次发送消息都会有一个确认反馈机制，确保消息正常的能够被收到，其中状态有0,1，-1。

  • 如果是同步模式：ack机制能够保证数据的不丢失，如果ack设置为0，风险很大，一般不建议设置为0。即使设置为1，也会随着leader宕机丢失数据。
1
2
producer.type=sync
request.required.acks=1
  • 如果是异步模式：也会考虑ack的状态，除此之外，异步模式下的有个buffer，通过buffer来进行控制数据的发送，有两个值来进行控制，时间阈值与消息的数量阈值，如果buffer满了数据还没有发送出去，有个选项是配置是否立即清空buffer。可以设置为-1，永久阻塞，也就数据不再生产。
  • 异步模式下，即使设置为-1。也可能因为程序员的不科学操作，操作数据丢失，比如kill -9，但这是特别的例外情况。
1
2
3
4
5
6
producer.type=async
request.required.acks=1
queue.buffering.max.ms=5000
queue.buffering.max.messages=10000
queue.enqueue.timeout.ms = -1
batch.num.messages=200

结论：producer有丢数据的可能，但是可以通过配置保证消息的不丢失。

2.消费者数据的不丢失

通过offset commit 来保证数据的不丢失，kafka自己记录了每次消费的offset数值，下次继续消费的时候，会接着上次的offset进行消费。

而offset的信息在kafka0.8版本之前保存在zookeeper中，在0.8版本之后保存到topic中，即使消费者在运行过程中挂掉了，再次启动的时候会找到offset的值，找到之前消费消息的位置，接着消费，由于offset的信息写入的时候并不是每条消息消费完成后都写入的，所以这种情况有可能会造成重复消费，但是不会丢失消息。

唯一例外的情况是，我们在程序中给原本做不同功能的两个consumer组设置KafkaSpoutConfig.bulider.setGroupid的时候设置成了一样的groupid，这种情况会导致这两个组共享同一份数据，就会产生组A消费partition1，partition2中的消息，组B消费partition3的消息，这样每个组消费的消息都会丢失，都是不完整的。 为了保证每个组都独享一份消息数据，groupid一定不要重复才行。

3.集群中的broker的数据不丢失

每个broker中的partition我们一般都会设置有replication（副本）的个数，生产者写入的时候首先根据分发策略（有partition按partition，有key按key，都没有轮询）写入到leader中，follower（副本）再跟leader同步数据，这样有了备份，也可以保证消息数据的不丢失。

配置列表

因为换取了更高的数据安全性,所以producer/consumer 吞吐量会下降

1. Producer端

  • block.on.buffer.full = true 尽管该参数在0.9.0.0已经被标记为“deprecated”，但鉴于它的含义非常直观，所以这里还是显式设置它为true，使得producer将一直等待缓冲区直至其变为可用。否则如果producer生产速度过快耗尽了缓冲区，producer将抛出异常
  • acks=all 很好理解，所有follower都响应了才认为消息提交成功，即"committed"
  • retries = MAX 无限重试，直到你意识到出现了问题:)
  • max.in.flight.requests.per.connection = 1 限制客户端在单个连接上能够发送的未响应请求的个数。设置此值是1表示kafka broker在响应请求之前client不能再向同一个broker发送请求。注意：设置此参数是为了避免消息乱序
  • 使用KafkaProducer.send(record, callback)而不是send(record)方法 自定义回调逻辑处理消息发送失败
  • callback逻辑中最好显式关闭producer：close(0) 注意：设置此参数是为了避免消息乱序
  • unclean.leader.election.enable=false 关闭unclean leader选举，即不允许非ISR中的副本被选举为leader，以避免数据丢失
  • replication.factor >= 3 这个完全是个人建议了，参考了Hadoop及业界通用的三备份原则
  • min.insync.replicas > 1 消息至少要被写入到这么多副本才算成功，也是提升数据持久性的一个参数。与acks配合使用
  • 保证replication.factor > min.insync.replicas 如果两者相等，当一个副本挂掉了分区也就没法正常工作了。通常设置replication.factor = min.insync.replicas + 1即可

2. Consumer端

  • enable.auto.commit=false 关闭自动提交位移, 在消息被完整处理之后再手动提交位移

Kafka保证消息的消费顺序

每次添加消息到 Partition(分区) 的时候都会采用尾加法, Kafka 只能为我们保证 同Partition(分区) 中的消息有序，而不能保证同 Topic(主题) 多 Partition(分区) 的有序。

消息在被追加到 Partition(分区)的时候都会分配一个特定的偏移量（offset）。Kafka 通过偏移量（offset）来保证消息在分区内的顺序性。

保证 Kafka 中消息消费的顺序，有了下面两种方法：

  1. 1 个 Topic 只对应一个 Partition。
  2. （推荐）发送消息的时候指定 key/Partition。

Kafka保证消息不重复消费

保证数据不重复消费, 有了下面两种方法：

（1）去重：将消息的唯一标识保存到外部介质中，每次消费处理时判断是否处理过。
（2）不管：大数据场景中，报表系统或者日志信息丢失几条都无所谓，不会影响最终的统计分析结果。

-------------本文结束-------------
0%