kafka无法收到flume采集的数据的解决办法

柔光的暖阳◎ 2022-05-19 07:43 1036阅读 0赞

# 问题重现 #

在写黑名单那篇博文的时候，我是通过直接copy log日志文件到监控目录下的方式来模拟数据的，在前几次模拟访问日志文件的时候挺正常的，copy进去基本都是秒采集（文件显示直接加了.COMPLETED后缀）。

但到后来再往采集目录下copy log日志文件的时候，待采集目录下的文件并不会显示被采集（文件没有.COMPLETED后缀），kafka也一直收不到flume采集来的数据。但重启flume客户端后又会正常采集。经过多方排查后，最终定位问题在flume配置文件有错，之前flume配置文件如下：

flume2kafka.sources=r3
    flume2kafka.sinks=k3
    flume2kafka.channels=c3
    
    # 配置source
    flume2kafka.sources.r3.type = spooldir
    flume2kafka.sources.r3.channels = c3
    flume2kafka.sources.r3.spoolDir = /home/jbw/log/
    flume2kafka.sources.r3.fileHeader = true
    
    # 配置channel，将buffer事件放在内存中
    flume2kafka.channels.c3.type = memory
    flume2kafka.channels.c3.capacity = 100000
    flume2kafka.channels.c3.transactionCapacity = 1000
    
    # 配置sink
    flume2kafka.sinks.k3.type=org.apache.flume.sink.kafka.KafkaSink
    flume2kafka.sinks.k3.brokerList=172.18.9.119:9092,172.18.9.120:9092,172.18.9.121:9092
    flume2kafka.sinks.k3.topic=peopleVisitTopic
    flume2kafka.sinks.k3.requiredAcks = 0
    flume2kafka.sinks.k3.batchSize = 20
    #flume2kafka.sinks.k3.channel = memcnl
    
    # 把source和sink绑定在channel上
    flume2kafka.sources.r3.channels=c3

# 解决方案 #

检查source和channel部分配置都没问题，后来发现是sinks.k3.requiredAcks这一参数导致的问题，之前是0，现在改为1后

配置文件如下，重新运行flume客户端，发现kafka消费端会收到数据了，问题解决。

flume2kafka.sources=r3
    flume2kafka.sinks=k3
    flume2kafka.channels=c3
    
    # 配置source
    flume2kafka.sources.r3.type = spooldir
    flume2kafka.sources.r3.spoolDir = /home/jbw/log/
    flume2kafka.sources.r3.fileHeader = true
    
    # 配置channel，将buffer事件放在内存中
    flume2kafka.channels.c3.type = memory
    flume2kafka.channels.c3.capacity = 10000
    flume2kafka.channels.c3.transactionCapacity = 1000
    
    # 配置sink
    flume2kafka.sinks.k3.type=org.apache.flume.sink.kafka.KafkaSink
    flume2kafka.sinks.k3.brokerList=172.18.9.119:9092,172.18.9.120:9092,172.18.9.121:9092
    flume2kafka.sinks.k3.topic=peopleVisitTopic
    flume2kafka.sinks.k3.serializer.class=kafka.serializer.StringEncoder
    flume2kafka.sinks.k3.requiredAcks = 1
    flume2kafka.sinks.k3.batchSize = 20
    
    # 把source和sink绑定在channel上
    flume2kafka.sources.r3.channels=c3
    flume2kafka.sinks.k3.channel=c3

# 原因分析   #

我们先看下 request.required.acks 这个参数到底是什么玩意。

它其实是配置 kafka 中的 ack确认机制的参数，kafka 中有三种确认机制，如下：

*   机制一：producer 端不等待来自 broker 的确认消息，直接发送下一条消息。
 *   机制二：producer 端会在得到 leader broker 确认收到数据消息后，才发送下一条消息。
 *   机制三：producer 端会在得到所有 follower broker 副本确认收到数据消息后，才会发送下一条数据。

现在大家应该就明白 requiredAcks 参数的意义了，**它就是制定需要多少副本确认收到消息才会被认定为成功写入。**不难发现，三种机制，kafka 性能依次递减 (producer吞吐量降低)，但数据安全性依次递增。

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>request.required.acks</td> 
   <td>意义</td> 
  </tr> 
  <tr> 
   <td>0</td> 
   <td><span style="color:#444444;">代表不需要等待任何 broker 的确认，对应于机制一。</span></td> 
  </tr> 
  <tr> 
   <td>1</td> 
   <td><span style="color:#444444;">代表仅需要角色为 leader 的 broker 确认收到消息即可，对应于机制二。</span></td> 
  </tr> 
  <tr> 
   <td>-1</td> 
   <td><span style="color:#444444;">代表需要所有</span> follower broker 副本都需要确认收到消息才行，对应于机制三。</td> 
  </tr> 
 </tbody> 
</table>

我们之前遇到的问题，可能是 kafka 的 leader broker 挂掉了，但由于 request.required.acks 参数指定的是0，说明 producer 端不需要经过broker端的确认就可以发下一条数据，然而此时我们 broker 已经挂了，你再写就没用了。故在保持性能的情况下，将此参数改为1即可。