Flink之用户代码生成调度层图结构

在Flink中，由用户代码生成调度层图结构，可以分成3步走：通过Stream API编写的用户代码 -> StreamGraph -> JobGraph -> ExecutionGraph。

  • StreamGraph：根据用户通过Stream API编写的代码生成的最初的图，用来表示程序的拓扑结构。
  • JobGraph：StreamGraph经过算子连接等优化后生成的图，它是提交给JobManager的数据结构。
  • ExecutionGraph：JobManager根据JobGraph生成的分布式执行图，是调度层最核心的数据结构。

以SocketWindowWordCount为例，其执行图的演变过程如下图所示：

转换演示图

详解三部曲

Step1：通过Stream API编写的用户代码 —> StreamGraph

本小节主要介绍Flink是如何根据用户用Stream API编写的程序，构造出一个代表拓扑结构的StreamGraph的。

找突破口

StreamGraph的相关代码主要在org.apache.flink.streaming.api.graph包中。主要逻辑集中在StreamGraphGenerator类，入口函数是StreamGraphGenerator.generate(env, transformations)，该函数由触发程序执行的方法StreamExecutionEnvironment.execute()调用到。

理关键点
  • 根据用户通过Stream API编写的程序，构建transformations参数。
  • 遍历transformations集合，创建StreamNode和StreamEdge，构造StreamGraph。
构建过程

StreamGraphGenerator.generate()的一个关键参数是transformations，它是env的成员变量之一，用List<StreamTransformation<? >>来保存。其中，StreamTransformation代表了从一个或多个DataStream生成新DataStream的操作。

DataStream上常见的transformation有map、flatmap、filter等。这些transformation会构造出一棵StreamTransformation树，通过这棵树转换成StreamGraph。

DataStream上的每一个transformation都对应了一个StreamOperator，StreamOperator是运行时的具体实现，会决定UDF(User-Defined Funtion)的调用方式。

以dataStream.map为例，用户编写的UDF(User-Defined Funtion)构造出StreamTransformation的过程如下图所示：

StreamTransformation

从上图可以看出，map转换，首先是将用户自定义的函数MapFunction包装到StreamMap这个StreamOperator中；然后是将StreamMap包装到OneInputTransformation中，并建立与上游的关系；最后将transformation存到env的transformations集合中。

我们看一下SocketWindowWordCount示例，其transformations树的结构如下图所示，其中符号*为input指针，指向上游的transformation，从而形成了一颗transformations树。

transformations.png

当调用env.execute()时，会触发StreamGraphGenerator.generate(env, transformations)遍历其中的transformations集合构造出StreamGraph。自底向上递归调用每一个transformation，所以真正的处理顺序是Source->Flat Map->Hash(keyBy)->TriggerWindow->Sink。

入口：

public static StreamGraph generate(StreamExecutionEnvironment env, List<StreamTransformation<?>> transformations) {
    return new StreamGraphGenerator(env).generateInternal(transformations);
}

真正的构建过程：

private StreamGraph generateInternal(List<StreamTransformation<?>> transformations) {
    for (StreamTransformation<?> transformation: transformations) {
        transform(transformation);
    }
    return streamGraph;
}

遍历transformations集合，并对其每一个StreamTransformation调用transform()方法。

private Collection<Integer> transform(StreamTransformation<?> transform) {
    ...
    if (transform instanceof OneInputTransformation<?, ?>) {
        transformedIds = transformOneInputTransform((OneInputTransformation<?, ?>) transform);
    } else if (transform instanceof TwoInputTransformation<?, ?, ?>) {
        transformedIds = transformTwoInputTransform((TwoInputTransformation<?, ?, ?>) transform);
    } else if (transform instanceof SourceTransformation<?>) {
        transformedIds = transformSource((SourceTransformation<?>) transform);
    } else if (transform instanceof SinkTransformation<?>) {
        transformedIds = transformSink((SinkTransformation<?>) transform);
    } else if (transform instanceof UnionTransformation<?>) {
        transformedIds = transformUnion((UnionTransformation<?>) transform);
    } else if (transform instanceof SplitTransformation<?>) {
        transformedIds = transformSplit((SplitTransformation<?>) transform);
    } else if (transform instanceof SelectTransformation<?>) {
        transformedIds = transformSelect((SelectTransformation<?>) transform);
    } else if (transform instanceof FeedbackTransformation<?>) {
        transformedIds = transformFeedback((FeedbackTransformation<?>) transform);
    } else if (transform instanceof CoFeedbackTransformation<?>) {
        transformedIds = transformCoFeedback((CoFeedbackTransformation<?>) transform);
    } else if (transform instanceof PartitionTransformation<?>) {
        transformedIds = transformPartition((PartitionTransformation<?>) transform);
    } else if (transform instanceof SideOutputTransformation<?>) {
        transformedIds = transformSideOutput((SideOutputTransformation<?>) transform);
    } else {
        throw new IllegalStateException("Unknown transformation: " + transform);
    }
    ...
}

从上述代码可以看出，针对具体的某一种类型的StreamTransformation，会调用其相应的transformXXX()函数进行转换。transformXXX()首先会对transform的上游transform进行递归转换，确保上游的都已经完成了转化。然后通过addOperator()方法构造出StreamNode，通过addEdge()方法与上游的transform进行连接，构造出StreamEdge。

注意：

对逻辑转换(partition、union等)的处理，不会生成具体的StreamNode和StreamEdge，而是通过streamGraph.addVirtualXXXNode()方法添加一个虚拟节点。当下游transform添加edge时，会把虚拟节点信息写入到StreamEdge中。

示例讲解：

在SocketWindowWordCount示例中，Flat Map操作会被封装到OneInputTransformation类中，我们可以看一看transformOneInputTransform的实现：

private <IN, OUT> Collection<Integer> transformOneInputTransform(OneInputTransformation<IN, OUT> transform) {
    // 递归对transformation的直接上游transformation进行转换，获取直接上游id集合
    Collection<Integer> inputIds = transform(transform.getInput());
    if (alreadyTransformed.containsKey(transform)) {
        return alreadyTransformed.get(transform);
    }
    String slotSharingGroup = determineSlotSharingGroup(transform.getSlotSharingGroup(), inputIds);
    // 构建StreamNode的入口
    streamGraph.addOperator(transform.getId(), 
            slotSharingGroup, 
            transform.getOperator(), 
            transform.getInputType(), 
            transform.getOutputType(), 
            transform.getName());
    if (transform.getStateKeySelector() != null) {
        TypeSerializer<?> keySerializer = transform.getStateKeyType().createSerializer(env.getConfig());
        streamGraph.setOneInputStateKey(transform.getId(), transform.getStateKeySelector(), keySerializer);
    }
    streamGraph.setParallelism(transform.getId(), transform.getParallelism());
    streamGraph.setMaxParallelism(transform.getId(), transform.getMaxParallelism());
    // 构建StreamEdge的入口
    for (Integer inputId: inputIds) {
        streamGraph.addEdge(inputId, transform.getId(), 0);
    }
    return Collections.singleton(transform.getId());
}

类似的，在SocketWindowWordCount示例中，keyBy操作会被封装到PartitionTransformation类中，最后我们再以transformPartition为例看下对逻辑转换的处理。

private <T> Collection<Integer> transformPartition(PartitionTransformation<T> partition) {
    StreamTransformation<T> input = partition.getInput();
    List<Integer> resultIds = new ArrayList<>();
    // 递归对该transformation的直接上游transformation进行转换，获得直接上游id集合
    Collection<Integer> transformedIds = transform(input);
    for (Integer transformedId: transformedIds) {
        int virtualId = StreamTransformation.getNewNodeId();
        // 添加一个虚拟分区节点VirtualPartitionNode，不会生成StreamNode
        streamGraph.addVirtualPartitionNode(transformedId, virtualId, partition.getPartitioner());
        resultIds.add(virtualId);
    }
  return resultIds;
}

从transformPartition函数的实现可以看出，对transformPartition的转换没有生成具体的StreamNode和StreamEdge，而是通过streamGraph.addVirtualPartitionNode()方法添加了一个虚拟节点。当partition的下游transform添加edge时（调用streamGraph.addEdge()），会把partition信息写入到StreamEdge中。

最后的最后我们再看下streamGraph.addEdgeInternal()的实现：

private void addEdgeInternal(Integer upStreamVertexID, Integer downStreamVertexID, int typeNumber, StreamPartitioner<?> partitioner, List<String> outputNames, OutputTag outputTag) {
    // 当上游是sideOutput时，递归调用，并传入sideOutput信息
    if (virtualSideOutputNodes.containsKey(upStreamVertexID)) {
        int virtualId = upStreamVertexID;
        upStreamVertexID = virtualSideOutputNodes.get(virtualId).f0;
        if (outputTag == null) {
            outputTag = virtualSideOutputNodes.get(virtualId).f1;
        }
        addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, null, outputTag);
    }
    // 当上游是select时，递归调用，并传入select信息 
    else if (virtualSelectNodes.containsKey(upStreamVertexID)) {
        int virtualId = upStreamVertexID;
        upStreamVertexID = virtualSelectNodes.get(virtualId).f0;
        if (outputNames.isEmpty()) {
            outputNames = virtualSelectNodes.get(virtualId).f1;
        }
        addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames, outputTag);
    }
    // 当上游是partition时，递归调用，并传入partitioner信息 
    else if (virtualPartitionNodes.containsKey(upStreamVertexID)) {
        int virtualId = upStreamVertexID;
        upStreamVertexID = virtualPartitionNodes.get(virtualId).f0;
        if (partitioner == null) {
            partitioner = virtualPartitionNodes.get(virtualId).f1;
        }
        addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames, outputTag);
    }
    // 不是以上逻辑转换的情况，真正构建StreamEdge 
    else {
        StreamNode upstreamNode = getStreamNode(upStreamVertexID);
        StreamNode downstreamNode = getStreamNode(downStreamVertexID);
        // 没有指定partitioner时，会为其选择forward或者rebalance分区
        if (partitioner == null && upstreamNode.getParallelism() == downstreamNode.getParallelism()) {
            partitioner = new ForwardPartitioner<Object>();
        } else if (partitioner == null) {
            partitioner = new RebalancePartitioner<Object>();
        }
        // 创建StreamEdge，并将该StreamEdge添加到上游的输出，下游的输入
        StreamEdge edge = new StreamEdge(upstreamNode, downstreamNode, typeNumber, outputNames, partitioner, outputTag);
        getStreamNode(edge.getSourceId()).addOutEdge(edge);
        getStreamNode(edge.getTargetId()).addInEdge(edge);
    }
}

一句话总结：
首先，根据用户通过Stream API自定义UDF编写的程序，即在DataStream上做的一系列转换（map、shuffle、window等），我们可以得到StreamTransformation集合。然后通过调用streamGraphGenerator.generate(env, transformations)，遍历transformations集合，并对其每一个StreamTransformation调用transform()方法，构造出StreamNode，并通过StreamEdge与上游的transformation进行连接，此处需要特别注意对逻辑转换（partition等）的处理，最后构造出StreamGraph。

Step2：StreamGraph —> JobGraph

本小节主要介绍Flink是如何将StreamGraph转换成JobGraph的。该转换的关键在于，将多个符合条件的StreamNode节点chain在一起作为一个JobVertex节点，这样可以减少数据在节点之间流动所需要的序列化/反序列化/传输消耗。

找突破口

JobGraph的相关数据结构主要在flink-runtime模块下的org.apache.flink.runtime.jobgraph包中。构造JobGraph的代码主要集中在StreamingJobGraphGenerator类中，开启构建之路的入口函数是StreamingJobGraphGenerator(streamGraph).createJobGraph()。

理关键点

判断算子chain，合并创建JobVertex，并生成JobEdge。JobVertex和JobEdge之间通过创建IntermediateDataSet来连接。可以简单分为3个关键点：

  • chain的判断
  • 生成JobVertex
  • 创建JobEdge和IntermediateDataSet
构建过程

入口：

private JobGraph createJobGraph() {

    ...
    setChaining();  //递归创建JobVertex、JobEdge、IntermediateDataSet，用以构建JobGraph
    ...
    return jobGraph;
}

真正的构建过程：

void setChaining(){
    for (Integer sourceNodeId : streamGraph.getSourceIDs()) {
        createChain(sourceNodeId, sourceNodeId, ...);
    }
}


private List<StreamEdge> createChain(Integer startNodeId, Integer currentNodeId, ...){
    ...
    // 将当前节点的出边分为两类：chainableOutputs和nonChainableOutputs
    for (StreamEdge outEdge : streamGraph.getStreamNode(currentNodeId).getOutEdges()) {
        if (isChainable(outEdge, streamGraph)) {
            chainableOutputs.add(outEdge);
        } else {
            nonChainableOutputs.add(outEdge);
        }
    }
    // 分别遍历chainableOutputs和nonChainableOutputs，递归调用自身方法creatChain()
    // 并将nonChainableOutputs的边或者chainableOutputs调用createChain()的返回值添加到transitiveOutEdges中
    for (StreamEdge chainable : chainableOutputs) {
        transitiveOutEdges.addAll(createChain(startNodeId, chainable.getTargetId(), ...));
    }
    for (StreamEdge nonChainable : nonChainableOutputs) {
        transitiveOutEdges.add(nonChainable);
        createChain(nonChainable.getTargetId(), nonChainable.getTargetId(), ...);
    }
    // 如果当前节点是起始节点，则直接创建JobVertex，否则返回一个空的SteamConfig
    StreamConfig config = currentNodeId.equals(startNodeId)
        ? createJobVertex(startNodeId, ...) : new StreamConfig(new Configuration());
    // 将StreamNode中的配置信息序列化到StreamConfig中
    setVertexConfig(currentNodeId, config, chainableOutputs, nonChainableOutputs);
    // 再次判断，如果是chain的起始节点，执行connect()方法，创建JobEdge和IntermediateDataSet；否则将当前节点的StreamConfig添加到chainedConfig中
    if (currentNodeId.equals(startNodeId)) {
        for (StreamEdge edge : transitiveOutEdges) {
            connect(startNodeId, edge);
        }
    } else {
        chainedConfigs.get(startNodeId).put(currentNodeId, config);
    }
    ...
    return transitiveOutEdges;
}

setChaining()会依次对source调用createChain()方法，该方法会递归调用其下游节点，从而构建出node chains。createChain()会分析当前节点的出边，根据Operator Chains中的条件进行判断isChainable()，并将出边分成两类：chainalbeOutputs和noChainableOutputs，接着分别递归调用自身方法。之后会将StreamNode中的配置信息序列化到StreamConfig中。如果当前不是chain中的子节点，则会构建JobVertex和JobEdge相连。如果是chain中的子节点，则会将StreamConfig添加到该chain的chainedConfigs集合中。

示例讲解：

同样地，以SocketWindowWordCount为例，我们分析下其创建过程：

StreamGraph

如上图所示，我们先给4个StreamNode节点进行编号，Source用1表示，Flat Map用2表示，Trigger Window用3表示，Sink用4表示；相应地，3条StreamEdge则分别用1->2，2->3，3->4表示。

递归调用过程如下：

  • 递归始于Source，调用createChain(1, 1)，当前节点1的出边为1->2，不可chain，将边1->2直接加入transitiveOutEdges；
  • 然后递归调用createChain(2, 2)，当前节点2的出边为2->3，同样的，不可chain，将边2->3直接加入transitiveOutEdges；
  • 继续递归调用createChain(3, 3)，当前节点3的出边为3->4，要注意了，可chain，等着将下游createChain()的返回值加入transitiveOutEdges；
  • 此处递归调用createChain(3, 4)，当前节点4没有出边，递归终止。

递归结束条件：

  • 当前节点不再有出边集合，即streamGraph.getStreamNode(currentId).getOutEdges()为空
  • 当前节点已经转换完成，即builtVertices.contains(startNodeId)为false

递归调用过程中各种操作以及变量情况一览表如下：

creatChain() getOutEdges() chainAble nonChainAble transitiveOutEdges JobVertex connect()
(1, 1) 1->2 无 1->2 1->2 JobVertex Y
(2, 2) 2->3 无 2->3 2->3 JobVertex Y
(3, 3) 3->4 3->4 无 无 JobVertex Y
(3, 4) 无 无 无 无 StreamConfig N

关键方法分析：

isChainable()，用来判断StreamNode chain，一共有9个条件：

  • 下游节点的入边为1
  • StreamEdge的下游节点对应的算子不为null
  • StreamEdge的上游节点对应的算子不为null
  • StreamEdge的上下游节点拥有相同的slotSharingGroup，默认都是default
  • 下游算子的连接策略为ALWAYS
  • 上游算子的连接策略为ALWAYS或者HEAD
  • StreamEdge的分区类型为ForwardPartitioner
  • 上下游节点的并行度一致
  • 当前StreamGraph允许做chain

createJobVertex()，用来创建JobVertex节点，并返回StreamConfig。

createJobVertex()传入的参数为StreamNode。首先会通过new JobVertex()构造出JobVertex节点，然后通过JobVertex.setInvokableClass(streamNode.getJobVertexClass())设置运行时执行类，再通过jobVertex.setParallelism(parallelism)设置并行度，最后返回StreamConfig。

connect()，用来创建JobEdge和IntermediateDataSet，连接上下游JobVertex节点。

遍历transitiveOutEdges，并将每一条StreamEdge边作为参数传入connect( )函数中。接下来就是依据StreamEdge得到上下游JobVertex节点；然后，通过StreamEdge.getPartitioner()方法得到StreamPartitioner属性，对于ForwardPartitioner和RescalePartitioner两种分区方式建立DistributionPattern.POINTWISE类型的JobEdge和IntermediateDataSet，而其他的分区方式则是DistributionPattern.ALL_TO_ALL类型。至此已经建立好上下游JobVertex节点间的联系。

一句话总结：

首先，通过streamGraph.getSourceIDs()拿到source节点集合，紧接着依次从source节点开始遍历，判断StreamNode Chain，递归创建JobVertex，所以，其真正的处理顺序其实是从sink开始的。然后通过connect()遍历当前节点的物理出边transitiveOutEdges集合，创建JobEdge，建立当前节点与下游节点的联系，即JobVertex与IntermediateDataSet之间。

Step3：JobGraph —> ExecutionGraph

本小节主要介绍Flink是如何将JobGraph转换成ExecutionGraph的。简单来说，就是并行化JobGraph，为调度做好准备。

找突破口

ExecutionGraph的相关数据结构主要在org.apache.flink.runtime.executiongraph包中，构造ExecutionGraph的代码集中在ExecutionGraphBuilder类和ExecutionGraph类中，入口函数是ExecutionGraphBuilder.buildGraph(executionGraph, jobGraph, ...)。

理关键点
  • 客户端提交JobGraph给JobManager
  • 构建ExecutionGraph对象
    • 将JobGraph进行拓扑排序，得到sortedTopology顶点集合
    • 将JobVertex封装成ExecutionJobVertex
    • 把ExecutionVertex节点通过ExecutionEdge连接起来
构建过程

1、JobClient提交JobGraph给JobManager

一个程序的JobGraph真正被提交始于对JobClient的submitJobAndWait()方法的调用，而且submitJobAndWait()方法会触发基于Akka的Actor之间的消息通信。JobClient在这其中起到了“桥接”的作用，它连接了同步的方法调用和异步的消息通信。

在submitJobAndWait()方法中，首先会创建一个JobClientActor的ActorRef，并向其发送一个包含JobGraph实例的SubmitJobAndWait消息。该SubmitJobAndWait消息被JobClientActor接收后，调用trySubmitJob()方法触发真正的提交动作，即通过jobManager.tell( )的方式给JobManager Actor发送封装JobGraph的SubmitJob消息。随后，JobManager Actor会接收到来自JobClientActor的该SubmitJob消息，进而触发submitJob()方法。

由此可见，一个JobGraph从提交开始会经过多个对象层层递交，各个对象之间的交互关系如下图所示：

Actor交互

2、构建ExecutionGraph对象

入口：
JobManager作为Actor，在handleMessage()方法中，针对SubmitJob消息调用submitJob()方法。

private def submitJob(jobGraph: JobGraph, jobInfo: JobInfo, isRecovery: Boolean = false): Unit = {
    ...
    executionGraph = ExecutionGraphBuilder.buildGraph(executionGraph, jobGraph, ...)
    ...
}

真正的构建过程：

public static ExecutionGraph buildGraph(){
    ...
    //对JobGraph中的JobVertex节点进行拓扑排序，得到List<JobVertex>
    List<JobVertex> sortedTopology = jobGraph.getVerticesSortedTopologicallyFromSources();
    executionGraph.attachJobGraph(sortedTopology);  //构建ExecutionGraph的核心方法
    ...
}

由上可知，attachJobGraph()方法是构建ExecutionGraph图结构的核心方法。

public void attachJobGraph(List<JobVertex> topologiallySorted){
    ...
    for (JobVertex jobVertex : topologiallySorted) {
        ...
        // create the execution job vertex and attach it to the graph
        ExecutionJobVertex ejv = new ExecutionJobVertex(this, jobVertex, 1, rpcCallTimeout, globalModVersion, createTimestamp);
        ejv.connectToPredecessors(this.intermediateResults);
        ...
    }
    ...
}

下面详细分析下，attachJobGraph()方法主要完成的两件事情：

  • 将JobVertex封装成ExecutionJobVertex
  • 把节点通过ExecutionEdge连接

关键方法分析：

new ExecutionJobVertex()方法，用来将一个个JobVertex封装成ExecutionJobVertex，并依次创建ExecutionVertex、Execution、IntermediateResult、IntermediateResultPartition，用于丰富ExecutionGraph。

在ExecutionJobVertex的构造函数中，首先是依据对应的JobVertex的并发度，生成对应个数的ExecutionVertex。其中，一个ExecutionVertex代表着一个ExecutionJobVertex的并发子task。然后是将原来JobVertex的中间结果IntermediateDataSet转化为ExecutionGraph中的IntermediateResult。

类似的，ExecutionVertex的构造函数中，首先会创建IntermediateResultPartition，并通过IntermediateResult.setPartition( )建立IntermediateResult和IntermediateResultPartition之间的关系；然后生成Execution，并配置资源相关。

新创建的ExecutionJobVertex调用ejv.connectToPredecessor()方法，按照不同的分发策略连接上游，其参数为上游生成的IntermediateResult集合。其中，根据JobEdge中两种不同的DistributionPattern属性，分别调用connectPointWise()或者connectAllToAll( )方法，创建ExecutionEdge，将ExecutionVertex和上游的IntermediateResultPartition连接起来。

其中，SocketWindowWordCount示例中，就是采用了connectAllToAll()的方式建立与上游的关系。

接下来，我们详细介绍下connectPointWise()方法的实现，即DistributionPattern.POINTWISE策略，该策略用来连接当前ExecutionVertex与上游的IntermediateResultPartition。首先，获取上游IntermediateResult的partition数，用numSources表示，以及此ExecutionJobVertex的并发度，用parallelism表示；然后，根据其并行度的不同，分别创建ExecutionEdge。共分3种情况：

(1) 如果并发数等于partition数，则一对一进行连接。如下图所示：
即numSources == parallelism

OneToOne

(2) 如果并发数大于partition数，则一对多进行连接。如下图所示：
即numSources < parallelism，且parallelism % numSources == 0

OneToMany-1

即numSources < parallelism，且parallelism % numSources != 0
OneToMany-2

(3) 如果并发数小于partition数，则多对一进行连接。如下图所示：
即numSources > parallelism，且numSources % parallelism == 0

ManyToOne-1

即numSources > parallelism，且numSources % parallelism != 0
ManyToOne-2

一句话总结：

将JobGraph按照拓扑排序后得到一个JobVertex集合，遍历该JobVertex集合，即从source开始，将JobVertex封装成ExecutionJobVertex，并依次创建ExecutionVertex、Execution、IntermediateResult和IntermediateResultPartition。然后通过ejv.connectToPredecessor()方法，创建ExecutionEdge，建立当前节点与其上游节点之间的联系，即连接ExecutionVertex和IntermediateResultPartition。

终章

构建好ExecutionGraph，接下来会基于ExecutionGraph触发Job的调度，申请Slot，真正的部署任务，这是Task被执行的前提：

if (leaderElectionService.hasLeadership) {
    log.info(s"Scheduling job $jobId ($jobName).")
    executionGraph.scheduleForExecution()  // 将生成好的ExecutionGraph进行调度
} else {
    self ! decorateMessage(RemoveJob(jobId, removeJobFromStateBackend = false))
    log.warn(s"Submitted job $jobId, but not leader. The other leader needs to recover " +
              "this. I am not scheduling the job for execution.")
}

下一篇文章将介绍Flink的Schedule机制，敬请期待~

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 206,723评论 6赞 481
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 88,485评论 2赞 382
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 152,998评论 0赞 344
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 55,323评论 1赞 279
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 64,355评论 5赞 374
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,079评论 1赞 285
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,389评论 3赞 400
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,019评论 0赞 259
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 43,519评论 1赞 300
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,971评论 2赞 325
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,100评论 1赞 333
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,738评论 4赞 324
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,293评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,289评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,517评论 1赞 262
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,547评论 2赞 354
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,834评论 2赞 345