**
 * A StreamingListener that receives batch completion     updates, and maintains
 * an estimate of the speed at which this stream should ingest messages,
 * given an estimate computation from a `RateEstimator`
 */
private[streaming] abstract class RateController(val streamUID: Int, rateEstimator: RateEstimator)
extends StreamingListener with Serializable {
  /**
   * Compute the new rate limit and publish it asynchronously.
   */
  private def computeAndPublish(time: Long, elems: Long, workDelay: Long, waitDelay: Long): Unit =
Future[Unit] {
  val newRate = rateEstimator.compute(time, elems, workDelay, waitDelay)
  newRate.foreach { s =>
    rateLimit.set(s.toLong)
    publish(getLatestRate())
  }
}
def getLatestRate(): Long = rateLimit.get()

override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted) {
val elements = batchCompleted.batchInfo.streamIdToInputInfo
for {
  processingEnd <- batchCompleted.batchInfo.processingEndTime
  workDelay <- batchCompleted.batchInfo.processingDelay
  waitDelay <- batchCompleted.batchInfo.schedulingDelay
  elems <- elements.get(streamUID).map(_.numRecords)
} computeAndPublish(processingEnd, elems, workDelay, waitDelay)
}
}

def start(): Unit = synchronized {
   if (eventLoop != null) return // scheduler has already been started

   logDebug("Starting JobScheduler")
   eventLoop = new EventLoop[JobSchedulerEvent]("JobScheduler") {
   override protected def onReceive(event: JobSchedulerEvent): Unit = processEvent(event)

   override protected def one rror(e: Throwable): Unit = reportError("Error in job scheduler", e)
 }
 eventLoop.start()

 // attach rate controllers of input streams to receive batch completion updates
 for {
   inputDStream <- ssc.graph.getInputStreams
   rateController <- inputDStream.rateController
 } ssc.addStreamingListener(rateController)</span>

 listenerBus.start()
 receiverTracker = new ReceiverTracker(ssc)
 inputInfoTracker = new InputInfoTracker(ssc)
 receiverTracker.start()
 jobGenerator.start()
 logInfo("Started JobScheduler")
}

 /** Generate jobs and perform checkpoint for the given `time`.  */
private def generateJobs(time: Time) {
  // Set the SparkEnv in this thread, so that job generation code can access the environment
  // Example: BlockRDDs are created in this thread, and it needs to access BlockManager
  // Update: This is probably redundant after threadlocal stuff in SparkEnv has been removed.
  SparkEnv.set(ssc.env)

  // Checkpoint all RDDs marked for checkpointing to ensure their lineages are
  // truncated periodically. Otherwise, we may run into stack overflows (SPARK-6847).
  ssc.sparkContext.setLocalProperty(RDD.CHECKPOINT_ALL_MARKED_ANCESTORS, "true")
  Try {
    jobScheduler.receiverTracker.allocateBlocksToBatch(time) // allocate received blocks to batch
    graph.generateJobs(time) // generate jobs using allocated block
  } match {
    case Success(jobs) =>
      val streamIdToInputInfos = jobScheduler.inputInfoTracker.getInfo(time)
jobScheduler.submitJobSet(JobSet(time, jobs, streamIdToInputInfos))
    case Failure(e) =>
      jobScheduler.reportError("Error generating jobs for time " + time, e)
}
eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = false))
}

def submitJobSet(jobSet: JobSet) {
  if (jobSet.jobs.isEmpty) {
    logInfo("No jobs added for time " + jobSet.time)
  } else {
    listenerBus.post(StreamingListenerBatchSubmitted(jobSet.toBatchInfo))
    jobSets.put(jobSet.time, jobSet)
    jobSet.jobs.foreach(job => jobExecutor.execute(new JobHandler(job)))
    logInfo("Added jobs for time " + jobSet.time)
  }
}

private def handleJobCompletion(job: Job, completedTime: Long) {
   val jobSet = jobSets.get(job.time)
   jobSet.handleJobCompletion(job)
   job.setEndTime(completedTime)
   listenerBus.post(StreamingListenerOutputOperationCompleted(job.toOutputOperationInfo))
   logInfo("Finished job " + job.id + " from job set of time " + jobSet.time)
   if (jobSet.hasCompleted) {
     jobSets.remove(jobSet.time)
     jobGenerator.onBatchCompletion(jobSet.time)
     logInfo("Total delay: %.3f s for time %s (execution: %.3f s)".format(
     jobSet.totalDelay / 1000.0, jobSet.time.toString,
     jobSet.processingDelay / 1000.0
   ))
 listenerBus.post(StreamingListenerBatchCompleted(jobSet.toBatchInfo))
 }
 job.result match {
   case Failure(e) =>
       reportError("Error running job " + job, e)
   case _ =>
 }
}

override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted) {
  val elements = batchCompleted.batchInfo.streamIdToInputInfo

  for {
    processingEnd <- batchCompleted.batchInfo.processingEndTime
    workDelay <- batchCompleted.batchInfo.processingDelay
    waitDelay <- batchCompleted.batchInfo.schedulingDelay
    elems <- elements.get(streamUID).map(_.numRecords)
  } computeAndPublish(processingEnd, elems, workDelay, waitDelay)
}

/**
   * Compute the new rate limit and publish it asynchronously.
   */
  private def computeAndPublish(time: Long, elems: Long, workDelay: Long, waitDelay: Long): Unit =
Future[Unit] {
  val newRate = rateEstimator.compute(time, elems, workDelay, waitDelay)
  newRate.foreach { s =>
    rateLimit.set(s.toLong)
    publish(getLatestRate())
  }
}

/**
   * A RateController that sends the new rate to receivers, via the receiver tracker.
   */
 private[streaming] class ReceiverRateController(id: Int, estimator: RateEstimator)
  extends RateController(id, estimator) {
  override def publish(rate: Long): Unit =
    ssc.scheduler.receiverTracker.sendRateUpdate(id, rate)
}

/** Update a receiver‘s maximum ingestion rate */
def sendRateUpdate(streamUID: Int, newRate: Long):   Unit = synchronized {
  if (isTrackerStarted) {
    endpoint.send(UpdateReceiverRateLimit(streamUID, newRate))
  }
}

/**
  * Set the rate limit to `newRate`. The new rate will not exceed the maximum rate configured by
  * {{{spark.streaming.receiver.maxRate}}}, even if `newRate` is higher than that.
  *
  * @param newRate A new rate in events per second. It has no effect if it‘s 0 or negative.
  */
 private[receiver] def updateRate(newRate: Long): Unit =
   if (newRate > 0) {
   if (maxRateLimit > 0) {
     rateLimiter.setRate(newRate.min(maxRateLimit))
   } else {
     rateLimiter.setRate(newRate)
   }
 }

public final void setRate(double permitsPerSecond) {
  Preconditions.checkArgument(permitsPerSecond > 0.0
    && !Double.isNaN(permitsPerSecond), "rate must be positive");
  synchronized (mutex) {
    resync(readSafeMicros());
    double stableIntervalMicros = TimeUnit.SECONDS.toMicros(1L) / permitsPerSecond;  //固定间隔
    this.stableIntervalMicros = stableIntervalMicros;
    doSetRate(permitsPerSecond, stableIntervalMicros);
  }
}

  /**
   * Push a single data item into the buffer.
   */
  def addData(data: Any): Unit = {
      if (state == Active) {
         waitToPush()  //获取令牌
        synchronized {
          if (state == Active) {
            currentBuffer += data
          } else {
            throw new SparkException(
        "Cannot add data as BlockGenerator has not been started or has been stopped")
          }
        }
      } else {
        throw new SparkException(
    "Cannot add data as BlockGenerator has not been started or has been stopped")
}