RocketMQ之NameServer源码解析

前言

NameServer是整个RocketMQ的路由中心，功能类似于Zookeeper，用于服务注册和服务发现，是轻量级别的Zookeeper。

NameServer核心的路由表

private final HashMap<String/*topic*/, List<QueueData>> topicQueueTable;
private final HashMap<String/*brokerName*/, BroKerData> brokerAddrTable;
private final HashMap<String/*ClusterName*/, Set<String/*brokerName*/>> clusterAddrTable;
private final HashMap<String/*brokerAddr*/, BrokerLiveInfo> brokerLiveTable;
private final HashMap<String/*brokerAddr*/, List<String>/*Filter Server*/> filterServerTable;

<img src="https://user-gold-cdn.xitu.io/2020/7/8/1732c4da696d92bf?w=2196&h=2240&f=png&s=177045" alt="关键的表.png" title="关键的表.png" />

topicQueueTable：topic表示消息类型，消息发送时根据路由表进行负载均衡。

brokerAddrTable：Broker基础信息：Broker名称，所属集群名称，主从Broker地址。

clusterAddrTable：Broker集群信息，存储集群中所有的Broker名称。

brokerLiveTable：Broker状态信息，NameServer每次收到Broker的心跳包时都会更新对应的Broker信息。NameServer的心跳检测主要通过扫面该表完成。

其中关键的四张表的结构如上图所示。在了解表结构之后，可以更加容易理解源码中对于表的操作。

路由注册

Broker启动时回向集群中所有的NameServer发送心跳请求，并且每隔30秒都会向所有的NameServer发送心跳包，Name Server收到Broker的心跳包时会更新brokerLiveTable中对应Broker的BrokerLiveInfo的lastUpdateTimeStamp属性，并且NameServer会每个十秒扫面brokerLiveTable，如果连续120s没有收到某个Broker的心跳包，则将所有该Broker对应的路由信息同时关闭Socket通道。

Broker发送心跳包

Broker在启动时会创建一个包含单线程的线程池ScheduledThreadPoolExector对象用于执行定时任务，每30秒向所有的NameServer发送心跳包。如下面代码所示。

/*BrokerController#start*/

//  Broker每隔30秒向NameServer注册中心发送心跳包。
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                try {
                    BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
                } catch (Throwable e) {
                    log.error("registerBrokerAll Exception", e);
                }
            }
        }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);

/*BrokerouterAPI#registerBrokerAll*/

// 获取所有的NameServer服务器的地址列表
List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
// 遍历所有NameServer服务器的地址列表
for (final String namesrvAddr : nameServerAddressList) {
    brokerOuterExecutor.execute(new Runnable() {
        @Override
        public void run() {
            try {
                // 向NameServer服务器发送心跳包
                RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);
                if (result != null) {
                    registerBrokerResultList.add(result);
                }

                log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
            } catch (Exception e) {
                log.warn("registerBroker Exception, {}", namesrvAddr, e);
            } finally {
                countDownLatch.countDown();
            }
        }
    });
}

/*BrokerouterAPI#registerBroker*/

RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.REGISTER_BROKER, requestHeader);
        request.setBody(body);

        if (oneway) {
            try {
                this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);
            } catch (RemotingTooMuchRequestException e) {
                // Ignore
            }
            return null;
        }

NameServer处理来自Broker的心跳包

/*DefaultRequestProcessor#processRequest*/
switch (request.getCode()) {
    case RequestCode.PUT_KV_CONFIG:
        return this.putKVConfig(ctx, request);
    case RequestCode.GET_KV_CONFIG:
        return this.getKVConfig(ctx, request);
    case RequestCode.DELETE_KV_CONFIG:
        return this.deleteKVConfig(ctx, request);
    case RequestCode.QUERY_DATA_VERSION:
        return queryBrokerTopicConfig(ctx, request);
        // 走这里
    case RequestCode.REGISTER_BROKER:
        Version brokerVersion = MQVersion.value2Version(request.getVersion());
        if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {
            return this.registerBrokerWithFilterServer(ctx, request);
        } else {
            return this.registerBroker(ctx, request);
        }
    case RequestCode.UNREGISTER_BROKER:
        return this.unregisterBroker(ctx, request);
        ...
            case RequestCode.UPDATE_NAMESRV_CONFIG:
        return this.updateConfig(ctx, request);
    case RequestCode.GET_NAMESRV_CONFIG:
        return this.getConfig(ctx, request);
    default:
        break;
}

/*DefaultRequestProcessor#registerBroker*/
final RegisterBrokerRequestHeader requestHeader =
            (RegisterBrokerRequestHeader) request.decodeCommandCustomHeader(RegisterBrokerRequestHeader.class);

RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
            requestHeader.getClusterName(),
            requestHeader.getBrokerAddr(),
            requestHeader.getBrokerName(),
            requestHeader.getBrokerId(),
            requestHeader.getHaServerAddr(),
            topicConfigWrapper,
            null,
            ctx.channel()
        );

调用RouteInfoManager类的registerBroker()方法更新缓存

/*RouteInfoManager#registerBroker*/
// 防止并发修改
this.lock.writeLock().lockInterruptibly();
//--------------更新clusterAddrTable缓存-----------
Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
// 如果为空，则Broker第一次注册
if (null == brokerNames) {
    brokerNames = new HashSet<String>();
    this.clusterAddrTable.put(clusterName, brokerNames);
}
brokerNames.add(brokerName);

//------------更新brokerAddrTable缓存------------
boolean registerFirst = false;
BrokerData brokerData = this.brokerAddrTable.get(brokerName);
// 如果通过brokerName无法得到brokerData，说明是第一次注册
if (null == brokerData) {
    // 第一次注册
    registerFirst = true;
    // 创建BrokerData
    brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
    // 将brokerData注册到brokerAddrTable中
    this.brokerAddrTable.put(brokerName, brokerData);
}
// 拿到brokerName对应的所有地址（主从服务器地址）
Map<Long, String> brokerAddrsMap = brokerData.getBrokerAddrs();
//Switch slave to master: first remove <1, IP:PORT> in namesrv, then add <0, IP:PORT>
//The same IP:PORT must only have one record in brokerAddrTable
Iterator<Entry<Long, String>> it = brokerAddrsMap.entrySet().iterator();
// 遍历brokerName对应的所有地址（主从服务器地址）
while (it.hasNext()) {
    Entry<Long, String> item = it.next();
    // 地址相同，但key值和最新的brokerId不同（brokerId为0代表Master，大于0代表Slave），则说明发生了主从服务器broker变更，删除旧的缓存
    if (null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) {
        it.remove();
    }
}
// 将最新的brokerId, brokerAddr放入缓存
String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);



// --------------更新TopicQueueTable缓存-------------
if (null != topicConfigWrapper
    && MixAll.MASTER_ID == brokerId) {// TopicQueueTable中存的是Master的信息，所以brokerId等于0并且null != topicConfigWrapper时，才更新缓存
    if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
        || registerFirst) {// 如果TopicQueueTable缓存中没有broker的地址brokerAddr的信息或收到心跳包的时间和旧的时间不相等或broker为第一次注册。
        ConcurrentMap<String, TopicConfig> tcTable =
            topicConfigWrapper.getTopicConfigTable();
        if (tcTable != null) {
            for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
                // 更新TopicQueueTable缓存
                this.createAndUpdateQueueData(brokerName, entry.getValue());
            }
        }
    }
}

// --------------更新brokerLiveTable缓存-------------
BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr,
                    new BrokerLiveInfo(
                        System.currentTimeMillis(),
                        topicConfigWrapper.getDataVersion(),
                        channel,
                        haServerAddr));
                if (null == prevBrokerLiveInfo) {
                    log.info("new broker registered, {} HAServer: {}", brokerAddr, haServerAddr);
                }




/*RouteInfoManager#createAndUpdateQueueData*/
// 更新TopicQueueTable缓存
private void createAndUpdateQueueData(final String brokerName, final TopicConfig topicConfig) {
        QueueData queueData = new QueueData();
        queueData.setBrokerName(brokerName);
        queueData.setWriteQueueNums(topicConfig.getWriteQueueNums());
        queueData.setReadQueueNums(topicConfig.getReadQueueNums());
        queueData.setPerm(topicConfig.getPerm());
        queueData.setTopicSynFlag(topicConfig.getTopicSysFlag());
        // ---------topicQueueTable缓存---------
        List<QueueData> queueDataList = this.topicQueueTable.get(topicConfig.getTopicName());
        if (null == queueDataList) {// queueDataList为空这创建并添加最新的queueData
            queueDataList = new LinkedList<QueueData>();
            queueDataList.add(queueData);
            this.topicQueueTable.put(topicConfig.getTopicName(), queueDataList);
            log.info("new topic registered, {} {}", topicConfig.getTopicName(), queueData);
        } else {
            boolean addNewOne = true;

            Iterator<QueueData> it = queueDataList.iterator();
            while (it.hasNext()) {
                QueueData qd = it.next();
                // 判断是否已经存在brokerame对应的queueData信息
                if (qd.getBrokerName().equals(brokerName)) {
                    if (qd.equals(queueData)) {// 判断旧的queueData和新的queueData是否等价
                        addNewOne = false;
                    } else {
                        log.info("topic changed, {} OLD: {} NEW: {}", topicConfig.getTopicName(), qd,
                            queueData);
                        // 不等价则删除旧的
                        it.remove();
                    }
                }
            }
            // 如果最新queueData不等价旧的，或者没有brokerName对应的信息，则添加最新queueData
            if (addNewOne) {
                queueDataList.add(queueData);
            }
        }
    }

路由删除

在Broker发生宕机无法发送心跳包时，NameServer无法收到心跳包，NameServer每个10秒就会扫面brokerLiveTable，检测每个Broker地址上次收到心跳包的时间lastUpdateTimeStamp是否超过120s，如果超过120s则认为broker失效，NameServer会移除在topicQueueTable，brokerAddrTable，clusterAddrTable，brokerLiveTable，filterServerTable对应的失效Broker信息。

执行路由删除的两个触发点：

  1. NameServer每十秒的扫描
  2. Broker在正常状态下执行unregisterBroker指令

由于两个出发点执行都是同一段路由删除代码，下面通过第一个出发点进行源码分析。

/*NamesrvController#initialize*/

// 每个10秒扫描一次Broker，移除处于不激活状态的Broker（遍历brokerLiveTable（HashMap<String/* brokerAddr */, BrokerLiveInfo>,BrokerLiveInfo.lastUpdateTimestamp用于记录Broker上一次发送心跳包的时间，如果超过了120就删除对应的Broker）
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                NamesrvController.this.routeInfoManager.scanNotActiveBroker();
            }
        }, 5, 10, TimeUnit.SECONDS);

/*RouteInfoManager#scanNotActiveBroker*/

Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<String, BrokerLiveInfo> next = it.next();
            long last = next.getValue().getLastUpdateTimestamp();
            // 上次更新的时间如果超过120s则删除
            if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {
                // 关闭与该Broker的通道
                RemotingUtil.closeChannel(next.getValue().getChannel());
                // 删除
                it.remove();
                log.warn("The broker channel expired, {} {}ms", next.getKey(), BROKER_CHANNEL_EXPIRED_TIME);
                // 更新其他缓存中对应的信息
                this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
            }
        }

/*RouteInfoManager#onChannelDestroy*/

if (brokerAddrFound != null && brokerAddrFound.length() > 0) {

            try {
                try {
                    this.lock.writeLock().lockInterruptibly();
                    // 从brokerLiveTable中删除失效地址brokerAddrFound的信息
                    this.brokerLiveTable.remove(brokerAddrFound);
                    // 从filterServerTable中删除失效地址brokerAddrFound的信息
                    this.filterServerTable.remove(brokerAddrFound);
                    String brokerNameFound = null;
                    boolean removeBrokerName = false;
                    Iterator<Entry<String, BrokerData>> itBrokerAddrTable =
                        this.brokerAddrTable.entrySet().iterator();
                    // // 从brokerAddrTable找到失效地址对应得Broker名字
                    while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {
                        BrokerData brokerData = itBrokerAddrTable.next().getValue();

                        Iterator<Entry<Long, String>> it = brokerData.getBrokerAddrs().entrySet().iterator();
                        while (it.hasNext()) {
                            Entry<Long, String> entry = it.next();
                            Long brokerId = entry.getKey();
                            String brokerAddr = entry.getValue();
                            if (brokerAddr.equals(brokerAddrFound)) {
                                // 失效地址的Broker名字
                                brokerNameFound = brokerData.getBrokerName();
                                // 从失效地址的Broker名字对应的brokerData的主从地址集中删除失效地址
                                it.remove();
                                log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",
                                    brokerId, brokerAddr);
                                break;
                            }
                        }
                        // 如果失效地址的Broker名字对应的brokerData的主从地址集中已经没有地址，说明Broker名字对应的所有服务器失效，需要从brokerAddrTable中删除失效地址的Broker名字对应的brokerData
                        if (brokerData.getBrokerAddrs().isEmpty()) {
                            removeBrokerName = true;
                            itBrokerAddrTable.remove();
                            log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",
                                brokerData.getBrokerName());
                        }
                    }
                    // 失效地址的Broker名字对应的地址已经全部失效，删除其在clusterAddrTable中对应的信息
                    if (brokerNameFound != null && removeBrokerName) {
                        Iterator<Entry<String, Set<String>>> it = this.clusterAddrTable.entrySet().iterator();
                        while (it.hasNext()) {
                            Entry<String, Set<String>> entry = it.next();
                            String clusterName = entry.getKey();
                            Set<String> brokerNames = entry.getValue();
                            boolean removed = brokerNames.remove(brokerNameFound);
                            if (removed) {
                                log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",
                                    brokerNameFound, clusterName);

                                if (brokerNames.isEmpty()) {
                                    log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",
                                        clusterName);
                                    it.remove();
                                }

                                break;
                            }
                        }
                    }
                    // 失效地址的Broker名字对应的地址已经全部失效，删除其在topicQueueTable中对应的信息
                    if (removeBrokerName) {
                        Iterator<Entry<String, List<QueueData>>> itTopicQueueTable =
                            this.topicQueueTable.entrySet().iterator();
                        while (itTopicQueueTable.hasNext()) {
                            Entry<String, List<QueueData>> entry = itTopicQueueTable.next();
                            String topic = entry.getKey();
                            List<QueueData> queueDataList = entry.getValue();

                            Iterator<QueueData> itQueueData = queueDataList.iterator();
                            while (itQueueData.hasNext()) {
                                QueueData queueData = itQueueData.next();
                                if (queueData.getBrokerName().equals(brokerNameFound)) {
                                    itQueueData.remove();
                                    log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",
                                        topic, queueData);
                                }
                            }

                            if (queueDataList.isEmpty()) {
                                itTopicQueueTable.remove();
                                log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",
                                    topic);
                            }
                        }
                    }
                } finally {
                    this.lock.writeLock().unlock();
                }
            } catch (Exception e) {
                log.error("onChannelDestroy Exception", e);
            }
        }

路由发现

通过主题先从topicQueueTable中获取到对应的所有Broker信息，然后再从brokerAddrTable中根据broker名称获取对应的信息，最后通过从brokerAddrTable获取的地址信息从filterServerTable获取对应的信息。

/*DefaultRequestProcessor#processRequest*/

switch (request.getCode()) {
    case RequestCode.PUT_KV_CONFIG:
        return this.putKVConfig(ctx, request);
        ...
    case RequestCode.UNREGISTER_BROKER:
        return this.unregisterBroker(ctx, request);
    // 走这里
    case RequestCode.GET_ROUTEINTO_BY_TOPIC:
        return this.getRouteInfoByTopic(ctx, request);
    case RequestCode.GET_BROKER_CLUSTER_INFO:
        return this.getBrokerClusterInfo(ctx, request);
    ...
    case RequestCode.GET_NAMESRV_CONFIG:
        return this.getConfig(ctx, request);
    default:
        break;
        }

/*DefaultRequestProcessor#getRouteInfoByTopic*/

final RemotingCommand response = RemotingCommand.createResponseCommand(null);
final GetRouteInfoRequestHeader requestHeader =
    (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
// 通过主题先后从topicQueueTable，brokerAddrTable，filterServerTable获取对应的信息
TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());

if (topicRouteData != null) {
    if (this.namesrvController.getNamesrvConfig().isOrderMessageEnable()) {
        String orderTopicConf =
            this.namesrvController.getKvConfigManager().getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG,
                                                                    requestHeader.getTopic());
        topicRouteData.setOrderTopicConf(orderTopicConf);
    }

    byte[] content = topicRouteData.encode();
    response.setBody(content);
    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
}

response.setCode(ResponseCode.TOPIC_NOT_EXIST);
response.setRemark("No topic route info in name server for the topic: " + requestHeader.getTopic()
                   + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
return response;

/*RouteInfoManager#pickupTopicRouteData*/

TopicRouteData topicRouteData = new TopicRouteData();
        boolean foundQueueData = false;
        boolean foundBrokerData = false;
        // 用于存放主题对应的所有broker名称
        Set<String> brokerNameSet = new HashSet<String>();
        // 用于存放从brokerAddrTable获取的brokerData
        List<BrokerData> brokerDataList = new LinkedList<BrokerData>();
        topicRouteData.setBrokerDatas(brokerDataList);
        // 用于存放从filterServerTable获取的信息
        HashMap<String, List<String>> filterServerMap = new HashMap<String, List<String>>();
        topicRouteData.setFilterServerTable(filterServerMap);

        try {
            try {
                this.lock.readLock().lockInterruptibly();
                // 获取主题对应的broker服务器信息，即topicQueueTable中主题对应的所有QueueData
                List<QueueData> queueDataList = this.topicQueueTable.get(topic);
                if (queueDataList != null) {
                    topicRouteData.setQueueDatas(queueDataList);
                    foundQueueData = true;

                    Iterator<QueueData> it = queueDataList.iterator();
                    while (it.hasNext()) {
                        QueueData qd = it.next();
                        brokerNameSet.add(qd.getBrokerName());
                    }
                    //根据主题对应的broker名称，从brokerAddrTable中获取对应brokerData信息，brokerData包含集群名称，broker名称，broker主从地址信息
                    for (String brokerName : brokerNameSet) {
                        BrokerData brokerData = this.brokerAddrTable.get(brokerName);
                        if (null != brokerData) {
                            BrokerData brokerDataClone = new BrokerData(brokerData.getCluster(), brokerData.getBrokerName(), (HashMap<Long, String>) brokerData
                                .getBrokerAddrs().clone());
                            brokerDataList.add(brokerDataClone);
                            foundBrokerData = true;
                            // 根据brokerDataClone的地址信息获取到filterServerTable中的信息
                            for (final String brokerAddr : brokerDataClone.getBrokerAddrs().values()) {
                                List<String> filterServerList = this.filterServerTable.get(brokerAddr);
                                filterServerMap.put(brokerAddr, filterServerList);
                            }
                        }
                    }
                }
            } finally {
                this.lock.readLock().unlock();
            }
        } catch (Exception e) {
            log.error("pickupTopicRouteData Exception", e);
        }

        log.debug("pickupTopicRouteData {} {}", topic, topicRouteData);

        if (foundBrokerData && foundQueueData) {
            return topicRouteData;
        }

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