解决 flume KafkaSink 启动后cpu占用100%的问题

解决 flume KafkaSink 启动后cpu占用100%的问题

 

Flume 版本 ：1.6.0-cdh5.5.0

问题描述：

配置kafkasink，将实时数据发送到kafka。

Flume启动完成后，没有日志处理时，cpu使用率飙升到100%

当有日志数据处理时，并发稳定时，cpu不定时会有一瞬间飙升。

当日志数据量比较大时，cpu不会飙升。

发现：

使用 jstack -F <pid> > /home/name/flume-dump.log命令，查看flume的堆栈信息

发现很多BLOCKED信息如下：

Thread 16599: (state = BLOCKED)

 - java.util.concurrent.locks.ReentrantReadWriteLock$FairSync.readerShouldBlock() @bci=1, line=695 (Compiled frame; information may be imprecise)

 - java.util.concurrent.locks.ReentrantReadWriteLock$Sync.tryAcquireShared(int) @bci=33, line=470 (Compiled frame)

 - java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireShared(int) @bci=2, line=1282 (Compiled frame)

 - java.util.concurrent.locks.ReentrantReadWriteLock$ReadLock.lock() @bci=5, line=727 (Compiled frame)

 - org.apache.flume.channel.file.Log.lockShared() @bci=4, line=785 (Compiled frame)

 - org.apache.flume.channel.file.FileChannel$FileBackedTransaction.doTake() @bci=72, line=501 (Compiled frame)

 - org.apache.flume.channel.BasicTransactionSemantics.take() @bci=51, line=113 (Compiled frame)

 - org.apache.flume.channel.BasicChannelSemantics.take() @bci=26, line=95 (Compiled frame)

 - org.apache.flume.sink.kafka.KafkaSink.process() @bci=57, line=97 (Compiled frame)

 - org.apache.flume.sink.DefaultSinkProcessor.process() @bci=4, line=68 (Compiled frame)

 - org.apache.flume.SinkRunner$PollingRunner.run() @bci=24, line=147 (Compiled frame)

 - java.lang.Thread.run() @bci=11, line=745 (Interpreted frame)

 

分析：

颜色选中部分可以看到线程的执行过程，走了哪些类，哪些方法。

那就去看一下源码吧。

下载flume-ng-1.6.0-cdh5.5.0-src.tar.gz flume的源码包

使用intellj idea打开。

Ctrl+n 搜索第一个类org.apache.flume.SinkRunner

点击 scroll from source

 

定位到这个类，发现是在flume-ng-core包中

接着看这个类中的内部类PollingRunner的run方法

 

这个类执行policy.process，policy的类型时SinkProcessor

，是所有sink的顶层接口，用来执行所有sink的process方法。

While循环中根据process方法返回的sink Status，判断当前channel中的event是否处理完毕。

如果处理完了，当前sink sleep。

如果这个sink没有返回BACKOFF，会一直while死循环的执行policy.process()方法。并判断其返回的状态。

Cpu占用率100%极有可能时死循环造成的，带着这个猜想，我看了一下process的实现类。

DefaultSinkProcessor类的process方法如下

 

那sink时哪个呢？上面的栈信息提示是KafkaSink。来看这个sink的process方法：

@Override
public Status process() throws EventDeliveryException {

// 一开始设置了status为ready
  Status result = Status.READY;
  Channel channel = getChannel();
  Transaction transaction = null;
  Event event = null;
  String eventTopic = null;
  String eventKey = null;

  try {
    long processedEvents = 0;

    transaction = channel.getTransaction();
    transaction.begin();

    messageList.clear();
    for (; processedEvents < batchSize; processedEvents += 1) {

// 从channel获取event
      event = channel.take();

      if (event == null) {
        // no events available in channel

// 我们flume一启动直接飙升，当前肯定是没有日志的。所以执行到这里 // for退出
        break;
      }

      byte[] eventBody = event.getBody();
      Map<String, String> headers = event.getHeaders();

      if ((eventTopic = headers.get(TOPIC_HDR)) == null) {
        eventTopic = topic;
      }

      eventKey = headers.get(KEY_HDR);

      if (logger.isDebugEnabled()) {
        logger.debug("{Event} " + eventTopic + " : " + eventKey + " : "
          + new String(eventBody, "UTF-8"));
        logger.debug("event #{}", processedEvents);
      }

      // create a message and add to buffer
      KeyedMessage<String, byte[]> data = new KeyedMessage<String, byte[]>
        (eventTopic, eventKey, eventBody);
      messageList.add(data);

    }
    //  break后到这里。Event是null，而且我们并没有日志写进来，处理的event

    // 数一定是0，下面的if也不会进
    // publish batch and commit.
    if (processedEvents > 0) {
      long startTime = System.nanoTime();
      producer.send(messageList);
      long endTime = System.nanoTime();
      counter.addToKafkaEventSendTimer((endTime-startTime)/(1000*1000));
      counter.addToEventDrainSuccessCount(Long.valueOf(messageList.size()));
    }
   // 出来if直接事务提交了。肯定也是null，到目前位置没有看到status的改变
    transaction.commit();

  } catch (Exception ex) {
    String errorMsg = "Failed to publish events";
    logger.error("Failed to publish events", ex);
    result = Status.BACKOFF;
    if (transaction != null) {
      try {
        transaction.rollback();
        counter.incrementRollbackCount();
      } catch (Exception e) {
        logger.error("Transaction rollback failed", e);
        throw Throwables.propagate(e);
      }
    }
    throw new EventDeliveryException(errorMsg, ex);
  } finally {
    if (transaction != null) {
      transaction.close();
    }
  }
//  这里竟然直接返回了。。Channel是空的时候返回了。除了判断什么都没做！！！
  return result;
}

 

可以看到，KafkaSink的process的实现，除了第一行代码设置了status为reday。之后就没有对状态进行改变，而PollingRunner的run方法是根据这个status判断当前sink是否需要sleep。Channel中没有event需要处理，当然要sleep啊，不然就是死循环了。只有在有数据的时候，处理数据才不会对cpu造成太大的压力。

这就解释了开头说的数据量打的时候cpu占用并不会太高的原因。

为了再确认一下，这个思路是不是正确的，再看一下flume实现的其他sink。再event为null的时候是怎么处理的。

HDFSEventSink的process方法如下

 

在commit后，判断了txnEventCount(for循环的计数器，循环一次说明处理了一个event)数。如果小于1（表示没有event），返回了BACKOFF。

IrcSink的处理方式：

可以看到这几个sink再event为null的时候，都将status设置为了BACKOFF

 

解决方法：

这个问题再flume的1.7版本中已经解决了。

1.7中的KafkaSink是这样做：

if (event == null) {
    // no events available in channel
    if(processedEvents == 0) {
        result = Status.BACKOFF;
        counter.incrementBatchEmptyCount();
    } else {
        counter.incrementBatchUnderflowCount();
    }
    break;
}