flume MemoryChannel 源码解析


1.先分析三个LinkedBlockingDeque<Event>类型的takeList,putList,queue

putList:  存放的是来自source生产的数据,通过调用doPut(Event event)方法,它是如何到queue的,在每次执行doCommit的时候，会循环放到queue，其实doCommit()放法只做了putlist交给queue,

      synchronized(queueLock) {
        if(puts > 0 ) {
          while(!putList.isEmpty()) {
            if(!queue.offer(putList.removeFirst())) {
              throw new RuntimeException("Queue add failed, this shouldn't be able to happen");
            }
          }
        }
        putList.clear();
        takeList.clear();
      }

takelist:    每次sink消费，都会加到takelist,一般不起什么作用，但是操作失败，rollback就起作用了

protected void doRollback() {

     int takes = takeList.size();
      synchronized(queueLock) {
        Preconditions.checkState(queue.remainingCapacity() >= takeList.size(), "Not enough space in memory channel " +
            "queue to rollback takes. This should never happen, please report");
        while(!takeList.isEmpty()) {
          queue.addFirst(takeList.removeLast());
        }
        putList.clear();
      }
      bytesRemaining.release(putByteCounter);
      putByteCounter = 0;
      takeByteCounter = 0;
      queueStored.release(takes);
      channelCounter.setChannelSize(queue.size());
    }
  }

queue:存放的即将传递给sink的所有数据.


2.参数对应关系

transactionCapacity：设置takelist,putlist最大容量

capacity：           设置queue的最大容量

keep-alive：          Semaphore的tryAcquire的timeout的参数，增加到takelist,putlist,queue超时时间

相关内容