private void flushOrSync(boolean isSync, EnumSet<SyncFlag> syncFlags)
protected synchronized void writeChunk(byte[] b, int offset, int len, byte[] checksum) 
public synchronized void close()

implements Syncable, CanSetDropBehind {
	private final LinkedList<Packet> dataQueue = new LinkedList<Packet>(); // dataQueue是数据队列，用于保存等待发送给datanode的数据包
	private final LinkedList<Packet> ackQueue = new LinkedList<Packet>(); // ackQueue是确认队列，保存还没有被datanode确认接收的数据包
	private DataStreamer streamer;  // streamer线程，不停的从dataQueue中取出数据包，发送给datanode
	...
}

class DataStreamer extends Daemon {
	private volatile boolean streamerClosed = false;
	private ExtendedBlock block; // its length is number of bytes acked
	private Token<BlockTokenIdentifier> accessToken;
	private DataOutputStream blockStream; // socket的输出流(client->datanode)，用于将数据传输给datanode
	private DataInputStream blockReplyStream; // socket的输入流(datanode->client)，用户收到datanode的确认包
	private ResponseProcessor response = null; // response线程，用于接收从datanode返回的反馈信息
	...
}

synchronized (dataQueue) {
  // move packet from dataQueue to ackQueue
  if (!one.isHeartbeatPacket()) {
    dataQueue.removeFirst();
    ackQueue.addLast(one);
    dataQueue.notifyAll();
  }
}
// write out data to remote datanode
try {
  one.writeTo(blockStream);
  blockStream.flush();   
}

还需要一个线程接受ack，为ResponseProcessor
//获取返回的序号
ack.readFields(blockReplyStream);
long seqno = ack.getSeqno();
//获取dataQueue里第一个包的序号
Packet one;
synchronized (dataQueue) {
  one = ackQueue.getFirst();
}
if (one.seqno != seqno) {
  throw new IOException("ResponseProcessor: Expecting seqno " +
                        " for block " + block +
                        one.seqno + " but received " + seqno);
}
//移除ackQueuede里的第一个，
synchronized (dataQueue) {
  lastAckedSeqno = seqno; //lastAckedSeqno在waitForAckedSeqno有使用
  ackQueue.removeFirst();
  dataQueue.notifyAll();
}

long org.apache.hadoop.hbase.regionserver.StoreFlusher.performFlush(InternalScanner scanner, CellSink sink, long smallestReadPoint) throws IOException

public void append(final KeyValue kv) throws IOException {
  //如下两行主要是来加入到bf中的，下面的这个就是我们经常说的bf索引。
  appendGeneralBloomfilter(kv);
  appendDeleteFamilyBloomFilter(kv);
  //这行是重点
  writer.append(kv);
  //这行先不管，处理时间戳
  trackTimestamps(kv);
}

//检查key是否有问题，是否按顺序（memstore使用ConcurrentSkipListMap存储，应该不会有此问题）。
//并且返回key是否重复
boolean dupKey = checkKey(key, koffset, klength);
checkValue(value, voffset, vlength);
//如果不重复，则不检查边界，答案不能，因为如果有重复，不会检查边界更不会新建一个block。***问题5***
if (!dupKey) {
  //此出会检查block的大小，并且有一处需要注意，在里面的代码中有一些记录block信息的，这个以后会有用。
  //此处会写出chunk，处理readyChunks
  checkBlockBoundary();
}

byte[] indexKey = comparator.calcIndexKey(lastKeyOfPreviousBlock, firstKeyInBlock);
dataBlockIndexWriter.addEntry(indexKey, lastDataBlockOffset, onDiskSize);

//在此代码中会判断key的个数，如果key的个数达到了一定程度就新建一个chunk，放入readyChunks（这个会在checkBlockBoundary中处理），此出会写bf。***问题2***
enqueueReadyChunk(false);
... 这种是处理chunk被写出的时候的操作。重置一些值 ...
//真正的添加到bf中
chunk.add(bloomKey, keyOffset, keyLength);

ReadyChunk readyChunk = new ReadyChunk();
readyChunk.chunkId = numChunks - 1;
readyChunk.chunk = chunk;
readyChunk.firstKey = firstKeyInChunk; 
readyChunks.add(readyChunk);

//此处组织了block，将加入到此HFile的chunk生成树的结构。
long rootIndexOffset = dataBlockIndexWriter.writeIndexBlocks(outputStream);

private void writeInlineBlocks(boolean closing) throws IOException {
  //inlineBlockWriters 应该就3个，两个bf和一个block（待确定）
  for (InlineBlockWriter ibw : inlineBlockWriters) {
    while (ibw.shouldWriteBlock(closing)) {
      long offset = outputStream.getPos();
      boolean cacheThisBlock = ibw.getCacheOnWrite();
      ibw.writeInlineBlock(fsBlockWriter.startWriting(
          ibw.getInlineBlockType()));
      fsBlockWriter.writeHeaderAndData(outputStream);
      //此处添加leaf index block
      ibw.blockWritten(offset, fsBlockWriter.getOnDiskSizeWithHeader(),
          fsBlockWriter.getUncompressedSizeWithoutHeader());
      totalUncompressedBytes += fsBlockWriter.getUncompressedSizeWithHeader();

      if (cacheThisBlock) {
        doCacheOnWrite(offset);
      }
    }
  }
}

public boolean shouldWriteBlock(boolean closing) {
  enqueueReadyChunk(closing);
  //readyChunks中保存的是chunk，也就是lead index block
  return !readyChunks.isEmpty();
}

bloomBlockIndexWriter.writeSingleLevelIndex(out, "Bloom filter");

List<StoreFileScanner> sfScanners = StoreFileScanner.getScannersForStoreFiles(storeFilesToScan, cacheBlocks, usePread, isCompaction, false, matcher, readPt);

//此方法会调用Open方法
StoreFile.Reader r = file.createReader(canUseDrop);

   if (this.reader != null) {
     throw new IllegalAccessError("Already open");
   }

   // Open the StoreFile.Reader
   this.reader = fileInfo.open(this.fs, this.cacheConf, canUseDropBehind);

   // Load up indices and fileinfo. This also loads Bloom filter type.
   metadataMap = Collections.unmodifiableMap(this.reader.loadFileInfo());

   // Read in our metadata.
   byte [] b = metadataMap.get(MAX_SEQ_ID_KEY);
   if (b != null) {
     // By convention, if halfhfile, top half has a sequence number > bottom
     // half. Thats why we add one in below. Its done for case the two halves
     // are ever merged back together --rare.  Without it, on open of store,
     // since store files are distinguished by sequence id, the one half would
     // subsume the other.
     this.sequenceid = Bytes.toLong(b);
     if (fileInfo.isTopReference()) {
       this.sequenceid += 1;
     }
   }

   if (isBulkLoadResult()){
     // generate the sequenceId from the fileName
     // fileName is of the form <randomName>_SeqId_<id-when-loaded>_
     String fileName = this.getPath().getName();
     // Use lastIndexOf() to get the last, most recent bulk load seqId.
     int startPos = fileName.lastIndexOf("SeqId_");
     if (startPos != -1) {
       this.sequenceid = Long.parseLong(fileName.substring(startPos + 6,
           fileName.indexOf('_', startPos + 6)));
       // Handle reference files as done above.
       if (fileInfo.isTopReference()) {
         this.sequenceid += 1;
       }
     }
     this.reader.setBulkLoaded(true);
   }
   this.reader.setSequenceID(this.sequenceid);

   b = metadataMap.get(HFileWriterV2.MAX_MEMSTORE_TS_KEY);
   if (b != null) {
     this.maxMemstoreTS = Bytes.toLong(b);
   }

   b = metadataMap.get(MAJOR_COMPACTION_KEY);
   if (b != null) {
     boolean mc = Bytes.toBoolean(b);
     if (this.majorCompaction == null) {
       this.majorCompaction = new AtomicBoolean(mc);
     } else {
       this.majorCompaction.set(mc);
     }
   } else {
     // Presume it is not major compacted if it doesn't explicity say so
     // HFileOutputFormat explicitly sets the major compacted key.
     this.majorCompaction = new AtomicBoolean(false);
   }

   b = metadataMap.get(EXCLUDE_FROM_MINOR_COMPACTION_KEY);
   this.excludeFromMinorCompaction = (b != null && Bytes.toBoolean(b));

//此出会读取bloom filter
   BloomType hfileBloomType = reader.getBloomFilterType();
   if (cfBloomType != BloomType.NONE) {
     reader.loadBloomfilter(BlockType.GENERAL_BLOOM_META);
     if (hfileBloomType != cfBloomType) {
       LOG.info("HFile Bloom filter type for "
           + reader.getHFileReader().getName() + ": " + hfileBloomType
           + ", but " + cfBloomType + " specified in column family "
           + "configuration");
     }
   } else if (hfileBloomType != BloomType.NONE) {
     LOG.info("Bloom filter turned off by CF config for "
         + reader.getHFileReader().getName());
   }

   // load delete family bloom filter
   reader.loadBloomfilter(BlockType.DELETE_FAMILY_BLOOM_META);

   try {
     this.reader.timeRange = TimeRangeTracker.getTimeRange(metadataMap.get(TIMERANGE_KEY));
   } catch (IllegalArgumentException e) {
     LOG.error("Error reading timestamp range data from meta -- " +
         "proceeding without", e);
     this.reader.timeRange = null;
   }
   return this.reader;
 

//此处使用bloomfilter过滤。在此方法中会调用bloomFilter.contains，在此contains会先使用block index 判断。
reader.passesBloomFilter(scan, columns)

   //判断读取哪个block，rootBlockContaingKey里的blockKeys为chunk的个数。
//index是从bloommeta中读取，DataInput bloomMeta = reader.getGeneralBloomFilterMetadata(); 代码获取。
int block = index.rootBlockContainingKey(key, keyOffset, keyLength);
   if (block < 0) {
     result = false; // This key is not in the file.
   } else {
     HFileBlock bloomBlock;
     try {
       // We cache the block and use a positional read.
       //读取那个chunk的bf
       bloomBlock = reader.readBlock(index.getRootBlockOffset(block),
           index.getRootBlockDataSize(block), true, true, false, true,
           BlockType.BLOOM_CHUNK);
     } catch (IOException ex) {
       // The Bloom filter is broken, turn it off.
       throw new IllegalArgumentException(
           "Failed to load Bloom block for key "
               + Bytes.toStringBinary(key, keyOffset, keyLength), ex);
     }

     ByteBuffer bloomBuf = bloomBlock.getBufferReadOnly();
     result = ByteBloomFilter.contains(key, keyOffset, keyLength,
         bloomBuf.array(), bloomBuf.arrayOffset() + bloomBlock.headerSize(),
         bloomBlock.getUncompressedSizeWithoutHeader(), hash, hashCount);
   }

BlockWithScanInfo org.apache.hadoop.hbase.io.hfile.HFileBlockIndex.BlockIndexReader.loadDataBlockWithScanInfo(byte[] key, int keyOffset, int keyLength, HFileBlock currentBlock, boolean cacheBlocks, boolean pread, boolean isCompaction) throws IOException

private synchronized CompactionRequest requestCompactionInternal(final HRegion r, final Store s,
      final String why, int priority, CompactionRequest request, boolean selectNow, User user)

boolean completed = region.compact(compaction, store, compactionThroughputController, user);

store.compact(compaction, throughputController, user);

List<Path> newFiles = compaction.compact(throughputController, user);

ScanType scanType = scannerFactory.getScanType(request);

sfs = moveCompatedFilesIntoPlace(cr, newFiles, user);

writeCompactionWalRecord(filesToCompact, sfs);

replaceStoreFiles(filesToCompact, sfs);

completeCompaction(filesToCompact);