關(guān)于消息隊列這部分內(nèi)容，之前自己簡單了解和學(xué)習(xí)過Kafka，但是目前國內(nèi)公司對RocketMq使用的也比較多，并且我所在的項(xiàng)目組已經(jīng)引入了RocketMq，但是因?yàn)轫?xiàng)目進(jìn)度問題，在開發(fā)中還沒有正式的使用。所以自己準(zhǔn)別先提前了解一下，關(guān)于RocketMq我想不用我過多介紹，而且性能據(jù)說也是特別的好，并且也支持專門的事務(wù)消息，可以說非常的有吸引力。學(xué)習(xí)的資料無非也就是官方文檔。這次學(xué)習(xí)使用的是4.4.0版本。
下面就正式開始今天的RocketMq學(xué)習(xí)，當(dāng)然首先還是創(chuàng)建demo項(xiàng)目

一、創(chuàng)建項(xiàng)目

今天的主要目的就是學(xué)習(xí)如何在Spring Boot中使用RocketMq，包括消息的生產(chǎn)、消費(fèi)（同步、異步）。當(dāng)然如果時間允許的話我也會嘗試使用消息廣播和事務(wù)消息的發(fā)送、消費(fèi)。
首先創(chuàng)建項(xiàng)目，引入相關(guān)依賴，這里說一下因?yàn)槭呛?code>Spring Boot整合使用，我并沒有使用官方的依賴，而是使用了starter其對rocketmq-client又做了一層的封裝，個人感覺使用起來更方便，Spring Boot版本使用的是：2.3.1.RELEASE，項(xiàng)目的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>

關(guān)于普通消息的生產(chǎn)、消費(fèi)我準(zhǔn)備用一個異步寫庫的場景（這個在上家公司遇到過，不過當(dāng)時使用的是阿里云上的消息隊列），首先創(chuàng)建相應(yīng)的Service和其實(shí)現(xiàn)類，為了同時模擬同步和異步消息，我決定在新增數(shù)據(jù)時發(fā)送同步消息，更新時發(fā)送異步消息，代碼如下：

@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，就是因?yàn)樗诠俜降幕A(chǔ)之上又進(jìn)行了一層的封裝，使用起來非常的方便，消息的發(fā)送通過RocketMQTemplate就可以了。RocketMq有幾個核心的概念分別是：Producer Group、Consumer Group、Topic、Tag等需要提前了解一下，這里簡單說一下Producer Group即生產(chǎn)者組，就是相同類型消息生產(chǎn)者的集合，如果這個組的某一個生產(chǎn)者掛掉，那么其這個組的其他生產(chǎn)者依然可以提供服務(wù)。Consumer Group即消費(fèi)者組，同Producer Group類似。Topic，即主題，也就是生產(chǎn)者發(fā)生消息以及消費(fèi)者拉取消息的地方，它和生產(chǎn)者、消費(fèi)者的關(guān)聯(lián)比較松散。Tag也叫子主題，相對與Topic而言它的粒度更小一些，它可以讓用戶對消息處理更精細(xì)一點(diǎn)。在發(fā)送消息時的目的地址就是Topic:Tag，當(dāng)然如果也可以使用*代替Tag，這樣消息會發(fā)送到所有訂閱該主題的消費(fèi)者。
這里使用了RocketMQTemplate發(fā)送消息，消息發(fā)出去了，該如何接收消息呢，starter提供了RocketMQListener接口，該接口主要是普通消息的接收，如果是事務(wù)消息，則需實(shí)現(xiàn)RocketMQLocalTransactionListener接口。我編寫兩個消費(fèi)者，消費(fèi)不同的Tag，代碼如下：

## 訂閱user_topic  insert_tag的消息監(jiān)聽器，注意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));
        // 寫入數(shù)據(jù)庫
        userRepository.save(userEntity);
    }
}
## 訂閱user_topic  update_tag的消息監(jiān)聽器，注意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));
        // 寫入數(shù)據(jù)庫
        userRepository.save(userEntity);
    }
}

好了，接下來啟動本地的RocketMq進(jìn)行測試。

二、測試消息的發(fā)送、接收

RocketMq其實(shí)對內(nèi)存占用是非常高的，JVM堆內(nèi)存大小在4G到8G，個人覺得除非你的內(nèi)存真的很大，不然還是改小一些比較好。主要是修改bin目錄下的runserver.sh和runbroker.sh(windows系統(tǒng)修改runserver.cmd和runbroker.cmd)，然后分別啟動nameserver和broker

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

啟動好RocketMq之后，需要在項(xiàng)目的配置文件配置好RocketMq的相關(guān)配置，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地址和生產(chǎn)者組名以及發(fā)送消息超時時間，啟動項(xiàng)目，并調(diào)用分別執(zhí)行新增和更新的接口。
但是在項(xiàng)目啟動過程中遇到了問題，導(dǎo)致項(xiàng)目啟動失敗，異常信息如下：

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

首先根據(jù)報錯信息The consumer group[user_group] has been created before, specify another name please.可以得出一個結(jié)論，那就是消費(fèi)者組的名稱需要唯一，至少在一個項(xiàng)目中要唯一。所以先修改UpdateUserMessageListener消費(fèi)者的消費(fèi)者組名稱為user_update_group。這樣的話如果訂閱相同Topic不同的Tag就需要指定不同的消費(fèi)者組名稱，我之前的理解是，一個Topic一個消費(fèi)者組，看來這種理解是錯誤的，而應(yīng)該是一個Topic:Tag一個消費(fèi)者組。因?yàn)橄l(fā)送的地址是Topic:Tag，那么顯即使Topic相同但Tag不同的話，其實(shí)消費(fèi)者也是不同的，自然消費(fèi)者的組也就不同。也就是說消費(fèi)者的分組歸根到底是按照Topic:Tag來區(qū)分的，我之所以認(rèn)為是按照Topic，是因?yàn)橹?消費(fèi)者中Tag定義的都是*（默認(rèn)值），所以才有這樣的誤解。
修改名稱之后，重新啟動項(xiàng)目，首先調(diào)用新增數(shù)據(jù)借口，即發(fā)送一個同步消息，日志如下：

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 (?, ?, ?, ?, ?, ?, ?, ?)

這里開始自己也有點(diǎn)迷惑，為什么在寫庫之前返回的狀態(tài)就是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配置的情況下，只要發(fā)送的消息成功寫到磁盤，也就是消息持久化到broker的服務(wù)器，就會返回SEND_OK，至于消費(fèi)者有沒有消費(fèi)，生產(chǎn)者并不關(guān)心，生產(chǎn)者在消息落地之后，其任務(wù)就已經(jīng)完成。不過如果是事務(wù)消息結(jié)果應(yīng)該就不同了。
然后再調(diào)用一個異步消息的接口，日志如下：

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=?

其實(shí)不管是新增還是修改操作，其實(shí)使用的都是相同的save方法，所以可以將兩個消息監(jiān)聽器合并成一個。另外RocketMQMessageListener默認(rèn)使用的消息模式為Clustering，即集群模式，當(dāng)然還有Broadcasting，即廣播模式，廣播模式下所有的消費(fèi)著都會收到相同的消息。使用廣播模式的好處就是可以自動的實(shí)現(xiàn)負(fù)載均衡，一般場景還是建議使用集群模式，但是如果又真的想使用廣播模式的話需要最好通過集群模式來實(shí)現(xiàn)，不建議直接使用廣播模式。
將上面的兩個消息監(jiān)聽器合并，并創(chuàng)建一個新的消息監(jiān)聽器模擬廣播模式，代碼如下：

## 監(jiān)聽多個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));
        // 寫入數(shù)據(jù)庫
        userRepository.save(userEntity);
    }
}
## 另一個監(jiān)聽器，模擬廣播模式，默認(rèn)是所有的  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);
    }
}

使用集群模式模擬廣播模式，只要監(jiān)聽的Topic:Tag一致即可。重新啟動項(xiàng)目，并發(fā)送一個新增接口，日志如下：

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 (?, ?, ?, ?, ?, ?, ?, ?)

可以看到有兩個消費(fèi)者都有日志輸出，且有兩條插入sql，查看數(shù)據(jù)庫也有兩條完全一樣的數(shù)據(jù)，說明模擬實(shí)現(xiàn)廣播模式是成功的。

三、總結(jié)

本次的內(nèi)容主要是學(xué)習(xí)在Spring Boot下使用rocketmq-spring-boot-starter如何進(jìn)行普通消息的發(fā)送、接收以及如何使用集群模式來實(shí)現(xiàn)廣播模式。這次學(xué)習(xí)自己對RocketMq的一些核心概念也有了進(jìn)一步的了解，收獲還是蠻多的，當(dāng)然RocketMq內(nèi)容還是很多的，通過一兩個demo來搞清楚是不現(xiàn)實(shí)的，以后需要更多時間和精力來學(xué)習(xí)，因?yàn)闀r間的問題自己也沒有能嘗試使用它的事務(wù)消息，等之后有時間再通過demo來進(jìn)一步學(xué)習(xí)吧。