Scala深海奇遇记-当case class遇到了Spark的聚集函数

自从知道有case class这个东西以后，一直都比较常用这个东西。但是，最近在测试的时候，突然发现，其实这个东西并不简单，它导致了一个看起来很无厘头的错误，并且花了我两天的时间来调试。

在这篇文章里，我会详细记录调试的过程，以及结论。

致谢

在调试的过程中，得到了我们Hadoop组老大，项目组老大，以及其他同事的深度支持与帮助，非常感谢他们。

结论

先说结论。如果有朋友不感兴趣，不想深究原理，只是想知道怎么用，可以跳过后面的部分。

不要把case class放在class里面。例如:

class Test {

  case class A(a: String)

}

可以放到object里面，或者放到package里面。例如:

object Test {

  case class A(a: String)

}

或者

package Test {

  case class A(a: String)

}

如果不遵循这条原则，那么有极大的概率，Spark在执行的时候，相同的key并不会聚集到一起。

分析

首先，我们有如下代码:

package org.apache.spark.test

import org.apache.spark.{Partitioner, SparkConf, SparkContext}

import scala.util.Random

object VerifySparkBug {

  def main(args: Array[String]): Unit = {
    new VerifySparkBug().run()
  }

}

class VerifySparkBug extends Serializable {

  def run() = {

    val sparkConf = new SparkConf()
    sparkConf.setMaster("local")
    sparkConf.setAppName("VerifySparkBug")
    val sparkContext = new SparkContext(sparkConf)

    val inputRDD = sparkContext.parallelize(Seq(
      (UidAndPartition("1", "2"), 1L),
      (UidAndPartition("1", "1"), 1L),
      (UidAndPartition("1", "3"), 1L),
      (UidAndPartition("2", "1"), 1L),
      (UidAndPartition("2", "2"), 1L),
      (UidAndPartition("3", "3"), 1L)
    )).partitionBy(new RandomPartitioner(10))

    inputRDD.foreachPartition {
      case iterator => {
        val partitionId = new Random().nextInt()
        while (iterator.hasNext) {
          val item = iterator.next()
          System.out.println("------ inputRDD partitionId: " + partitionId
            + ", key: " + item._1
            + ", hash:" + item._1.hashCode()
            + ",value: " + item._2)
        }
      }
    }

    val uidRDD = inputRDD.map{
      case pair => {
        (Uid(pair._1.uid), pair._2)
      }
    }

    uidRDD.foreachPartition {
      case iterator => {
        val partitionId = new Random().nextInt()
        while (iterator.hasNext) {
          val item = iterator.next()
          System.out.println("------ uidRDD partitionId: " + partitionId
            + ", key: " + item._1
            + ", hash:" + item._1.hashCode()
            + ",value: " + item._2)
        }
      }
    }

    val result = uidRDD.aggregateByKey(0L)(
      (counter: Long, number: Long) => {
        counter + number
      },
      (counter1: Long, counter2: Long) => {
        counter1 + counter2
      }
    ).collect()

    for (pair <- result) {
      System.out.println("------ key: " + pair._1 + ", uv: " + pair._2)
    }

  }

  case class UidAndPartition(uid: String, partition: String)

  case class Uid(idType: String)
}

class RandomPartitioner(partitions: Int) extends Partitioner {
  override def numPartitions: Int = partitions

  override def getPartition(key: Any): Int = {
    new Random().nextInt(partitions)
  }
}

它的输出如下:

------ key: Uid(1), uv: 1
------ key: Uid(1), uv: 1
------ key: Uid(1), uv: 1
------ key: Uid(3), uv: 1
------ key: Uid(2), uv: 1
------ key: Uid(2), uv: 1

很诡异，对吧？明明key相同，却并没有被聚合到一起。而且，更诡异的是，key相同的hash值也相同。

猜想1

开始，我们怀疑是case class虽然默认实现了Serializable接口，但是并没有生成serialVersionUID，所以在序列化以及反序列化时，生成的类其实并不是同一个。导致反序列化出来的对象不是同一个。

反编译后的Uid的部分代码如下:

于是，我就用Netty写了一个服务器端和客户端的程序，来测试这个问题:

服务器端:

package org.apache.spark.test

import java.io._

import io.netty.bootstrap.ServerBootstrap
import io.netty.buffer.ByteBuf
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter, ChannelInitializer, SimpleChannelInboundHandler}
import io.netty.channel.nio.NioEventLoopGroup
import io.netty.channel.socket.nio.{NioServerSocketChannel, NioSocketChannel}
import org.apache.spark.test.VerifySparkBug.Uid

object ServerForCaseClassEquals {

  def main(args: Array[String]): Unit = {

    val serverBootstrap = new ServerBootstrap()
    val bossEventLoopGroup = new NioEventLoopGroup()
    val workerEventLoopGroup = new NioEventLoopGroup()

    serverBootstrap.group(bossEventLoopGroup, workerEventLoopGroup)
      .channel(classOf[NioServerSocketChannel])
      .childHandler(new ChannelInitializer[NioSocketChannel] {
        override def initChannel(c: NioSocketChannel) = {
          c.pipeline().addLast(new ChannelInboundHandlerAdapter {
            override def channelRead(ctx: ChannelHandlerContext, msg: scala.Any): Unit = {
              val byteBuf = msg.asInstanceOf[ByteBuf]
              val length = byteBuf.readInt()
              val bytes = new Array[Byte](length)
              byteBuf.readBytes(bytes)

              val byteInputStream = new ByteArrayInputStream(bytes)
              val objectInputStream = new ObjectInputStream(byteInputStream)

              println(objectInputStream.readObject() == Uid("a"))

            }
          })
        }
      })
      .bind(8000)

  }

}

客户端:

package org.apache.spark.test

import java.io.{ByteArrayOutputStream, ObjectOutputStream}

import io.netty.bootstrap.Bootstrap
import io.netty.channel.nio.NioEventLoopGroup
import io.netty.channel.socket.nio.NioSocketChannel
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter, ChannelInitializer}
import org.apache.spark.test.VerifySparkBug.Uid

object ClientForCaseClassEquals {

  def main(args: Array[String]): Unit = {

    val bootstrap = new Bootstrap()
    val eventLoopGroup = new NioEventLoopGroup()

    bootstrap.group(eventLoopGroup)
      .channel(classOf[NioSocketChannel])
      .handler(new ChannelInitializer[NioSocketChannel] {
        override def initChannel(c: NioSocketChannel) = {
          c.pipeline().addLast(new ChannelInboundHandlerAdapter {
            override def channelActive(ctx: ChannelHandlerContext): Unit = {
              val uid = Uid("a")
              val byteArrayOutputStream = new ByteArrayOutputStream()
              val objectOutputStream = new ObjectOutputStream(byteArrayOutputStream)

              objectOutputStream.writeObject(uid)

              objectOutputStream.close()

              val bytes = byteArrayOutputStream.toByteArray

              val byteBuf = ctx.alloc().buffer()
              byteBuf.writeInt(bytes.length)
              byteBuf.writeBytes(bytes)

              ctx.channel().writeAndFlush(byteBuf)
            }
          })
        }
      })
      .connect("localhost", 8000)

  }

}

而这个测试自然失败了。测试的结果是，反序列化解析出来的对象，是==反序列化之前的对象的。

为什么用==？因为在scala中，==其实就是equals()方法的语法糖。而且，spark的代码中，聚合函数中，对key进行比较时，也是用的==。

其实这种假设根本就经不起推敲。

如果序列化和反序列的时候，类不是一个，即serialVersionUID不同，那么，其实反序列化是不会成功的，它是会直接报错的。而我们的程序中并没有报错。说明其实是同一个。

那么既然case class在生成的时候，并没有生成serialVersionUID。那我们怎么能够确定，在运行的时候，即使是在不同的节点，不同的JVM上，它们就是同一个呢？这个问题，在Java Specification中的序列化和反序列化部分有详细描述。总之，就是根据类的相关信息，生成一个serialVersionUID。由于这个信息跟内存地址无关，所以即使是运行时生成，也是唯一且相同的。

猜想2

这个假设给推掉了。为了复现这个问题，我写了如下代码，打算来进行复现。上面的代码是最终版的复现版本。在下面的代码中，虽然问题看似得到了复现，实际上是完全不同的问题。

package org.apache.spark.test

import org.apache.spark.{Partitioner, SparkConf, SparkContext}

import scala.util.Random

object VerifySparkBug {

  def main(args: Array[String]): Unit = {
    new VerifySparkBug().run()
  }

}

class VerifySparkBug extends Serializable {

  def run() = {

    val sparkConf = new SparkConf()
    sparkConf.setMaster("local")
    sparkConf.setAppName("VerifySparkBug")
    val sparkContext = new SparkContext(sparkConf)

    val inputRDD = sparkContext.parallelize(Seq(
      (UidAndPartition("1", "2"), 1L),
      (UidAndPartition("1", "1"), 1L),
      (UidAndPartition("1", "3"), 1L),
      (UidAndPartition("2", "1"), 1L),
      (UidAndPartition("2", "2"), 1L),
      (UidAndPartition("3", "3"), 1L)
    ))

    inputRDD.foreachPartition {
      case iterator => {
        val partitionId = new Random().nextInt()
        while (iterator.hasNext) {
          val item = iterator.next()
          System.out.println("------ inputRDD partitionId: " + partitionId
            + ", key: " + item._1
            + ", hash:" + item._1.hashCode()
            + ",value: " + item._2)
        }
      }
    }

    val uidRDD = inputRDD.map{
      case pair => {
        (Uid(pair._1.uid), pair._2)
      }
    }.partitionBy(new RandomPartitioner(10))

    uidRDD.foreachPartition {
      case iterator => {
        val partitionId = new Random().nextInt()
        while (iterator.hasNext) {
          val item = iterator.next()
          System.out.println("------ uidRDD partitionId: " + partitionId
            + ", key: " + item._1
            + ", hash:" + item._1.hashCode()
            + ",value: " + item._2)
        }
      }
    }

    val result = uidRDD.aggregateByKey(0L)(
      (counter: Long, number: Long) => {
        counter + number
      },
      (counter1: Long, counter2: Long) => {
        counter1 + counter2
      }
    ).collect()

    for (pair <- result) {
      System.out.println("------ key: " + pair._1 + ", uv: " + pair._2)
    }

  }

  case class UidAndPartition(uid: String, partition: String)

  case class Uid(idType: String)
}

class RandomPartitioner(partitions: Int) extends Partitioner {
  override def numPartitions: Int = partitions

  override def getPartition(key: Any): Int = {
    new Random().nextInt(partitions)
  }
}

代码跟上面的很相似，只是partitionBy()的位置变了一下。

初衷很简单，因为我们并不能保证相同的Uid会在同一个partition，于是就通过RandomPartitioner来模拟随机分区的效果。

这段代码的问题是，如果在进行聚集(如aggregateByKey, reduceByKey等)之前，你有进行过partitionBy操作，那么，Spark会认为，你已经将相同的Key放到一个分区了，所以它在进行聚集的时候，是不会考虑聚集不同partition上，相同的key的情况的。

我们通过调试聚集函数通用的PairRDDFunctions.combineByKeyWithClassTag方法，就能看到这个结论:

我们看到，如果聚集函数前面刚好调用了partitionBy方法，聚集函数内部是不会调用partitionBy来将相同的Key分到一个分区的。

这是因为partitionBy以及聚集函数，其实都是ShuffleDependency。Spark内部做了优化，如果ShuffleDependency前面已经是Narrow Dependency了，那么就会把这个ShuffleDependency转换成Narrow Dependency。

所以上面看似正确的复现，实际上刚好走进了这个陷阱里面。

那么，上面的代码，如果考虑到这个问题，该怎么解决？一种方案是，通过HashPartitioner，将相同的key确实分到同一个分区里面去。另一种方法就是，在上面的代码中的partitionBy(new RandomPartitioner(10)) 后面，加一个map(r => r)。 为什么加个这个就会生效?

我们先看map() 函数，以及MapPartitionRDD的代码。

我们可以看到,map()函数，实际上是生成了一个没有partitioner的MapPartitionRDD。就是这个没有partitioner，帮了我们的大忙。没有partitioner，聚集函数就不知道前面是Narrow Dependency，所以就会考虑相同的key在不同分区的问题。

下面这张图片，是我采用第二种方式，调试的过程。可以看到，现在走的是ShuffleRDD。

然后，我们可以看到，丫的，相同的key还是没有被聚集到一起。

其实当时在写这个复现的Demo的时候，是聚到一起了，一度手舞足蹈。但是聚到一起的原因，是case class是放在Object里面了，没有放到class里，所以并没有正确的复现问题。

猜想3

采用猜想2中的第1种方法，我们把不同分区中，相同的Key放到同一个分区里，有如下代码:

package org.apache.spark.test

import org.apache.spark.{HashPartitioner, Partitioner, SparkConf, SparkContext}

import scala.util.Random

object VerifySparkBug {

  def main(args: Array[String]): Unit = {
    new VerifySparkBug().run()
  }

}

class VerifySparkBug extends Serializable {

  def run() = {

    val sparkConf = new SparkConf()
    sparkConf.setMaster("local")
    sparkConf.setAppName("VerifySparkBug")
    val sparkContext = new SparkContext(sparkConf)

    val inputRDD = sparkContext.parallelize(Seq(
      (UidAndPartition("1", "2"), 1L),
      (UidAndPartition("1", "1"), 1L),
      (UidAndPartition("1", "3"), 1L),
      (UidAndPartition("2", "1"), 1L),
      (UidAndPartition("2", "2"), 1L),
      (UidAndPartition("3", "3"), 1L)
    ))

    inputRDD.foreachPartition {
      case iterator => {
        val partitionId = new Random().nextInt()
        while (iterator.hasNext) {
          val item = iterator.next()
          System.out.println("------ inputRDD partitionId: " + partitionId
            + ", key: " + item._1
            + ", hash:" + item._1.hashCode()
            + ",value: " + item._2)
        }
      }
    }

    val uidRDD = inputRDD.map{
      case pair => {
        (Uid(pair._1.uid), pair._2)
      }
    }.partitionBy(new RandomPartitioner(10)).partitionBy(new HashPartitioner(10))

    uidRDD.foreachPartition {
      case iterator => {
        val partitionId = new Random().nextInt()
        while (iterator.hasNext) {
          val item = iterator.next()
          System.out.println("------ uidRDD partitionId: " + partitionId
            + ", key: " + item._1
            + ", hash:" + item._1.hashCode()
            + ",value: " + item._2)
        }
      }
    }

    val result = uidRDD.aggregateByKey(0L)(
      (counter: Long, number: Long) => {
        counter + number
      },
      (counter1: Long, counter2: Long) => {
        counter1 + counter2
      }
    ).collect()

    for (pair <- result) {
      System.out.println("------ key: " + pair._1 + ", uv: " + pair._2)
    }

  }

  case class UidAndPartition(uid: String, partition: String)

  case class Uid(idType: String)
}

class RandomPartitioner(partitions: Int) extends Partitioner {
  override def numPartitions: Int = partitions

  override def getPartition(key: Any): Int = {
    new Random().nextInt(partitions)
  }
}

其实这些代码差距很小，用文本比较工具，很容易看出差距。

在运行这个过程中，我们发现，丫的，把它们放到同一个分区，还是不能聚到一起？太他么过分了。

于是，我们查看聚集函数中，是如何比较key的。

我们可以看到，是调用equals方法。

于是，我们就想，case class的equals方法，到底长什么样子？

这里有一点很坑，就是你看反编译出来的代码，在这里是会反编译错误的。你必须找其他的方式来看。我们采用的方式，是看Scala编译的中间结果。

运行scalac -Xprint:typer VerifySparkBug.scala，我们能看到，上面的代码中，Uid的equals()方法，scala会被编译成这种形式:

override <synthetic> def equals(x$1: Any): Boolean = Uid.this.eq(x$1.asInstanceOf[Object]).||(x$1 match {
  case (_: VerifySparkBug.this.Uid) => true
  case _ => false
}.&&({
        <synthetic> val Uid$1: VerifySparkBug.this.Uid = x$1.asInstanceOf[VerifySparkBug.this.Uid];
        Uid.this.idType.==(Uid$1.idType).&&(Uid$1.canEqual(Uid.this))
      }))
    };

我们再来看看，把Uid放到Object里面，生成的equals方法是什么样子。

代码如下:

package org.apache.spark.test

import org.apache.spark.test.VerifySparkBug.{Uid, UidAndPartition}
import org.apache.spark.{HashPartitioner, Partitioner, SparkConf, SparkContext}

import scala.util.Random

object VerifySparkBug {

  def main(args: Array[String]): Unit = {
    new VerifySparkBug().run()
  }

  case class UidAndPartition(uid: String, partition: String)

  case class Uid(idType: String)
}

class VerifySparkBug extends Serializable {

  def run() = {

    val sparkConf = new SparkConf()
    sparkConf.setMaster("local")
    sparkConf.setAppName("VerifySparkBug")
    val sparkContext = new SparkContext(sparkConf)

    val inputRDD = sparkContext.parallelize(Seq(
      (UidAndPartition("1", "2"), 1L),
      (UidAndPartition("1", "1"), 1L),
      (UidAndPartition("1", "3"), 1L),
      (UidAndPartition("2", "1"), 1L),
      (UidAndPartition("2", "2"), 1L),
      (UidAndPartition("3", "3"), 1L)
    ))

    inputRDD.foreachPartition {
      case iterator => {
        val partitionId = new Random().nextInt()
        while (iterator.hasNext) {
          val item = iterator.next()
          System.out.println("------ inputRDD partitionId: " + partitionId
            + ", key: " + item._1
            + ", hash:" + item._1.hashCode()
            + ",value: " + item._2)
        }
      }
    }

    val uidRDD = inputRDD.map{
      case pair => {
        (Uid(pair._1.uid), pair._2)
      }
    }.partitionBy(new RandomPartitioner(10)).partitionBy(new HashPartitioner(10))

    uidRDD.foreachPartition {
      case iterator => {
        val partitionId = new Random().nextInt()
        while (iterator.hasNext) {
          val item = iterator.next()
          System.out.println("------ uidRDD partitionId: " + partitionId
            + ", key: " + item._1
            + ", hash:" + item._1.hashCode()
            + ",value: " + item._2)
        }
      }
    }

    val result = uidRDD.aggregateByKey(0L)(
      (counter: Long, number: Long) => {
        counter + number
      },
      (counter1: Long, counter2: Long) => {
        counter1 + counter2
      }
    ).collect()

    for (pair <- result) {
      System.out.println("------ key: " + pair._1 + ", uv: " + pair._2)
    }

  }

}

class RandomPartitioner(partitions: Int) extends Partitioner {
  override def numPartitions: Int = partitions

  override def getPartition(key: Any): Int = {
    new Random().nextInt(partitions)
  }
}

生成的equals()方法如下:

override <synthetic> def equals(x$1: Any): Boolean = Uid.this.eq(x$1.asInstanceOf[Object]).||(x$1 match {
  case (_: org.apache.spark.test.VerifySparkBug.Uid) => true
  case _ => false
}.&&({
        <synthetic> val Uid$1: org.apache.spark.test.VerifySparkBug.Uid = x$1.asInstanceOf[org.apache.spark.test.VerifySparkBug.Uid];
        Uid.this.idType.==(Uid$1.idType).&&(Uid$1.canEqual(Uid.this))
      }))
    };

我们可以看到，它们的区别，就在于放在class中的Uid,调用的是VerifySparkBug.this.Uid。而放在object中的Uid,调用的是com.hyper.cdp.label.spark.test.VerifySparkBug.Uid。后者很容易理解，它肯定是唯一且相同的。但是前者呢？它包含了一个VerifySparkBug.this啊。

那么什么是VerifySparkBug.this呢？我也不清楚。但是我写了一个Demo，验证了一个事实，对于每一个VerifySparkBug实例，它们的值都是不同的。

代码如下:

package org.apache.spark.test

class VerifyCaseClassInClass {

    case class Test(a: String)

    def test() = {
        println(VerifyCaseClassInClass.this)
    }
}

object VerifyCaseClassInClass {

    def main(args: Array[String]): Unit = {

        println(new VerifyCaseClassInClass().Test("a").equals(new VerifyCaseClassInClass().Test("a")))

        new VerifyCaseClassInClass().test()
        new VerifyCaseClassInClass().test()

    }

}

这就很容易解释为什么case class在Class中时，即使Key相同，仍然不会不会聚到一起了。

在Shuffle的时候，对于收到的其他分区的Key，肯定是为每个分区都new一个VerifySparkBug。这样就导致看起来相同的key，即使它们的hash值也一样，但是就是聚不到一起。

case class的hashCode()方法倒是很中立，它跟内存地址等都无关。只要字段的值一样，生成的hash就相同。感兴趣的读者可以自行查看scala.runtime.ScalaRunTime._hashCode这个方法。

这次调试，也深刻验证了，两个对象的hashcode相同，它们却不一定equals这一个简单却并没有重视过的问题。

最终，能够正常工作的代码如下:

package org.apache.spark.test

import org.apache.spark.test.VerifySparkBug.{Uid, UidAndPartition}
import org.apache.spark.{HashPartitioner, Partitioner, SparkConf, SparkContext}

import scala.util.Random

object VerifySparkBug {

  def main(args: Array[String]): Unit = {
    new VerifySparkBug().run()
  }

  case class UidAndPartition(uid: String, partition: String)

  case class Uid(idType: String)
}

class VerifySparkBug extends Serializable {

  def run() = {

    val sparkConf = new SparkConf()
    sparkConf.setMaster("local")
    sparkConf.setAppName("VerifySparkBug")
    val sparkContext = new SparkContext(sparkConf)

    val inputRDD = sparkContext.parallelize(Seq(
      (UidAndPartition("1", "2"), 1L),
      (UidAndPartition("1", "1"), 1L),
      (UidAndPartition("1", "3"), 1L),
      (UidAndPartition("2", "1"), 1L),
      (UidAndPartition("2", "2"), 1L),
      (UidAndPartition("3", "3"), 1L)
    ))

    inputRDD.foreachPartition {
      case iterator => {
        val partitionId = new Random().nextInt()
        while (iterator.hasNext) {
          val item = iterator.next()
          System.out.println("------ inputRDD partitionId: " + partitionId
            + ", key: " + item._1
            + ", hash:" + item._1.hashCode()
            + ",value: " + item._2)
        }
      }
    }

    val uidRDD = inputRDD.map{
      case pair => {
        (Uid(pair._1.uid), pair._2)
      }
    }.partitionBy(new RandomPartitioner(10)).partitionBy(new HashPartitioner(10))

    uidRDD.foreachPartition {
      case iterator => {
        val partitionId = new Random().nextInt()
        while (iterator.hasNext) {
          val item = iterator.next()
          System.out.println("------ uidRDD partitionId: " + partitionId
            + ", key: " + item._1
            + ", hash:" + item._1.hashCode()
            + ",value: " + item._2)
        }
      }
    }

    val result = uidRDD.aggregateByKey(0L)(
      (counter: Long, number: Long) => {
        counter + number
      },
      (counter1: Long, counter2: Long) => {
        counter1 + counter2
      }
    ).collect()

    for (pair <- result) {
      System.out.println("------ key: " + pair._1 + ", uv: " + pair._2)
    }

  }

}

class RandomPartitioner(partitions: Int) extends Partitioner {
  override def numPartitions: Int = partitions

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

推荐阅读更多精彩内容