// obj就是被广播的对象
 private val numBlocks: Int = writeBlocks(obj)

  override protected def getValue() = {
    _value
  }

@transient private lazy val _value: T = readBroadcastBlock()

private[spark] class TorrentBroadcast[T: ClassTag](obj: T, id: Long)
  extends Broadcast[T](id) with Logging with Serializable {
　　……
}

@transient private lazy val _value: T = readBroadcastBlock()

lazy val myLazyField = create();

   public volatile int bitmap$0;
   private Object myLazyField;

   public String myLazyField() {
        if((bitmap$0 & 1) == 0)
        {
            synchronized(this)
            {
                if((bitmap$0 & 1) == 0)
                {
                    myLazyField = ...
                    bitmap$0 = bitmap$0 | 1;
                }
            }
        }
        return myLazyField;
    }

val numBlocks: Int = writeBlocks(obj)

  private def writeBlocks(value: T): Int = {
    // Store a copy of the broadcast variable in the driver so that tasks run on the driver
    // do not create a duplicate copy of the broadcast variable‘s value.
    SparkEnv.get.blockManager.putSingle(broadcastId, value, StorageLevel.MEMORY_AND_DISK,
      tellMaster = false)
    val blocks =
      TorrentBroadcast.blockifyObject(value, blockSize, SparkEnv.get.serializer, compressionCodec)
    //blocks的类型是Array[ByteBuffer]
    blocks.zipWithIndex.foreach { case (block, i) =>
      SparkEnv.get.blockManager.putBytes(
        BroadcastBlockId(id, "piece" + i),//以BroadcastBlockId为BlockId存储
        block,
        StorageLevel.MEMORY_AND_DISK_SER,
        tellMaster = true)
    }
    blocks.length
  }

  def blockifyObject[T: ClassTag](
      obj: T,
      blockSize: Int,
      serializer: Serializer,
      compressionCodec: Option[CompressionCodec]): Array[ByteBuffer] = {
    val bos = new ByteArrayChunkOutputStream(blockSize)
    val out: OutputStream = compressionCodec.map(c => c.compressedOutputStream(bos)).getOrElse(bos)
    val ser = serializer.newInstance()
    val serOut = ser.serializeStream(out)
    serOut.writeObject[T](obj).close()
    bos.toArrays.map(ByteBuffer.wrap)
  }

private[spark] class ByteArrayChunkOutputStream(chunkSize: Int) extends OutputStream {

  private val chunks = new ArrayBuffer[Array[Byte]]
  private var lastChunkIndex = -1
  private var position = chunkSize
  override def write(b: Int): Unit = {
    allocateNewChunkIfNeeded()
    chunks(lastChunkIndex)(position) = b.toByte
    position += 1
  }

　override def write(bytes: Array[Byte], off: Int, len: Int): Unit = { ... }

　def toArrays: Array[Array[Byte]] = { ... }

 ... 

}

 @transient private lazy val _value: T = readBroadcastBlock()

 private def readBroadcastBlock(): T = Utils.tryOrIOException {
    TorrentBroadcast.synchronized {
      setConf(SparkEnv.get.conf)
      //从本地的blockManager里读这个被broadcast的对象，根据broadcastId
      SparkEnv.get.blockManager.getLocal(broadcastId).map(_.data.next()) match {
        case Some(x) => //本地有
          x.asInstanceOf[T]

        case None => //本地无
          logInfo("Started reading broadcast variable " + id)
          val startTimeMs = System.currentTimeMillis()
          val blocks = readBlocks()//如果本地没有broadcastId对应的broadcast的block，就读
          logInfo("Reading broadcast variable " + id + " took" + Utils.getUsedTimeMs(startTimeMs))

          val obj = TorrentBroadcast.unBlockifyObject[T](
            blocks, SparkEnv.get.serializer, compressionCodec)
          // Store the merged copy in BlockManager so other tasks on this executor don‘t
          // need to re-fetch it.
          SparkEnv.get.blockManager.putSingle( //读了之后再放进BlockManager
            broadcastId, obj, StorageLevel.MEMORY_AND_DISK, tellMaster = false)
          obj
      }
    }
  }

  /** Fetch torrent blocks from the driver and/or other executors. */
  private def readBlocks(): Array[ByteBuffer] = {
    //获取到的block被存在本地的BlockManager中并且上报给driver,这样其它的executor就可以从这个executor获取这些block了
    val blocks = new Array[ByteBuffer](numBlocks)
    val bm = SparkEnv.get.blockManager

    //需要shuffle，避免所有executor以同样的顺序下载block，使得driver依然是瓶颈
    for (pid <- Random.shuffle(Seq.range(0, numBlocks))) {
      val pieceId = BroadcastBlockId(id, "piece" + pid)//组装BroadcastBlockId
      logDebug(s"Reading piece $pieceId of $broadcastId")
      // 先试着从本地获取，因为之前的尝试可能已经获取了一些block
      def getLocal: Option[ByteBuffer] = bm.getLocalBytes(pieceId)
      def getRemote: Option[ByteBuffer] = bm.getRemoteBytes(pieceId).map { block =>
        //如果从remote获取了block，就把它存在本地的BlockManager
        SparkEnv.get.blockManager.putBytes(
          pieceId,
          block,
          StorageLevel.MEMORY_AND_DISK_SER,
          tellMaster = true)
        block
      }
      val block: ByteBuffer = getLocal.orElse(getRemote).getOrElse(
        throw new SparkException(s"Failed to get $pieceId of $broadcastId"))
      blocks(pid) = block
    }
    blocks
  }