Kafka-分区日志文件的清理原理
清理
一般情况下,kafka会根据设置的时间保留数据,把熬过时效的旧数据删除掉。
早于保留时间的旧事件会被删除,为每个键保留最新的值,从而达到清理的效果。只有当应用程序生成的事件里包含了键值对时,为这些主题设置compact策略才有意义。如果主题包含null键,清理就有失败。
清理的工作原理
每个日志片段可以分为以下两个部分
干净的部分:这些消息之前被清理过,每个键只有一个对应的值,这个值是上一次清理时保留下来的。
污浊的部分:这些消息时上一次清理之后写入的。
如果kafka启动时启用了清理功能(log.cleaner.enabled参数),每个broker会启动一个清理管理器线程和多个清理线程,它们负责执行清理任务。这些线程会选择污浊率(污浊消息占分区总大小的比例)较高的分区进行清理。
为了清理分区,清理线程会读取分区的污浊部分,并在内存里创建一个map。map里的每个元素包含了消息键的散列值和消息的偏移量,键的散列值是16B,加上偏移量总共是24B。如果要清理一个1GB的日志片段,并假设每个消息的大小为1KB,那么这个片段就包含一百万个消息,而我们只需要24MB的map就可以清理这个片段。如果有重复的键,可以重用散列项,从而使用更少的内存。这是非常高效的。
kafka配置中可以对map使用的内存大小进行配置。每个线程都有自己的map,而这个参数指的是所有线程可使用的内存总大小。如果为map分配了1GB内存,并使用了5个清理线程,那么每个线程可以使用200MB内存来创建自己的map。kafka并不要求分区的整个污浊部分来适应这个map的大小,但要求至少有一个完整的片段必须符合。如果不符合,那么kafka就会报错,要么配置分配更多的内存,要么减少清理线程数量,让每个线程可用内存变大。如果只有少部分片段可以完全符合,kafka将从最旧的片段开始清理,等待下一次清理剩余的部分。
清理线程在创建好偏移量map后,开始从干净的片段处读取消息,从最旧的消息开始,把它们的内容与map里的内容进行比对。它会检查消息的键是否存在于map中,如果不存在,那么说明消息的值是最新的,就把消息复制到替换片段上。如果键已存在,消息会被忽略,因为在分区的后部已经有一个具有相同键的消息存在。在复制完所有消息之后,我们就将替换片段与原始片段进行交换,然后开始清理下一个片段。完成整个清理过程之后,每个键对应一个不同的消息,这些消息的值都是最新的。