Flume Spool Source 源码过程分析（未运行）

主要涉及到的类：

SpoolDirectorySource 读取用户配置，并按照batchSize去读取这么多量的Event从用户指定的Spooling Dir中。SpoolDirectorySource 不会去读取某一个具体的文件，而是通过内部的reader去读取。文件切换等操作，都是reader去实现

内部类：SpoolDirectoryRunnable是一个线程，其中的run方法，完成从Spooling Dir读取Event（使用reader去读取）

 

@Override
    public void run() {
      int backoffInterval = 250;
      try {
        while (!Thread.interrupted()) {
          List<Event> events = reader.readEvents(batchSize);
          if (events.isEmpty()) {
            break;
          }
          sourceCounter.addToEventReceivedCount(events.size());
          sourceCounter.incrementAppendBatchReceivedCount();

          try {
            getChannelProcessor().processEventBatch(events);
            reader.commit();
          } catch (ChannelException ex) {
            logger.warn("The channel is full, and cannot write data now. The " +
              "source will try again after " + String.valueOf(backoffInterval) +
              " milliseconds");
            hitChannelException = true;
            if (backoff) {
              TimeUnit.MILLISECONDS.sleep(backoffInterval);
              backoffInterval = backoffInterval << 1;
              backoffInterval = backoffInterval >= maxBackoff ? maxBackoff :
                                backoffInterval;
            }
            continue;
          }
          backoffInterval = 250;
          sourceCounter.addToEventAcceptedCount(events.size());
          sourceCounter.incrementAppendBatchAcceptedCount();
        }
        logger.info("Spooling Directory Source runner has shutdown.");
      } catch (Throwable t) {
        logger.error("FATAL: " + SpoolDirectorySource.this.toString() + ": " +
            "Uncaught exception in SpoolDirectorySource thread. " +
            "Restart or reconfigure Flume to continue processing.", t);
        hasFatalError = true;
        Throwables.propagate(t);
      }
    }

 

ReliableSpoolingFileEventReader 定义在SpoolDirectorySource中的reader。看这个名字就知道碉堡了，reliable的，怎么实现reliable的？？

reader的readEvent方法，会根据batchSize大小读取指定的Event

 

该方法的大致意思：

如果没有提交，如果当前文件空，错，否则获取EventDeserializer

如果已经提交，如果当前文件空，则获得下一个文件，之后，如果文件还是空，则返回空Event列表。

 

之后，调用EventDeserializer的readEvents。

 

 public List<Event> readEvents(int numEvents) throws IOException {
    if (!committed) {
      if (!currentFile.isPresent()) {
        throw new IllegalStateException("File should not roll when " +
            "commit is outstanding.");
      }
      logger.info("Last read was never committed - resetting mark position.");
      currentFile.get().getDeserializer().reset();
    } else {
      // Check if new files have arrived since last call
      if (!currentFile.isPresent()) {
        currentFile = getNextFile();
      }
      // Return empty list if no new files
      if (!currentFile.isPresent()) {
        return Collections.emptyList();
      }
    }

    EventDeserializer des = currentFile.get().getDeserializer();
    List<Event> events = des.readEvents(numEvents);

    /* It's possible that the last read took us just up to a file boundary.
     * If so, try to roll to the next file, if there is one. */
    if (events.isEmpty()) {
      retireCurrentFile();
      currentFile = getNextFile();
      if (!currentFile.isPresent()) {
        return Collections.emptyList();
      }
      events = currentFile.get().getDeserializer().readEvents(numEvents);
    }

    //写入header值，略47 
    committed = false;
    lastFileRead = currentFile;
    return events;
  }

 在这个方法中，我们看到了

currentFile：该对象采用了谷歌的Optional进行封装，更加容易判断空指针等等。Optional<FileInfo>，该FileInfo封装了普通的File对象和针对该file对象的EventDeserializer（事件序列器）

该currentFile主要在ReliableSpoolingFileEventReader 类中的Optional<FileInfo> openFile(File file)，Optional<FileInfo> getNextFile() 方法中调用。

EventDeserializer：事件序列器的主要作用在于定义一些读取的基本操作

其中mark是读取的行position进行标记

EventDeserializer的实现子类，很多，这里只讲LineDeserializer，顾名思义，按照行去读取，一行就是一个Event

虽然EventDeserializer已经涉及到读取行了，但是真正读取记录的还不是他。

我们看openfile函数中

String nextPath = file.getPath();
      PositionTracker tracker =
          DurablePositionTracker.getInstance(metaFile, nextPath);
      if (!tracker.getTarget().equals(nextPath)) {
        tracker.close();
        deleteMetaFile();
        tracker = DurablePositionTracker.getInstance(metaFile, nextPath);
      }
      ResettableInputStream in =
          new ResettableFileInputStream(file, tracker,
              ResettableFileInputStream.DEFAULT_BUF_SIZE, inputCharset,
              decodeErrorPolicy);
      EventDeserializer deserializer = EventDeserializerFactory.getInstance
          (deserializerType, deserializerContext, in);

因此可以看出EventDeserializer读取记录是靠 ResettableFileInputStream（in对象），ResettableFileInputStream的初始化需要File类和一个DurablePositionTracker，

因此，ResettableFileInputStream在读取File内容同时，使用DurablePositionTracker去记录position的信息。

DurablePositionTracker使用了apache avro来进行持久化

private final DataFileWriter<TransferStateFileMeta> writer;
private final DataFileReader<TransferStateFileMeta> reader;

这样，当我们使用EventDeserializer读取一个event的时候，就会从当前文件流中获取信息，同时也能够记录读取的位置信息。

当读取batchsize数量的event都正确处理后，ReliableSpoolingFileEventReader 会commit()，持久化位置信息

 public void commit() throws IOException {
    if (!committed && currentFile.isPresent()) {
      currentFile.get().getDeserializer().mark();
      committed = true;
    }
  }

这里的mark方法，调用

LineDeserializer的

 @Override
  public void mark() throws IOException {
    ensureOpen();
    in.mark();
  }

在调用ResettableFileInputStream（in）的mark方法

 @Override
  public void mark() throws IOException {
    tracker.storePosition(tell());
  }

之后调用位置tracker的storePostition方法（DurablePositionTracker）

 @Override
  public synchronized void storePosition(long position) throws IOException {
    metaCache.setOffset(position);
    writer.append(metaCache);
    writer.sync();
    writer.flush();
  }

之后，调用avro的DataFileWriter，完成写入操作。

最后，至于postition位置的持久化逻辑判断，基本也能猜到，当出现trash时候，从未读取的地方开始读取，等等，所以说，是ResettableFileInputStream的输入流，因为他能够读取信息，也能持久化读取的信息位置。