JMeter源码解析之结果收集器

 void runNonGui(String testFile, String logFile, boolean remoteStart, String remoteHostsString, boolean generateReportDashboard){
 ....
 ResultCollector resultCollector = null;
   if (logFile != null) {
     resultCollector = new ResultCollector(summariser);
     resultCollector.setFilename(logFile);
     clonedTree.add(clonedTree.getArray()[0], resultCollector);
     }
   else {
     // only add Summariser if it can not be shared with the ResultCollector
   if (summariser != null) {
      clonedTree.add(clonedTree.getArray()[0], summariser);
      }
      }
 ....

 }

private void executeSamplePackage(Sampler current,
      TransactionSampler transactionSampler,
      SamplePackage transactionPack,
      JMeterContext threadContext) {
  threadContext.setCurrentSampler(current);
  // Get the sampler ready to sample
  SamplePackage pack = compiler.configureSampler(current);
  runPreProcessors(pack.getPreProcessors());//运行前置处理器
  // Hack: save the package for any transaction controllers
  threadVars.putObject(PACKAGE_OBJECT, pack);
  delay(pack.getTimers());//定时器timer
  SampleResult result = null;
  if (running) {
       Sampler sampler = pack.getSampler();
       result = doSampling(threadContext, sampler);
   }
   // If we got any results, then perform processing on the result
   if (result != null) {
   if (!result.isIgnore()) {
          ...                
   runPostProcessors(pack.getPostProcessors());//运行后置处理器
   checkAssertions(pack.getAssertions(), result, threadContext);//运行断言处理器
            // PostProcessors can call setIgnore, so reevaluate here
            if (!result.isIgnore()) {
            // Do not send subsamples to listeners which receive the transaction sample
            List<SampleListener> sampleListeners = getSampleListeners(pack, transactionPack, transactionSampler);
            notifyListeners(sampleListeners, result);//执行监听器,此处为执行报告收集器的sampleOccurred方法
            }
            compiler.done(pack);
            ...
    }

@Override
public void sampleOccurred(SampleEvent event) {
    SampleResult result = event.getResult();
    if (isSampleWanted(result.isSuccessful())) {
        sendToVisualizer(result);
        if (out != null && !isResultMarked(result) && !this.isStats) {
        SampleSaveConfiguration config = getSaveConfig();
        result.setSaveConfig(config);
        try {
               if (config.saveAsXml()) {
                   SaveService.saveSampleResult(event, out);
               } else { // !saveAsXml
                   CSVSaveService.saveSampleResult(event, out);
               }
          } catch (Exception err) {
              log.error("Error trying to record a sample", err); // should throw exception back to caller
           }
      }
  }
   if(summariser != null) {
       summariser.sampleOccurred(event);
   }
}

@Override
public void processBatch(List<SampleEvent> samples) {
    if (samples != null && sampleListener != null) {
        for (SampleEvent e : samples) {
            sampleListener.sampleOccurred(e);
        }
    }
}
@Override
public void sampleOccurred(SampleEvent e) {
    if (sampleListener != null) {
        sampleListener.sampleOccurred(e);
    }
}

public class AsynchSampleSender extends AbstractSampleSender implements Serializable {
       protected Object readResolve() throws ObjectStreamException{
        int capacity = getCapacity();
        log.info("Using batch queue size (asynch.batch.queue.size): {}", capacity); // server log file
        queue = new ArrayBlockingQueue<>(capacity);
        Worker worker = new Worker(queue, listener);
        worker.setDaemon(true);
        worker.start();
        return this;
    }
@Override
public void testEnded(String host) 
    log.debug("Test Ended on {}", host);
    try {
        listener.testEnded(host);
        queue.put(FINAL_EVENT);
    } catch (Exception ex) {
        log.warn("testEnded(host)", ex);
    }
    if (queueWaits > 0) {
        log.info("QueueWaits: {}; QueueWaitTime: {} (nanoseconds)", queueWaits, queueWaitTime);
        }
    }
 @Override
public void sampleOccurred(SampleEvent e) 
    try {
        if (!queue.offer(e)){ // we failed to add the element first time
            queueWaits++;
            long t1 = System.nanoTime();
            queue.put(e);
            long t2 = System.nanoTime();
            queueWaitTime += t2-t1;
        }
    } catch (Exception err) {
        log.error("sampleOccurred; failed to queue the sample", err);
    }
}
private static class Worker extends Thread {
    @Override
    public void run() 
        try {
            boolean eof = false;
            while (!eof) {
                List<SampleEvent> l = new ArrayList<>();
                SampleEvent e = queue.take();
                // try to process as many as possible
                // The == comparison is not an error
                while (!(eof = e == FINAL_EVENT) && e != null) {
                     l.add(e);
                     e = queue.poll(); // returns null if nothing on queue currently
                 }
                int size = l.size();
                if (size > 0) {
                    try {
                       listener.processBatch(l);
                    } catch (RemoteException err) {
                        if (err.getCause() instanceof java.net.ConnectException){
                            throw new JMeterError("Could not return sample",err);
                        }
                        log.error("Failed to return sample", err);
                    }
                 }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            }
        log.debug("Worker ended");
        }
    }
}