spring boot整合rocketMQ

关于消息队列这部分内容，之前自己简单了解和学习过Kafka，但是目前国内公司对RocketMq使用的也比较多，并且我所在的项目组已经引入了RocketMq，但是因为项目进度问题，在开发中还没有正式的使用。所以自己准别先提前了解一下，关于RocketMq我想不用我过多介绍，而且性能据说也是特别的好，并且也支持专门的事务消息，可以说非常的有吸引力。学习的资料无非也就是官方文档。这次学习使用的是4.4.0版本。
下面就正式开始今天的RocketMq学习，当然首先还是创建demo项目

一、创建项目

今天的主要目的就是学习如何在Spring Boot中使用RocketMq，包括消息的生产、消费（同步、异步）。当然如果时间允许的话我也会尝试使用消息广播和事务消息的发送、消费。
首先创建项目，引入相关依赖，这里说一下因为是和Spring Boot整合使用，我并没有使用官方的依赖，而是使用了starter其对rocketmq-client又做了一层的封装，个人感觉使用起来更方便，Spring Boot版本使用的是：2.3.1.RELEASE，项目的pom文件如下：

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-jpa</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.postgresql</groupId>
            <artifactId>postgresql</artifactId>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <!--rocketMQ-->
<!--        <dependency>-->
<!--            <groupId>org.apache.rocketmq</groupId>-->
<!--            <artifactId>rocketmq-client</artifactId>-->
<!--            <version>4.4.0</version>-->
<!--        </dependency>-->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.0.4</version>
        </dependency>
    </dependencies>

关于普通消息的生产、消费我准备用一个异步写库的场景（这个在上家公司遇到过，不过当时使用的是阿里云上的消息队列），首先创建相应的Service和其实现类，为了同时模拟同步和异步消息，我决定在新增数据时发送同步消息，更新时发送异步消息，代码如下：

@Slf4j
@Service
public class UserServiceImpl implements UserService {
    
    private static final String USER_TOPIC = "user_topic:";
    private static final String UPDATE_TAG = "update_tag";
    private static final String INSERT_TAG = "insert_tag";
    private RocketMQTemplate rocketMQTemplate;
    public UserServiceImpl(RocketMQTemplate rocketMQTemplate) {
        this.rocketMQTemplate = rocketMQTemplate;
    }

    @Override
    public Boolean update(UserEntity user) {
        user.setUpdateTime(new Date());
        rocketMQTemplate.asyncSend(USER_TOPIC + UPDATE_TAG, user, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                String msgId = sendResult.getMsgId();
                log.info(">>>> async message success, send status={},message id={} <<<<",sendResult.getSendStatus().name(),msgId);
                // 其他
            }
            @Override
            public void onException(Throwable throwable) {
                log.error(">>>> async message success, exception message={} <<<<",throwable.getMessage());
                // 其他
            }
        });
        return Boolean.TRUE;
    }

    @Override
    public Boolean insert(UserEntity user) {
        Date now = new Date();
        user.setCreateTime(now);
        user.setUpdateTime(now);
        SendResult sendResult = rocketMQTemplate.syncSend(USER_TOPIC + INSERT_TAG,user);
        log.info(">>>> send message success, send status={} <<<<",sendResult.getSendStatus().name());
        if (sendResult.getSendStatus().name().equals(SendStatus.SEND_OK.name())) {
            return Boolean.TRUE;
        }
        return Boolean.FALSE;
    }
}

之所以选择使用RocketMq的starter，就是因为它在官方的基础之上又进行了一层的封装，使用起来非常的方便，消息的发送通过RocketMQTemplate就可以了。RocketMq有几个核心的概念分别是：Producer Group、Consumer Group、Topic、Tag等需要提前了解一下，这里简单说一下Producer Group即生产者组，就是相同类型消息生产者的集合，如果这个组的某一个生产者挂掉，那么其这个组的其他生产者依然可以提供服务。Consumer Group即消费者组，同Producer Group类似。Topic，即主题，也就是生产者发生消息以及消费者拉取消息的地方，它和生产者、消费者的关联比较松散。Tag也叫子主题，相对与Topic而言它的粒度更小一些，它可以让用户对消息处理更精细一点。在发送消息时的目的地址就是Topic:Tag，当然如果也可以使用*代替Tag，这样消息会发送到所有订阅该主题的消费者。
这里使用了RocketMQTemplate发送消息，消息发出去了，该如何接收消息呢，starter提供了RocketMQListener接口，该接口主要是普通消息的接收，如果是事务消息，则需实现RocketMQLocalTransactionListener接口。我编写两个消费者，消费不同的Tag，代码如下：

## 订阅user_topic  insert_tag的消息监听器，注意consumerGroup名称
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "user_group",topic = "user_topic",selectorType = SelectorType.TAG,selectorExpression = "insert_tag")
public class InsertUserMessageListener implements RocketMQListener<UserEntity> {

    private UserRepository userRepository;

    private static Gson GSON = new Gson();

    public InsertUserMessageListener(UserRepository userRepository) {
        this.userRepository = userRepository;
    }

    @Override
    public void onMessage(UserEntity userEntity) {
        log.info(">>>> user message listener start,Tag=insert_tag, message={} <<<<",GSON.toJson(userEntity));
        // 写入数据库
        userRepository.save(userEntity);
    }
}
## 订阅user_topic  update_tag的消息监听器，注意consumerGroup名称
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "user_group",topic = "user_topic",selectorType = SelectorType.TAG,selectorExpression = "update_tag")
public class UpdateUserMessageListener implements RocketMQListener<UserEntity> {

    private UserRepository userRepository;
    private static Gson GSON = new Gson();

    public UpdateUserMessageListener(UserRepository userRepository) {
        this.userRepository = userRepository;
    }

    @Override
    public void onMessage(UserEntity userEntity) {
        log.info(">>>> user message listener start,Tag=update_tag, message={} <<<<",GSON.toJson(userEntity));
        // 写入数据库
        userRepository.save(userEntity);
    }
}

好了，接下来启动本地的RocketMq进行测试。

二、测试消息的发送、接收

RocketMq其实对内存占用是非常高的，JVM堆内存大小在4G到8G，个人觉得除非你的内存真的很大，不然还是改小一些比较好。主要是修改bin目录下的runserver.sh和runbroker.sh(windows系统修改runserver.cmd和runbroker.cmd)，然后分别启动nameserver和broker

## 启动namesrv
 nohup sh bin/mqnamesrv &
## 启动broker
nohup sh bin/mqbroker -n localhost:9876 -c conf/broker.conf  &

启动好RocketMq之后，需要在项目的配置文件配置好RocketMq的相关配置，application.properties如下：

spring.application.name=rocketMQ-service

spring.datasource.url=jdbc:postgresql://localhost:5432/pgsql?useSSL=false&characterEncoding=utf8
spring.datasource.username=postgres
spring.datasource.password=123456
spring.datasource.driver-class-name=org.postgresql.Driver

spring.jpa.show-sql=true
spring.jpa.database=postgresql
spring.jpa.hibernate.ddl-auto=update
spring.jpa.hibernate.naming.physical-strategy=org.hibernate.boot.model.naming.PhysicalNamingStrategyStandardImpl
spring.jpa.properties.hibernate.dialect=org.hibernate.dialect.PostgreSQLDialect
spring.jpa.properties.hibernate.temp.use_jdbc_metadata_defaults=false

## rocketmq config
rocketmq.name-server=localhost:9876
rocketmq.producer.group=rocket_group
rocketmq.producer.send-message-timeout=10000

主要配置了rocketMq的namesrv地址和生产者组名以及发送消息超时时间，启动项目，并调用分别执行新增和更新的接口。
但是在项目启动过程中遇到了问题，导致项目启动失败，异常信息如下：

java.lang.RuntimeException: java.lang.IllegalStateException: Failed to start RocketMQ push consumer
    at org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration.registerContainer(ListenerContainerConfiguration.java:123) ~[rocketmq-spring-boot-2.0.4.jar:2.0.4]
    at java.util.LinkedHashMap.forEach(LinkedHashMap.java:684) ~[na:1.8.0_171]
    at org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration.afterSingletonsInstantiated(ListenerContainerConfiguration.java:82) ~[rocketmq-spring-boot-2.0.4.jar:2.0.4]
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:910) ~[spring-beans-5.2.7.RELEASE.jar:5.2.7.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:879) ~[spring-context-5.2.7.RELEASE.jar:5.2.7.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:551) ~[spring-context-5.2.7.RELEASE.jar:5.2.7.RELEASE]
    at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:143) ~[spring-boot-2.3.1.RELEASE.jar:2.3.1.RELEASE]
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:758) [spring-boot-2.3.1.RELEASE.jar:2.3.1.RELEASE]
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:750) [spring-boot-2.3.1.RELEASE.jar:2.3.1.RELEASE]
    at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:397) [spring-boot-2.3.1.RELEASE.jar:2.3.1.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:315) [spring-boot-2.3.1.RELEASE.jar:2.3.1.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1237) [spring-boot-2.3.1.RELEASE.jar:2.3.1.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1226) [spring-boot-2.3.1.RELEASE.jar:2.3.1.RELEASE]
    at com.ypc.mq.rocketmq.RocketmqApplication.main(RocketmqApplication.java:10) [classes/:na]
Caused by: java.lang.IllegalStateException: Failed to start RocketMQ push consumer
    at org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer.start(DefaultRocketMQListenerContainer.java:277) ~[rocketmq-spring-boot-2.0.4.jar:2.0.4]
    at org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration.registerContainer(ListenerContainerConfiguration.java:120) ~[rocketmq-spring-boot-2.0.4.jar:2.0.4]
    ... 13 common frames omitted
Caused by: org.apache.rocketmq.client.exception.MQClientException: The consumer group[user_group] has been created before, specify another name please.
See http://rocketmq.apache.org/docs/faq/ for further details.
    at org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl.start(DefaultMQPushConsumerImpl.java:628) ~[rocketmq-client-4.5.2.jar:4.5.2]
    at org.apache.rocketmq.client.consumer.DefaultMQPushConsumer.start(DefaultMQPushConsumer.java:693) ~[rocketmq-client-4.5.2.jar:4.5.2]
    at org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer.start(DefaultRocketMQListenerContainer.java:275) ~[rocketmq-spring-boot-2.0.4.jar:2.0.4]
    ... 14 common frames omitted

首先根据报错信息The consumer group[user_group] has been created before, specify another name please.可以得出一个结论，那就是消费者组的名称需要唯一，至少在一个项目中要唯一。所以先修改UpdateUserMessageListener消费者的消费者组名称为user_update_group。这样的话如果订阅相同Topic不同的Tag就需要指定不同的消费者组名称，我之前的理解是，一个Topic一个消费者组，看来这种理解是错误的，而应该是一个Topic:Tag一个消费者组。因为消息发送的地址是Topic:Tag，那么显即使Topic相同但Tag不同的话，其实消费者也是不同的，自然消费者的组也就不同。也就是说消费者的分组归根到底是按照Topic:Tag来区分的，我之所以认为是按照Topic，是因为之前 消费者中Tag定义的都是*（默认值），所以才有这样的误解。
修改名称之后，重新启动项目，首先调用新增数据借口，即发送一个同步消息，日志如下：

2020-07-18 17:49:52.977  INFO 21075 --- [nio-9090-exec-1] c.y.m.r.service.impl.UserServiceImpl     : >>>> send message success, send status=SEND_OK <<<<
2020-07-18 17:49:52.983  INFO 21075 --- [MessageThread_2] c.y.m.r.l.InsertUserMessageListener      : >>>> user message listener start,Tag=insert_tag, message={"id":0,"userName":"李四","mobile":"12364455","sex":"男","birthday":"Apr 3, 2015 12:15:14 PM","createTime":"Jul 18, 2020 5:49:50 PM","updateTime":"Jul 18, 2020 5:49:50 PM"} <<<<
Hibernate: select nextval ('hibernate_sequence')
Hibernate: insert into t_user (birthday, createTime, email, mobile, sex, updateTime, userName, id) values (?, ?, ?, ?, ?, ?, ?, ?)

这里开始自己也有点迷惑，为什么在写库之前返回的状态就是SEND_OK呢？看一下官方文档的解释：

When sending a message, you will get SendResult which contains SendStatus. Firstly, we assume that Message’s isWaitStoreMsgOK=true(default is true). If not, we will always get SEND_OK if no exception is thrown.
...
SEND_OK does not mean it is reliable. To make sure no message would be lost, you should also enable SYNC_MASTER or SYNC_FLUSH.

这里解释的很清楚了，如果在没有修改isWaitStoreMsgOK配置的情况下，只要发送的消息成功写到磁盘，也就是消息持久化到broker的服务器，就会返回SEND_OK，至于消费者有没有消费，生产者并不关心，生产者在消息落地之后，其任务就已经完成。不过如果是事务消息结果应该就不同了。
然后再调用一个异步消息的接口，日志如下：

2020-07-18 18:03:51.677  INFO 21075 --- [ublicExecutor_4] c.y.m.r.service.impl.UserServiceImpl     : >>>> async message success, send status=SEND_OK,message id=C0A80069525318B4AAC25B6C69F7000B <<<<
2020-07-18 18:03:51.692  INFO 21075 --- [MessageThread_1] c.y.m.r.l.UpdateUserMessageListener      : >>>> user message listener start,Tag=update_tag, message={"id":19,"userName":"李sisi","mobile":"88844111","sex":"男","birthday":"Apr 3, 2015 12:15:14 PM","updateTime":"Jul 18, 2020 6:03:51 PM"} <<<<
Hibernate: select userentity0_.id as id1_1_0_, userentity0_.birthday as birthday2_1_0_, userentity0_.createTime as createti3_1_0_, userentity0_.email as email4_1_0_, userentity0_.mobile as mobile5_1_0_, userentity0_.sex as sex6_1_0_, userentity0_.updateTime as updateti7_1_0_, userentity0_.userName as username8_1_0_ from t_user userentity0_ where userentity0_.id=?
Hibernate: update t_user set birthday=?, createTime=?, email=?, mobile=?, sex=?, updateTime=?, userName=? where id=?

其实不管是新增还是修改操作，其实使用的都是相同的save方法，所以可以将两个消息监听器合并成一个。另外RocketMQMessageListener默认使用的消息模式为Clustering，即集群模式，当然还有Broadcasting，即广播模式，广播模式下所有的消费着都会收到相同的消息。使用广播模式的好处就是可以自动的实现负载均衡，一般场景还是建议使用集群模式，但是如果又真的想使用广播模式的话需要最好通过集群模式来实现，不建议直接使用广播模式。
将上面的两个消息监听器合并，并创建一个新的消息监听器模拟广播模式，代码如下：

## 监听多个tag，tag使用'||'隔开
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "user_group",topic = "user_topic",selectorType = SelectorType.TAG,selectorExpression = "insert_tag || update_tag")
public class UserMessageListener implements RocketMQListener<UserEntity> {

    private UserRepository userRepository;

    private static Gson GSON = new Gson();

    public UserMessageListener(UserRepository userRepository) {
        this.userRepository = userRepository;
    }

    @Override
    public void onMessage(UserEntity userEntity) {
        log.info(">>>> user message listener start,Tag=insert_tag, message={} <<<<",GSON.toJson(userEntity));
        // 写入数据库
        userRepository.save(userEntity);
    }
}
## 另一个监听器，模拟广播模式，默认是所有的  tag
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "broad_group",topic = "user_topic")
public class BroadMessageListener implements RocketMQListener<UserEntity> {

    private UserRepository userRepository;
    private static Gson GSON = new Gson();

    public BroadMessageListener(UserRepository userRepository) {
        this.userRepository = userRepository;
    }

    @Override
    public void onMessage(UserEntity userEntity) {
        log.info(">>>> broadcasting model listener start, message={} <<<<",GSON.toJson(userEntity));
        // 处理
        userRepository.save(userEntity);
    }
}

使用集群模式模拟广播模式，只要监听的Topic:Tag一致即可。重新启动项目，并发送一个新增接口，日志如下：

2020-07-18 18:35:21.326  INFO 23280 --- [essageThread_11] c.y.m.r.listener.BroadMessageListener    : >>>> broadcasting model listener start, message={"id":0,"userName":"王五","mobile":"25464455","sex":"男","birthday":"Apr 3, 2010 12:15:14 PM","createTime":"Jul 18, 2020 6:35:19 PM","updateTime":"Jul 18, 2020 6:35:19 PM"} <<<<
Hibernate: select nextval ('hibernate_sequence')
Hibernate: insert into t_user (birthday, createTime, email, mobile, sex, updateTime, userName, id) values (?, ?, ?, ?, ?, ?, ?, ?)
2020-07-18 18:35:21.333  INFO 23280 --- [nio-9090-exec-1] c.y.m.r.service.impl.UserServiceImpl     : >>>> send message success, send status=SEND_OK <<<<
2020-07-18 18:35:21.342  INFO 23280 --- [MessageThread_2] c.y.m.r.listener.UserMessageListener     : >>>> user message listener start,Tag=insert_tag, message={"id":0,"userName":"王五","mobile":"25464455","sex":"男","birthday":"Apr 3, 2010 12:15:14 PM","createTime":"Jul 18, 2020 6:35:19 PM","updateTime":"Jul 18, 2020 6:35:19 PM"} <<<<
Hibernate: select nextval ('hibernate_sequence')
Hibernate: insert into t_user (birthday, createTime, email, mobile, sex, updateTime, userName, id) values (?, ?, ?, ?, ?, ?, ?, ?)

可以看到有两个消费者都有日志输出，且有两条插入sql，查看数据库也有两条完全一样的数据，说明模拟实现广播模式是成功的。

三、总结

本次的内容主要是学习在Spring Boot下使用rocketmq-spring-boot-starter如何进行普通消息的发送、接收以及如何使用集群模式来实现广播模式。这次学习自己对RocketMq的一些核心概念也有了进一步的了解，收获还是蛮多的，当然RocketMq内容还是很多的，通过一两个demo来搞清楚是不现实的，以后需要更多时间和精力来学习，因为时间的问题自己也没有能尝试使用它的事务消息，等之后有时间再通过demo来进一步学习吧。

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