图解ConcurrentHashMap

ConcurrentHashMap是什么

Concurrent翻译过来是并发的意思,字面理解它的作用是处理并发情况的 HashMap,在介绍它之前先回顾下之前的知识。通过前面两篇学习,我们知道多线程并发下 HashMap 是不安全的(如死循环),更普遍的是多线程并发下,由于堆内存对于各个线程是共享的,而 HashMap 的 put 方法不是原子操作,假设Thread1先 put 值,然后 sleep 2秒(也可以是系统时间片切换失去执行权),在这2秒内值被Thread2改了,Thread1“醒来”再 get 的时候发现已经不是原来的值了,这就容易出问题。

进群:697699179可以获取Java各类入门学习资料!

这是我的微信公众号【编程study】各位大佬有空可以关注下,每天更新Java学习方法,感谢!

学习中遇到问题有不明白的地方,推荐加小编Java学习群:697699179内有视频教程 ,直播课程 ,等学习资料,期待你的加入

那么如何避免这种多线程“奥迪变奥拓”的情况呢?常规思路就是给 HashMap 的 put 方法加锁(synchronized),保证同一个时刻只允许一个线程拥有对 hashmap 有写的操作权限即可。然而假如线程1中操作耗时,占着茅坑半天不出来,其他需要操作该 hashmap 的线程就需要在门口排队半天,严重影响用户体验(HashTable 就是这么干的)。举个生活中的例子,很多银行除了存取钱,还支持存取贵重物品,贵重物品都放在保险箱里,把 HashMap 和 HashTable 比作银行,结构:

把线程比作人,对应的情况如下:

HashMap牌银行:我们的服务宗旨是不用排队,同一时间多人都有机会修改保险柜里的东西,你以为你存的是美元?取出来的其实是日元,破产就在一瞬间,刺不刺激。

HashTable牌银行:我们的服务宗旨是要排队,同一时间只有一个人有机会修改保险柜里的东西,其余的人只能看不能动手改,保你存的是美元取得还是美元。什么?你说如果那人在里面睡着了不出来怎么办?不要着急,来,坐下来打会麻将等他出来。

多线程下用 HashMap 不确定性太高,有破产的风险,不能选;用 HashTable 不会破产,但是用户体验不太好,那么怎样才能做到多人存取既不影响他人存值,又不用排队呢?有人提议搞个「银行者联盟」,多开几个像HashTable 这种「带锁」的银行就好了,有多少人办理业务,就开多少个银行,一对一服务,这个区都是大老板,开银行的成本都是小钱,于是「银行者联盟」成立了。

接下来的情况是这样的:比如盖伦和亚索一起去银行存他们的大宝剑,这个「银行者联盟」一顿操作,然后对盖伦说,1号银行现在没人,你可以去那存,不用排队,然后盖伦就去1号银行存他的大宝剑,1号银行把盖伦接进门,马上拉闸,一顿操作,然后把盖伦的大宝剑放在第x行第x个保险箱,等盖伦办妥离开后,再开闸;同样「银行者联盟」对亚索说,2号银行现在没人,你可以去那存,不用排队,然后亚索去2号银行存他的大宝剑,2号银行把亚索接进门,马上拉闸,一顿操作把亚索的大宝剑放在第x行第x号保险箱,等亚索离开后再开闸,此时不管盖伦和亚索在各自银行里面待多久都不会影响到彼此,不用担心自己的大宝剑被人偷换了。这就是ConcurrentHashMap的设计思路,用一个图来理解

从上图可以看出,此时锁的是对应的单个银行,而不是整个「银行者联盟」。分析下这种设计的特点:

多个银行组成的「银行者联盟」

当有人来办理业务时,「银行者联盟」需要确定这个人去哪个银行

当此人去到指定银行办理业务后,该银行上锁,其他人不能同时执行修改操作,直到此人离开后解锁

由这几点基本思想可以引发一些思考,比如:

1.成立「银行者联盟」时初识银行数是多少?怎么设计合理?

上面这张图没有给出是否需要排队的结论,这是因为需要结合实际情况分析,比如初识化有16个银行,只有两个人来办理业务,那自然不需要排队;如果现在16个银行都有人在办理业务,这时候来了第17个人,那么他还是需要排队的。由于「银行者联盟」事先无法得知会有多少人来办理业务,所以在它创立的时候需要制定一个「标准」,即初始银行数量,人多的情况「银行者联盟」应该多开几家银行,避免别人排队;人少的情况应该少开,避免浪费钱(什么,你说不差钱?那也不行)

2.当有人来办理业务的时候,「银行者联盟」怎么确定此人去哪个银行?

正常情况下,如果所有银行都是未上锁状态,那么有人来办理业务去哪都不用排队,当其中有些银行已经上锁,那么后续「银行者联盟」给人推荐的时候就不能把客户往上锁的银行引了,否则分分钟给人锤成麻瓜。因此「银行者联盟」需要时刻保持清醒的头脑,对自己的银行空闲情况了如指掌,每次给用户推荐都应该是最好的选择。

3.「银行者联盟」怎么保证同一时间不会有两个人在同一个银行拥有存权限?

通过对指定银行加锁/解锁的方式实现。

源码分析

Java7 源码分析

通过 Java7 的源码分析下代码实现,先看下一些重要的成员

//默认的数组大小16(HashMap里的那个数组)staticfinalintDEFAULT_INITIAL_CAPACITY =16;//扩容因子0.75staticfinalfloatDEFAULT_LOAD_FACTOR =0.75f;//ConcurrentHashMap中的数组finalSegment[] segments//默认并发标准16staticfinalintDEFAULT_CONCURRENCY_LEVEL =16;//Segment是ReentrantLock子类,因此拥有锁的操作staticfinalclassSegmentextendsReentrantLockimplementsSerializable{//HashMap的那一套,分别是数组、键值对数量、阈值、负载因子transientvolatileHashEntry[] table;transientintcount;transientintthreshold;finalfloatloadFactor;  Segment(floatlf,intthreshold, HashEntry[] tab) {this.loadFactor = lf;this.threshold = threshold;this.table = tab;        } }//换了马甲还是认识你!!!HashEntry对象,存key、value、hash值以及下一个节点staticfinalclassHashEntry{finalinthash;finalK key;volatileV value;volatileHashEntry next; }//segment中HashEntry[]数组最小长度staticfinalintMIN_SEGMENT_TABLE_CAPACITY =2;//用于定位在segments数组中的位置,下面介绍finalintsegmentMask;finalintsegmentShift;

上面这些一下出来有点接受不了没关系,下面都会介绍到。

接下来从最简单的初识化开始分析

ConcurrentHashMapconcurrentHashMap = new ConcurrentHashMap();

默认构造函数会调用带三个参数的构造函数

publicConcurrentHashMap(){this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);    }publicConcurrentHashMap(intinitialCapacity,floatloadFactor,intconcurrencyLevel){if(!(loadFactor >0) || initialCapacity <0|| concurrencyLevel <=0)thrownewIllegalArgumentException();if(concurrencyLevel > MAX_SEGMENTS)            concurrencyLevel = MAX_SEGMENTS;// Find power-of-two sizes best matching arguments//步骤① startintsshift =0;intssize =1;while(ssize < concurrencyLevel) {            ++sshift;            ssize <<=1;        }this.segmentShift =32- sshift;this.segmentMask = ssize -1;//步骤① end//步骤② startif(initialCapacity > MAXIMUM_CAPACITY)            initialCapacity = MAXIMUM_CAPACITY;intc = initialCapacity / ssize;if(c * ssize < initialCapacity)            ++c;intcap = MIN_SEGMENT_TABLE_CAPACITY;while(cap < c)            cap <<=1;//步骤② end// create segments and segments[0]//步骤③ startSegment s0 =newSegment(loadFactor, (int)(cap * loadFactor),                            (HashEntry[])newHashEntry[cap]);        Segment[] ss = (Segment[])newSegment[ssize];        UNSAFE.putOrderedObject(ss, SBASE, s0);// ordered write of segments[0]this.segments = ss;//步骤③ end}

上面定义了许多临时变量,注释写的又少,第一次看名字根本不知道这鬼东西代表什么意思,不过我们可以把已知的数据代进去,算出这些变量的值,再分析能不能找出一些猫腻。假设这是第一次默认创建:

步骤① concurrencyLevel = 16 ,可以计算出 sshift = 4,ssize = 16,segmentShift = 28,segmentMask = 15;

步骤② c = 16/16 = 1,cap = 2;

步骤③有句注释,创建 Segment 数组 segments 并初始化 segments [0] ,所以 s0 初始化后数组长度为2,负载因子0.75,阈值为1;再看这里的ss的初始化(重点,圈起来要考!!!), ssize 此时为16,所以默认数组长度16,给人一种感觉正好和我们传的 concurrencyLevel 一样?看下下面的例子

例子1例子2

ssize = 1,concurrencyLevel = 10ssize = 1,concurrencyLevel = 8

ssize <<= 1 —> 2<10 满足ssize <<= 1 —> 2<10 满足

ssize <<= 1 —> 4<10 满足ssize <<= 1 —> 4<10 满足

ssize <<= 1 —> 8<10 满足ssize <<= 1 —> 8<10 不满足 ssize = 8

ssize <<= 1 —> 16<10 不满足 ssize = 16

所以我们传 concurrencyLevel 不一定就是最后数组的长度,长度的计算公式:

长度 = 2的n次方(2的n次方 >= concurrencyLevel)

到这里只是创建了一个长度为16的Segment 数组,并初始化数组0号位置,segmentShift和segmentMask还没派上用场,画图存档:

接着看 put 方法

publicVput(K key, Vvalue){        Segment s;//步骤①注意valus不能为空!!!if(value==null)thrownewNullPointerException();//根据key计算hash值,key也不能为null,否则hash(key)报空指针inthash = hash(key);//步骤②派上用场了,根据hash值计算在segments数组中的位置intj = (hash >>> segmentShift) & segmentMask;//步骤③查看当前数组中指定位置Segment是否为空//若为空,先创建初始化Segment再put值,不为空,直接put值。if((s = (Segment)UNSAFE.getObject// nonvolatile; recheck(segments, (j << SSHIFT) + SBASE)) ==null)//  in ensureSegments = ensureSegment(j);returns.put(key, hash,value,false);    }

步骤①可以看到和 HashMap 的区别,这里的 key/value 为空会报空指针异常;步骤②先根据 key 值计算 hash 值,再和前面算出来的两个变量计算出这个 key 应该放在哪个Segment中(具体怎么计算的有兴趣可以去研究下,先高位运算再取与),假设我们算出来该键值对应该放在5号,步骤③判断5号为空,看下 ensureSegment() 方法

privateSegmentensureSegment(intk){//获取segmentsfinalSegment[] ss =this.segments;longu = (k << SSHIFT) + SBASE;// raw offsetSegment seg;if((seg = (Segment)UNSAFE.getObjectVolatile(ss, u)) ==null) {//拷贝一份和segment 0一样的segmentSegment proto = ss[0];// use segment 0 as prototype//大小和segment 0一致,为2intcap = proto.table.length;//负载因子和segment 0一致,为0.75floatlf = proto.loadFactor;//阈值和segment 0一致,为1intthreshold = (int)(cap * lf);//根据大小创建HashEntry数组tabHashEntry[] tab = (HashEntry[])newHashEntry[cap];//再次检查if((seg = (Segment)UNSAFE.getObjectVolatile(ss, u))                ==null) {// recheck根据已有属性创建指定位置的Segment                Segment s =newSegment(lf, threshold, tab);while((seg = (Segment)UNSAFE.getObjectVolatile(ss, u))                      ==null) {if(UNSAFE.compareAndSwapObject(ss, u,null, seg = s))break;                }            }        }returnseg;    }

该方法重点在于拷贝了segments[0],因此新创建的Segment与segment[0]的配置相同,由于多个线程都会有可能执行该方法,因此这里通过UNSAFE的一些原子性操作的方法做了多次的检查,到目前为止画图存档:

现在“舞台”也有了,请开始你的表演,看下 Segment 的put方法

final Vput(K key,inthash, Vvalue, boolean onlyIfAbsent){//步骤① startHashEntry node = tryLock() ?null:                scanAndLockForPut(key, hash,value);//步骤① endV oldValue;try{//步骤② start//获取Segment中的HashEntry[]HashEntry[] tab = table;//算出在HashEntry[]中的位置intindex = (tab.length -1) & hash;//找到HashEntry[]中的指定位置的第一个节点HashEntry first = entryAt(tab, index);for(HashEntry e = first;;) {//如果不为空,遍历这条链if(e !=null) {                        K k;//情况① 之前已存过,则替换原值if((k = e.key) == key ||                            (e.hash == hash && key.equals(k))) {                            oldValue = e.value;if(!onlyIfAbsent) {                                e.value=value;                                ++modCount;                            }break;                        }                        e = e.next;                    }else{//情况② 另一个线程的准备工作if(node !=null)//链表头插入方式node.setNext(first);else//情况③ 该位置为空,则新建一个节点(注意这里采用链表头插入方式)node =newHashEntry(hash, key,value, first);//键值对数量+1intc = count +1;//如果键值对数量超过阈值if(c > threshold && tab.length < MAXIMUM_CAPACITY)//扩容rehash(node);else//未超过阈值,直接放在指定位置setEntryAt(tab, index, node);                        ++modCount;                        count = c;//插入成功返回nulloldValue =null;break;                    }                }//步骤② end}finally{//步骤③//解锁unlock();            }//修改成功,返回原值returnoldValue;        }

上面的 put 方法其实和 Java7 HashMap里大致是一样的,只是多了加锁/解锁两步,也正因为这样才保证了同一时刻只有一个线程拥有修改的权限。按步骤分析下上面的流程:

步骤① 执行 tryLock 方法获取锁,拿到锁返回null,没拿到锁执行 scanAndLockForPut 方法;

步骤② 和 HashMap 里的那一套思路是一样的,不理解可以看下之前的文章介绍(情况②下面介绍);

步骤③ 执行 unLock 方法解锁

假设现在Thread1进来存值,前面没人来过,它可以成功拿到锁,根据计算,得出它要存的键值对应该放在HashEntry[] 的0号位置,0号位置为空,于是新建一个 HashEntry,并通过 setEntryAt() 方法,放在0号位置,然而还没等 Thread1 释放锁,系统的时间片切到了 Thread2 ,先画图存档

Thread2 也来存值,通过前面的计算,恰好 Thread2 也被定位到 segments[5],接下来 Thread2 尝试获取锁,没有成功(Thread1 还未释放),执行 scanAndLockForPut() 方法:

privateHashEntryscanAndLockForPut(K key,inthash, Vvalue){//通过Segment和hash值寻找匹配的HashEntryHashEntry first = entryForHash(this, hash);            HashEntry e = first;            HashEntry node =null;//重试次数intretries =-1;// negative while locating node//循环尝试获取锁while(!tryLock()) {                HashEntry f;// to recheck first below//步骤①if(retries <0) {//情况① 没找到,之前表中不存在if(e ==null) {if(node ==null)// speculatively create node//新建 HashEntry 备用,retries改成0node =newHashEntry(hash, key,value,null);                        retries =0;                    }//情况② 找到,刚好第一个节点就是,retries改成0elseif(key.equals(e.key))                        retries=0;//情况③ 第一个节点不是,移到下一个,retries还是-1,继续找elsee = e.next;                }//步骤②//尝试了MAX_SCAN_RETRIES次还没拿到锁,简直B了dog!elseif(++retries > MAX_SCAN_RETRIES){//泉水挂机lock();break;                }//步骤③//在MAX_SCAN_RETRIES次过程中,key对应的entry发生了变化,则从头开始elseif((retries &1)==0&&                        (f = entryForHash(this, hash)) != first) {                    e = first = f;// re-traverse if entry changedretries =-1;                }            }returnnode;        }

通过上面的注释分析可以看出,Thread2 虽然此刻没有权限修改,但是它也没闲着,利用等锁的这个时间,把自己要放的键值对在数组中哪个位置计算出来了,这样当 Thread2 一拿到锁就可以立马定位到具体位置操作,节省时间。上面的步骤③稍微解释下,比如 Thread2 通过查找得知自己要修改的值在0号位置,但在 Thread1 里面又把该值改到了1号位置,如果它还去0号操作那肯定出问题了,所以需要重新确定。

假设 Thread2 put 值为("亚索",“98”),对应1号位置,那么在 scanAndLockForPut 方法中对应情况①,画图存档:

再回到 Segment put 方法中的情况②,当 Thread1 释放锁后,Thread2 持有锁,并准备把亚索放在1号位置,然而此时 Segment[5] 里的键值对数量2 > 阈值1,所以调用 rehash() 方法扩容,

privatevoidrehash(HashEntry node) {/*

            * Reclassify nodes in each list to new table.  Because we

            * are using power-of-two expansion, the elements from

            * each bin must either stay at same index, or move with a

            * power of two offset. We eliminate unnecessary node

            * creation by catching cases where old nodes can be

            * reused because their next fields won't change.

            * Statistically, at the default threshold, only about

            * one-sixth of them need cloning when a table

            * doubles. The nodes they replace will be garbage

            * collectable as soon as they are no longer referenced by

            * any reader thread that may be in the midst of

            * concurrently traversing table. Entry accesses use plain

            * array indexing because they are followed by volatile

            * table write.

            *///旧数组引用HashEntry[] oldTable = table;//旧数组长度intoldCapacity = oldTable.length;//新数组长度为旧数组的2倍intnewCapacity = oldCapacity <<1;//修改新的阈值threshold = (int)(newCapacity * loadFactor);//创建新表HashEntry[] newTable =                (HashEntry[])newHashEntry[newCapacity];intsizeMask = newCapacity -1;//遍历旧表for(inti =0; i < oldCapacity ; i++) {                HashEntry e = oldTable[i];if(e !=null) {                    HashEntrynext= e.next;//确定在新表中的位置intidx = e.hash & sizeMask;//情况① 链表只有一个节点,指定转移到新表指定位置if(next==null)//  Single node on listnewTable[idx] = e;else{// Reuse consecutive sequence at same slotHashEntry lastRun = e;intlastIdx = idx;for(HashEntry last =next;                            last !=null;                            last = last.next) {//情况② 扩容前后位置发生改变intk = last.hash & sizeMask;if(k != lastIdx) {                                lastIdx = k;                                lastRun = last;                            }                        }//将改变的键值对放到新表的对应位置newTable[lastIdx] = lastRun;// Clone remaining nodes//情况③ 把链表中剩下的节点拷到新表中for(HashEntry p = e; p != lastRun; p = p.next) {                            V v = p.value;inth = p.hash;intk = h & sizeMask;                            HashEntry n = newTable[k];                            newTable[k] =newHashEntry(h, p.key, v, n);                        }                    }                }            }//添加新的节点(链表头插入方式)intnodeIndex = node.hash & sizeMask;// add the new nodenode.setNext(newTable[nodeIndex]);            newTable[nodeIndex] = node;            table = newTable;        }

同样是扩容转移,这里的代码比 HashMap 中的 transfer 多了一些操作,在上上篇学习 HashMap 扩容可知,扩容后键值对的新位置要么和原位置一样,要么等于原位置+旧数组的长度,所以画个图来理解下上面代码这么写的原因:

前提:当前 HashEntry[] 长度为8,阈值为 8*0.75 = 6,所以 put 第7个键值对需要扩容 ,盖伦和亚索扩容前后位置不变,妖姬和卡特扩容后位置需要加上原数组长度,所以执行上面代码流程:

上面的代码先找出扩容前后需要转移的节点,先执行转移,然后再把该条链上剩下的节点转移,之所以这么写是起到复用的效果,注释中也说了,在使用默认阈值的情况下,只有大约 1/6 的节点需要被 clone 。注意到目前为止,可以看到无论是扩容转移还是新增节点,Java7都是采用的头插入方式,流程图如下:

相比之下,get 方法没有加锁/解锁的操作,代码比较简单就不分析了。

稍微说下Java8

Java8 对比Java7有很大的不同,比如取消了Segments数组,允许并发扩容。

先看下ConcurrentHashMap的初始化

publicConcurrentHashMap(){}

和Java7不一样,这里是个空方法,那么它具体的初始化操作呢?直接看下 put 方法

publicVput(K key, Vvalue){returnputVal(key,value,false);}/** Implementation for put and putIfAbsent */final VputVal(K key, Vvalue, boolean onlyIfAbsent){// key/value不能为空!!!if(key ==null||value==null)thrownewNullPointerException();//计算hash值inthash = spread(key.hashCode());intbinCount =0;for(Node[] tab = table;;) {        Node f;intn, i, fh;//注释① 表为null则初始化if(tab ==null|| (n = tab.length) ==0)            tab = initTable();//CAS方法判断指定位置是否为null,为空则通过创建新节点,通过CAS方法设置在指定位置elseif((f = tabAt(tab, i = (n -1) & hash))==null) {if(casTabAt(tab, i,null,newNode(hash, key,value,null)))break;// no lock when adding to empty bin}//当前节点正在扩容elseif((fh = f.hash)== MOVED)            tab = helpTransfer(tab, f);//指定位置不为空else{            V oldVal =null;//注释② 加锁synchronized (f) {if(tabAt(tab, i) == f) {//节点是链表的情况if(fh >=0) {                        binCount =1;//遍历整体链for(Node e = f;; ++binCount) {                            K ek;//如果已存在,替换原值if(e.hash == hash &&                                ((ek = e.key) == key ||                                (ek !=null&& key.equals(ek)))) {                                oldVal = e.val;if(!onlyIfAbsent)                                    e.val =value;break;                            }                            Node pred = e;//如果是新加节点,则以尾部插入实现添加if((e = e.next) ==null) {                                pred.next =newNode(hash, key,value,null);break;                            }                        }                    }//节点是红黑树的情况elseif(f instanceof TreeBin){                        Node p;                        binCount =2;//遍历红黑树if((p = ((TreeBin)f).putTreeVal(hash, key,value)) !=null) {                            oldVal = p.val;if(!onlyIfAbsent)                                p.val =value;                        }                    }elseif(f instanceof ReservationNode)                        throw newIllegalStateException("Recursive update");                }            }if(binCount !=0) {//链表中节点个数超过8转成红黑树if(binCount >= TREEIFY_THRESHOLD)                    treeifyBin(tab, i);if(oldVal !=null)returnoldVal;break;            }        }    }//注释③ 添加节点addCount(1L, binCount);returnnull;}

代码有点长,第一次看很有可能引起身体不适,主要是因为引入了红黑树的判断和操作,以及线程安全的操作。同样key/value 为空会报空指针异常,这也是和 HashMap 一个明显的区别。

注释①

调用 initTable 初始化数组

privatefinalNode[] initTable() {    Node[] tab; int sc;while((tab = table) ==null|| tab.length ==0) {// sizeCtl小于0,当前线程让出执行权if((sc = sizeCtl) <0)            Thread.yield();// lost initialization race; just spin//CAS 操作将 sizeCtl 值改为-1elseif(U.compareAndSwapInt(this, SIZECTL, sc,-1)) {try{if((tab = table) ==null|| tab.length ==0) {                    int n = (sc >0) ? sc : DEFAULT_CAPACITY;@SuppressWarnings("unchecked")//默认创建大小为16的数组Node[] nt = (Node[])new Node[n];                    table = tab = nt;                    sc = n - (n >>>2);                }            }finally{//初始化完再改回来sizeCtl = sc;            }break;        }    }returntab;}

put方法并没有加锁,那么它是如何保证创建新表的时候并发安全呢?答案就是这里的 sizeCtl ,sizeCtl 默认值为0,当一个线程初始化数组时,会将 sizeCtl 改成 -1,由于被 volatile 修饰,对于其他线程来说这个变化是可见的,上面代码看到后续线程判断 sizeCtl 小于0 就会让出执行权。

注释②

Java8 摒弃了Segment,而是对数组中单个位置加锁。当指定位置节点不为 null 时,情况与 Java8 HashMap 操作类似,新节点的添加还是尾部插入方式。

注释③

不管是链表的还是红黑树,确定之后总的节点数会加1,可能会引起扩容,Java8 ConcunrrentHashMap 支持并发扩容,之前扩容总是由一个线程将旧数组中的键值对转移到新的数组中,支持并发的话,转移所需要的时间就可以缩短了,当然相应的并发处理控制逻辑也就更复杂了,扩容转移通过 transfer 方法完成,Java8中该方法很长,感兴趣的可以看下源码。。。

用一个图来表示 Java8 ConcurrentHashMap的样子

总结

通过分析源码对比了 HashMap 与 ConcurrentHashMap的差别,以及Java7和Java8上 ConcurrentHashMap 设计的不同,当然还有很多坑没有填,比如其中调用了很多UNSAFE的CAS方法,可以减少性能上的消耗,平时很少用,了解的比较少;以及红黑树的具体原理和实现,后续慢慢填。。。

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 204,053评论 6 478
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,527评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 150,779评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,685评论 1 276
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,699评论 5 366
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,609评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,989评论 3 396
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,654评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,890评论 1 298
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,634评论 2 321
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,716评论 1 330
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,394评论 4 319
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,976评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,950评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,191评论 1 260
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 44,849评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,458评论 2 342

推荐阅读更多精彩内容