private case class MemoryEntry(value: Any, size: Long, deserialized: Boolean)

class MemoryStore(blockManager: BlockManager, maxMemory: Long)extends BlockStore(blockManager) {

    private val entries = new LinkedHashMap[BlockId, MemoryEntry](32, 0.75f, true)

    //存

    private def tryToPut(blockId: BlockId, value: Any,size: Long,deserialized: Boolean): ResultWithDroppedBlocks = {

         if (enoughFreeSpace) { //空闲内存是否足以容纳block

            val entry = new MemoryEntry(value, size, deserialized)

            entries.synchronized {

                entries.put(blockId, entry) //将Block放置到内部维护的HashMap中

            }

            //如果是反序列话的就以对象数组方式处理，否则就是以字节数组方式处理

            val valuesOrBytes = if (deserialized) "values" else "bytes"

         }else{//告诉BlockManager内存不足以存下该block，是否将其drop到硬盘中(如果该Block允许Disk存储)

            val droppedBlockStatus = blockManager.dropFromMemory(blockId, data)

         }

    }

    //取：直接从HashMap中根据blockid获取即可

    override def getValues(blockId: BlockId): Option[Iterator[Any]] = {

        val entry = entries.synchronized {

            entries.get(blockId)

        }

        if (entry == null) {

            None

        } else if (entry.deserialized) { //反序列话的就以对象数组方式处理

            Some(entry.value.asInstanceOf[Array[Any]].iterator)

        } else { //序列话的就以字节数组方式处理

            val buffer = entry.value.asInstanceOf[ByteBuffer].duplicate() // Doesn't actually copy data

            Some(blockManager.dataDeserialize(blockId, buffer))

        }

    }

}