kafka源码愫读(4)、KafkaController模块源码分析

1、模块组成

KafkaController是kafka集群的控制管理模块,且一个集群只能有一个leader。其主要通过向ZK注册各种监听事件来管理整个集群节点、分区的leader的选举、再平衡等问题。

KafkaController主要组成:

KafkaController.png

说明:

  • ControllerEventManager:Controller事件管理器。KafkaController监听zk节点变化事件,会将事件放入ControllerEventManager的事件队列,而ControllerEventManager内部的事件处理线程会从队列中获取事件,并调用KafkaController的process()处理事件;
  • PartitionStateMachine:分区状态机。定义及管理分区的状态;
  • ReplicaStateMachine:副本状态机。定义及管理副本的状态;
  • TopicDeletionManager:TopicDeletionManager 负责对管理员指定的 topic 执行删除操作,它定义了 DeleteTopicsThread 线程,采用异步的方式删除待删除的 topic 集合;
  • Handler:事件处理器。KafkaController在初始化的时候,会监听zk节点的事件,而不同节点会绑定不同的事件处理器;

2、初始化流程

2.1、KafkaServer#startup()方法

startup()方法会在KafkaServer中进行调用。

def startup() = {
  //注册zk状态变更处理器,当zk重新初始化时会调用
  zkClient.registerStateChangeHandler(new StateChangeHandler {
    override val name: String = StateChangeHandlers.ControllerHandler
    //zk初始化之后调用
    override def afterInitializingSession(): Unit = {
      //设置RegisterBrokerAndReelect事件,重新选举Controller的leader
      eventManager.put(RegisterBrokerAndReelect)
    }
    //zk初始化之前调用
    override def beforeInitializingSession(): Unit = {
      val queuedEvent = eventManager.clearAndPut(Expire)

      // Block initialization of the new session until the expiration event is being handled,
      // which ensures that all pending events have been processed before creating the new session
      //等待controller中的事件处理完
      queuedEvent.awaitProcessing()
    }
  })
  //设置Startup事件,eventManager会调用KafkaController#process()处理Startup事件,
  //Startup事件中会进行controller的选举等初始化处理
  eventManager.put(Startup)
  eventManager.start()
}

2.2、KafkaServer#processStartup()方法

processStartup()主要处理注册【/controller】节点监听、选举及初始化等工作

private def processStartup(): Unit = {
  //注册【/controller】节点变更处理器并检查节点是否存在
  zkClient.registerZNodeChangeHandlerAndCheckExistence(controllerChangeHandler)
  //选举处理
  elect()
}

ControllerChangeHandler为【/controller】节点变更处理器,当节点被创建/删除/修改时,会生成对应的事件。

class ControllerChangeHandler(eventManager: ControllerEventManager) extends ZNodeChangeHandler {
  override val path: String = ControllerZNode.path

  //当节点被创建,表明其他某个broker被选举为leader,需处理ControllerChange事件
  override def handleCreation(): Unit = eventManager.put(ControllerChange)
  //当节点被删除,表明当前无leader节点,处理Reelect事件,进行重新选举
  override def handleDeletion(): Unit = eventManager.put(Reelect)
  //当节点数据变更,表明leader可能变更,需处理ControllerChange事件
  override def handleDataChange(): Unit = eventManager.put(ControllerChange)
}

2.3、KafkaServer#elect()方法

elect()主要进行选举及Controller初始化处理。

private def elect(): Unit = {
  activeControllerId = zkClient.getControllerId.getOrElse(-1)
  /*
   * We can get here during the initial startup and the handleDeleted ZK callback. Because of the potential race condition,
   * it's possible that the controller has already been elected when we get here. This check will prevent the following
   * createEphemeralPath method from getting into an infinite loop if this broker is already the controller.
   */
   //当activeControllerId 不为-1,表明已经有其他Controller成为leader,无需进行后续处理
  if (activeControllerId != -1) {
    debug(s"Broker $activeControllerId has been elected as the controller, so stopping the election process.")
    return
  }

  try {
    //注册【/controller】及【/controller_epoch】节点,即利用zk节点唯一特性选举出leader
    val (epoch, epochZkVersion) = zkClient.registerControllerAndIncrementControllerEpoch(config.brokerId)
    controllerContext.epoch = epoch
    controllerContext.epochZkVersion = epochZkVersion
    activeControllerId = config.brokerId

    info(s"${config.brokerId} successfully elected as the controller. Epoch incremented to ${controllerContext.epoch} " +
      s"and epoch zk version is now ${controllerContext.epochZkVersion}")
    //若当前节点成为leader,则进行Controller的相关初始化处理
    onControllerFailover()
  } catch {
    //若controller已存在,则调用maybeResign(),其内进行Controller的一些清理操作
    case e: ControllerMovedException =>
      maybeResign()

      if (activeControllerId != -1)
        debug(s"Broker $activeControllerId was elected as controller instead of broker ${config.brokerId}", e)
      else
        warn("A controller has been elected but just resigned, this will result in another round of election", e)

    //异常,则进行Controller主动变更处理,即清除上下文信息及删除【/controller】节点,触发重新选举
    case t: Throwable =>
      error(s"Error while electing or becoming controller on broker ${config.brokerId}. " +
        s"Trigger controller movement immediately", t)
      triggerControllerMove()
  }
}

2.4、KafkaServer#onControllerFailover()方法

onControllerFailover()为Controller初始化的核心方法,其主要进行节点事件注册、删除leader不用的监听器、初始化上下文、topic删除管理器初始化、状态机初始、分区初始化处理等;

private def onControllerFailover(): Unit = {
  info("Registering handlers")

  // before reading source of truth from zookeeper, register the listeners to get broker/topic callbacks
  //子节点变更处理器注册
  val childChangeHandlers = Seq(brokerChangeHandler, topicChangeHandler, topicDeletionHandler, logDirEventNotificationHandler,
    isrChangeNotificationHandler)
  childChangeHandlers.foreach(zkClient.registerZNodeChildChangeHandler)
   //节点变更处理器注册
  val nodeChangeHandlers = Seq(preferredReplicaElectionHandler, partitionReassignmentHandler)
  nodeChangeHandlers.foreach(zkClient.registerZNodeChangeHandlerAndCheckExistence)

  info("Deleting log dir event notifications")
  zkClient.deleteLogDirEventNotifications(controllerContext.epochZkVersion)
  info("Deleting isr change notifications")
  zkClient.deleteIsrChangeNotifications(controllerContext.epochZkVersion)
  
  //上下文初始化
  info("Initializing controller context")
  initializeControllerContext()
  
  //获取待删除的topic,并初始化TopicDeletionManager
  info("Fetching topic deletions in progress")
  val (topicsToBeDeleted, topicsIneligibleForDeletion) = fetchTopicDeletionsInProgress()
  info("Initializing topic deletion manager")
  topicDeletionManager.init(topicsToBeDeleted, topicsIneligibleForDeletion)

  // We need to send UpdateMetadataRequest after the controller context is initialized and before the state machines
  // are started. The is because brokers need to receive the list of live brokers from UpdateMetadataRequest before
  // they can process the LeaderAndIsrRequests that are generated by replicaStateMachine.startup() and
  // partitionStateMachine.startup().
  //与其他server同步metadata信息
  info("Sending update metadata request")
  sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq, Set.empty)
  //副本状态机及分区状态机启动  
  replicaStateMachine.startup()
  partitionStateMachine.startup()

  info(s"Ready to serve as the new controller with epoch $epoch")
  //处理分区重分配
  initializePartitionReassignments()
  topicDeletionManager.tryTopicDeletion()
  val pendingPreferredReplicaElections = fetchPendingPreferredReplicaElections()
  //针对各个分区副本,选举出主分区
  onReplicaElection(pendingPreferredReplicaElections, ElectionType.PREFERRED, ZkTriggered)
  info("Starting the controller scheduler")
  kafkaScheduler.startup()
  
  //自动重分配任务
  if (config.autoLeaderRebalanceEnable) {
    scheduleAutoLeaderRebalanceTask(delay = 5, unit = TimeUnit.SECONDS)
  }

  if (config.tokenAuthEnabled) {
    info("starting the token expiry check scheduler")
    tokenCleanScheduler.startup()
    tokenCleanScheduler.schedule(name = "delete-expired-tokens",
      fun = () => tokenManager.expireTokens,
      period = config.delegationTokenExpiryCheckIntervalMs,
      unit = TimeUnit.MILLISECONDS)
  }
}

2.5、KafkaServer#onControllerResignation()方法

当broker从leader变更为follower时,调用onControllerResignation()进行一些上下文的清理及初始化工作。

private def onControllerResignation(): Unit = {
  debug("Resigning")
  // de-register listeners
  //注销一些监听器
  zkClient.unregisterZNodeChildChangeHandler(isrChangeNotificationHandler.path)
  zkClient.unregisterZNodeChangeHandler(partitionReassignmentHandler.path)
  zkClient.unregisterZNodeChangeHandler(preferredReplicaElectionHandler.path)
  zkClient.unregisterZNodeChildChangeHandler(logDirEventNotificationHandler.path)
  unregisterBrokerModificationsHandler(brokerModificationsHandlers.keySet)

  // shutdown leader rebalance scheduler
  //停止leader的重平衡调度器及一些全局计数器
  kafkaScheduler.shutdown()
  offlinePartitionCount = 0
  preferredReplicaImbalanceCount = 0
  globalTopicCount = 0
  globalPartitionCount = 0
  topicsToDeleteCount = 0
  replicasToDeleteCount = 0
  ineligibleTopicsToDeleteCount = 0
  ineligibleReplicasToDeleteCount = 0

  // stop token expiry check scheduler
  if (tokenCleanScheduler.isStarted)
    tokenCleanScheduler.shutdown()

  // de-register partition ISR listener for on-going partition reassignment task
  unregisterPartitionReassignmentIsrChangeHandlers()
  // shutdown partition state machine
  //关闭分区状态机
  partitionStateMachine.shutdown()
  
  //删除一些节点监听器
  zkClient.unregisterZNodeChildChangeHandler(topicChangeHandler.path)
  unregisterPartitionModificationsHandlers(partitionModificationsHandlers.keys.toSeq)
  zkClient.unregisterZNodeChildChangeHandler(topicDeletionHandler.path)
  // shutdown replica state machine
  //关闭副本状态机
  replicaStateMachine.shutdown()
  zkClient.unregisterZNodeChildChangeHandler(brokerChangeHandler.path)

  //关闭controller连接
  controllerChannelManager.shutdown()
  //恢复上下文信息
  controllerContext.resetContext()

  info("Resigned")
}

3、节点监听器及事件处理

节点 监听类型 监听器 事件类型 event
/controller 节点变更 ControllerChangeHandler Create ControllerChange
Delete Reelect DataChange ControllerChange
/brokers/ids 子节点变更 BrokerChangeHandler ChildChange BrokerChange
/brokers/ids/[brokerId] 节点变更 BrokerModificationsHandler DataChange BrokerModifications
/brokers/topics 子节点变更 TopicChangeHandler ChildChange TopicChange
/admin/delete_topics 子节点变更 TopicDeletionHandler ChildChange
TopicDeletion
/brokers/topics/[topic] 节点变更 PartitionModificationsHandler DataChange PartitionModifications
/admin/reassign_partitions 节点变更 PartitionReassignmentHandler Create ZkPartitionReassignment
/admin/preferred_replica_election 节点变更 PreferredReplicaElectionHandler Create ReplicaLeaderElection
/isr_change_notification 子节点变更 IsrChangeNotificationHandler ChildChange IsrChangeNotification
/log_dir_event_notification 子节点变更 LogDirEventNotificationHandler ChildChange LogDirEventNotification

4、副本及分区状态机

4.1、Replica状态机

副本状态转换图:

副本状态转换图.png

ReplicaState状态说明:

  • NewReplica:当创建了topic或者重分配分区时Controller会创建新的副本,就处在这个状态,此状态中的副本只能接收“成为follower”的状态变更请求,可由NonExistentReplica转换而来
  • OnlineReplica:一旦启动了一个副本以及该分区AR副本集合中的一部分,那么就将设置该副本状态为OnlineReplica。在此状态中的副本可以接收”成为leader”或”成为follower”的状态变更请求。可由NewRelica、OnlineReplica或OfflineReplica状态转换而来
  • OfflineReplica:如果一个副本挂掉(保存该副本的broker宕机)将被置于OfflineReplica状态,可由NewReplica或OnlineReplica状态转换而来
  • ReplicaDeletionStarted:开启副本删除操作时会将副本状态置于ReplicaDeletionStarted状态,可由OfflineReplica状态转换而来
  • ReplicaDeletionSuccessful:如果副本删除请求成功,返回的响应没有错误的话,该副本会被置于ReplicaDeletionSuccessful状态,可由ReplicaDeletionStarted状态转换而来
  • ReplicaDeletionIneligible:如果副本删除失败,将被置于ReplicaDeletionIneligible状态,可由ReplicaDeletionStarted状态转换而来
  • NonExistentReplica:如果副本被成功删除将被置于NonExistentReplica状态,可由ReplicaDeletionSuccessful状态转换而来

副本状态流转:

override def handleStateChanges(replicas: Seq[PartitionAndReplica], targetState: ReplicaState): Unit = {
  if (replicas.nonEmpty) {
    try {
      controllerBrokerRequestBatch.newBatch()
      replicas.groupBy(_.replica).foreach { case (replicaId, replicas) =>
        doHandleStateChanges(replicaId, replicas, targetState)
      }
      //follower节点状态同步
      controllerBrokerRequestBatch.sendRequestsToBrokers(controllerContext.epoch)
    } catch {
      case e: ControllerMovedException =>
        error(s"Controller moved to another broker when moving some replicas to $targetState state", e)
        throw e
      case e: Throwable => error(s"Error while moving some replicas to $targetState state", e)
    }
  }
}
private def doHandleStateChanges(replicaId: Int, replicas: Seq[PartitionAndReplica], targetState: ReplicaState): Unit = {
  replicas.foreach(replica => controllerContext.putReplicaStateIfNotExists(replica, NonExistentReplica))
  //检查前置状态
  val (validReplicas, invalidReplicas) = controllerContext.checkValidReplicaStateChange(replicas, targetState)
  invalidReplicas.foreach(replica => logInvalidTransition(replica, targetState))

  targetState match {
    case NewReplica =>
      validReplicas.foreach { replica =>
        val partition = replica.topicPartition
        val currentState = controllerContext.replicaState(replica)

        controllerContext.partitionLeadershipInfo.get(partition) match {
          case Some(leaderIsrAndControllerEpoch) =>
            //当前节点为leader节点?
            if (leaderIsrAndControllerEpoch.leaderAndIsr.leader == replicaId) {
              val exception = new StateChangeFailedException(s"Replica $replicaId for partition $partition cannot be moved to NewReplica state as it is being requested to become leader")
              logFailedStateChange(replica, currentState, OfflineReplica, exception)
            } else {
              //封装发送给这些replica所在的broker的LeaderAndIsrRequest请求
              controllerBrokerRequestBatch.addLeaderAndIsrRequestForBrokers(Seq(replicaId),
                replica.topicPartition,
                leaderIsrAndControllerEpoch,
                controllerContext.partitionFullReplicaAssignment(replica.topicPartition),
                isNew = true)
              logSuccessfulTransition(replicaId, partition, currentState, NewReplica)
              controllerContext.putReplicaState(replica, NewReplica)
            }
          case None =>
            logSuccessfulTransition(replicaId, partition, currentState, NewReplica)
            controllerContext.putReplicaState(replica, NewReplica)
        }
      }
    case OnlineReplica =>
      validReplicas.foreach { replica =>
        val partition = replica.topicPartition
        val currentState = controllerContext.replicaState(replica)

        currentState match {
          case NewReplica =>
            val assignment = controllerContext.partitionFullReplicaAssignment(partition)
            if (!assignment.replicas.contains(replicaId)) {
              error(s"Adding replica ($replicaId) that is not part of the assignment $assignment")
              val newAssignment = assignment.copy(replicas = assignment.replicas :+ replicaId)
              controllerContext.updatePartitionFullReplicaAssignment(partition, newAssignment)
            }
          case _ =>
            //可能之前已经存在,则向其发送leader和isr的request
            controllerContext.partitionLeadershipInfo.get(partition) match {
              case Some(leaderIsrAndControllerEpoch) =>
                controllerBrokerRequestBatch.addLeaderAndIsrRequestForBrokers(Seq(replicaId),
                  replica.topicPartition,
                  leaderIsrAndControllerEpoch,
                  controllerContext.partitionFullReplicaAssignment(partition), isNew = false)
              case None =>
            }
        }
        logSuccessfulTransition(replicaId, partition, currentState, OnlineReplica)
        controllerContext.putReplicaState(replica, OnlineReplica)
      }
    case OfflineReplica =>
      //封装发送给这些replica所在的broker的StopReplicaRequest请求,交由ControllerBrokerRequestBatch处理
      validReplicas.foreach { replica =>
        controllerBrokerRequestBatch.addStopReplicaRequestForBrokers(Seq(replicaId), replica.topicPartition, deletePartition = false)
      }
      val (replicasWithLeadershipInfo, replicasWithoutLeadershipInfo) = validReplicas.partition { replica =>
        controllerContext.partitionLeadershipInfo.contains(replica.topicPartition)
      }
      //删除该replica
      val updatedLeaderIsrAndControllerEpochs = removeReplicasFromIsr(replicaId, replicasWithLeadershipInfo.map(_.topicPartition))
      updatedLeaderIsrAndControllerEpochs.foreach { case (partition, leaderIsrAndControllerEpoch) =>
        //此topic的partition的replicas发生了shrink(缩减),需要通知其它的replica
        if (!controllerContext.isTopicQueuedUpForDeletion(partition.topic)) {
          val recipients = controllerContext.partitionReplicaAssignment(partition).filterNot(_ == replicaId)
          controllerBrokerRequestBatch.addLeaderAndIsrRequestForBrokers(recipients,
            partition,
            leaderIsrAndControllerEpoch,
            controllerContext.partitionFullReplicaAssignment(partition), isNew = false)
        }
        val replica = PartitionAndReplica(partition, replicaId)
        val currentState = controllerContext.replicaState(replica)
        logSuccessfulTransition(replicaId, partition, currentState, OfflineReplica)
        controllerContext.putReplicaState(replica, OfflineReplica)
      }

      replicasWithoutLeadershipInfo.foreach { replica =>
        val currentState = controllerContext.replicaState(replica)
        logSuccessfulTransition(replicaId, replica.topicPartition, currentState, OfflineReplica)
        controllerBrokerRequestBatch.addUpdateMetadataRequestForBrokers(controllerContext.liveOrShuttingDownBrokerIds.toSeq, Set(replica.topicPartition))
        controllerContext.putReplicaState(replica, OfflineReplica)
      }
    case ReplicaDeletionStarted =>
      validReplicas.foreach { replica =>
        val currentState = controllerContext.replicaState(replica)
        logSuccessfulTransition(replicaId, replica.topicPartition, currentState, ReplicaDeletionStarted)
        controllerContext.putReplicaState(replica, ReplicaDeletionStarted)
        controllerBrokerRequestBatch.addStopReplicaRequestForBrokers(Seq(replicaId), replica.topicPartition, deletePartition = true)
      }
    case ReplicaDeletionIneligible =>
      validReplicas.foreach { replica =>
        val currentState = controllerContext.replicaState(replica)
        logSuccessfulTransition(replicaId, replica.topicPartition, currentState, ReplicaDeletionIneligible)
        controllerContext.putReplicaState(replica, ReplicaDeletionIneligible)
      }
    case ReplicaDeletionSuccessful =>
      validReplicas.foreach { replica =>
        val currentState = controllerContext.replicaState(replica)
        logSuccessfulTransition(replicaId, replica.topicPartition, currentState, ReplicaDeletionSuccessful)
        controllerContext.putReplicaState(replica, ReplicaDeletionSuccessful)
      }
    case NonExistentReplica =>
      validReplicas.foreach { replica =>
        val currentState = controllerContext.replicaState(replica)
        val newAssignedReplicas = controllerContext
          .partitionFullReplicaAssignment(replica.topicPartition)
          .removeReplica(replica.replica)

        controllerContext.updatePartitionFullReplicaAssignment(replica.topicPartition, newAssignedReplicas)
        logSuccessfulTransition(replicaId, replica.topicPartition, currentState, NonExistentReplica)
        controllerContext.removeReplicaState(replica)
      }
  }
}

4.2、Partition状态机

分区状态转换图:

分区状态转换图.png

Partition状态说明:

  • NonExistentPartition:该分区要么没有被创建过或曾经被创建过但后面删除了
  • NewPartition:分区创建之后已经分配了副本,但是还没有选举出Leader和ISR
  • OnlinePartition:分区Leader一旦被选举出来,就处在该状态
  • OfflinePartition:如果leader选举出来后,leader broker宕机了,那么该分区就处于OfflinePartition状态

分区状态转换流程:

override def handleStateChanges(
  partitions: Seq[TopicPartition],
  targetState: PartitionState,
  partitionLeaderElectionStrategyOpt: Option[PartitionLeaderElectionStrategy]
): Map[TopicPartition, Either[Throwable, LeaderAndIsr]] = {
  if (partitions.nonEmpty) {
    try {
      controllerBrokerRequestBatch.newBatch()
      val result = doHandleStateChanges(
        partitions,
        targetState,
        partitionLeaderElectionStrategyOpt
      )
      controllerBrokerRequestBatch.sendRequestsToBrokers(controllerContext.epoch)
      result
    } catch {
      case e: ControllerMovedException =>
        error(s"Controller moved to another broker when moving some partitions to $targetState state", e)
        throw e
      case e: Throwable =>
        error(s"Error while moving some partitions to $targetState state", e)
        partitions.iterator.map(_ -> Left(e)).toMap
    }
  } else {
    Map.empty
  }
}

5、主要处理流程

process包含了Controller的所有集群管理事件。

override def process(event: ControllerEvent): Unit = {
  try {
    event match {
      case event: MockEvent =>
        // Used only in test cases
        event.process()
      case ShutdownEventThread =>
        error("Received a ShutdownEventThread event. This type of event is supposed to be handle by ControllerEventThread")
      case AutoPreferredReplicaLeaderElection =>
        processAutoPreferredReplicaLeaderElection()
      case ReplicaLeaderElection(partitions, electionType, electionTrigger, callback) =>
        processReplicaLeaderElection(partitions, electionType, electionTrigger, callback)
      case UncleanLeaderElectionEnable =>
        processUncleanLeaderElectionEnable()
      case TopicUncleanLeaderElectionEnable(topic) =>
        processTopicUncleanLeaderElectionEnable(topic)
      case ControlledShutdown(id, brokerEpoch, callback) =>
        processControlledShutdown(id, brokerEpoch, callback)
      case LeaderAndIsrResponseReceived(response, brokerId) =>
        processLeaderAndIsrResponseReceived(response, brokerId)
      case UpdateMetadataResponseReceived(response, brokerId) =>
        processUpdateMetadataResponseReceived(response, brokerId)
      case TopicDeletionStopReplicaResponseReceived(replicaId, requestError, partitionErrors) =>
        processTopicDeletionStopReplicaResponseReceived(replicaId, requestError, partitionErrors)
      case BrokerChange =>
        processBrokerChange()
      case BrokerModifications(brokerId) =>
        processBrokerModification(brokerId)
      case ControllerChange =>
        processControllerChange()
      case Reelect =>
        processReelect()
      case RegisterBrokerAndReelect =>
        processRegisterBrokerAndReelect()
      case Expire =>
        processExpire()
      case TopicChange =>
        processTopicChange()
      case LogDirEventNotification =>
        processLogDirEventNotification()
      case PartitionModifications(topic) =>
        processPartitionModifications(topic)
      case TopicDeletion =>
        processTopicDeletion()
      case ApiPartitionReassignment(reassignments, callback) =>
        processApiPartitionReassignment(reassignments, callback)
      case ZkPartitionReassignment =>
        processZkPartitionReassignment()
      case ListPartitionReassignments(partitions, callback) =>
        processListPartitionReassignments(partitions, callback)
      case PartitionReassignmentIsrChange(partition) =>
        processPartitionReassignmentIsrChange(partition)
      case IsrChangeNotification =>
        processIsrChangeNotification()
      case Startup =>
        processStartup()
    }
  } catch {
    case e: ControllerMovedException =>
      info(s"Controller moved to another broker when processing $event.", e)
      maybeResign()
    case e: Throwable =>
      error(s"Error processing event $event", e)
  } finally {
    updateMetrics()
  }
}
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 212,080评论 6 493
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 90,422评论 3 385
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 157,630评论 0 348
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 56,554评论 1 284
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 65,662评论 6 386
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,856评论 1 290
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 39,014评论 3 408
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,752评论 0 268
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 44,212评论 1 303
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,541评论 2 327
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,687评论 1 341
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 34,347评论 4 331
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,973评论 3 315
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,777评论 0 21
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,006评论 1 266
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 46,406评论 2 360
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 43,576评论 2 349

推荐阅读更多精彩内容