Springboot整合Rocketmq系列教程

本教程是基于Springboot2.6.3整合Rocketmq5.0,其中涉及了Rocketmq的安装,消息的发送及消费的代码实现。

本文不会对rocketmq的一些概念、原理,及注意事项进行讲解,因为官网已经写的很清楚,又有中文版,详细访问https://rocketmq.apache.org/zh/docs/quickStart/01quickstart

1.安装

1.1window下启动rocketmq

window下启动rocketmq需要注意的地方有两个:

  1. 环境变量的配置
  2. jdk的版本,例如jdk版本太高了,怎么办

1.1.1下载最新版

打开rocketmq的官网:https://rocketmq.apache.org/,点击下载按钮

在这里插入图片描述

然后解压放到相应的目录

1.1.2启动mqnamesrv

来到解压的目录,打开E:\rocketmq-all-5.1.0\bin,找到mqnamesrv.cmd

在这里插入图片描述

在启动之前还有两个事情要做:

第一、配置rocketmq环境变量

我们打开它的启动文件,内容如下

if not exist "%ROCKETMQ_HOME%\bin\runserver.cmd" echo Please set the ROCKETMQ_HOME variable in your environment! & EXIT /B 1

call "%ROCKETMQ_HOME%\bin\runserver.cmd" -Drmq.logback.configurationFile=%ROCKETMQ_HOME%\conf\rmq.namesrv.logback.xml org.apache.rocketmq.namesrv.NamesrvStartup %*

IF %ERRORLEVEL% EQU 0 (
    ECHO "Namesrv starts OK"
)

首先是配置ROCKETMQ_HOME环境变量,也就是rocketmq所在的目录。

从内容我们发现最终启动的是runserver.cmd,我们也看看它的内容

if not exist "%JAVA_HOME%\bin\java.exe" echo Please set the JAVA_HOME variable in your environment, We need java(x64)! & EXIT /B 1
set "JAVA=%JAVA_HOME%\bin\java.exe"

setlocal

set BASE_DIR=%~dp0
set BASE_DIR=%BASE_DIR:~0,-1%
for %%d in (%BASE_DIR%) do set BASE_DIR=%%~dpd

set CLASSPATH=.;%BASE_DIR%conf;%BASE_DIR%lib\*;%CLASSPATH%

set "JAVA_OPT=%JAVA_OPT% -server -Xms2g -Xmx2g -Xmn1g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
set "JAVA_OPT=%JAVA_OPT% -XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSInitiatingOccupancyFraction=70 -XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8 -XX:-UseParNewGC"
set "JAVA_OPT=%JAVA_OPT% -verbose:gc -Xloggc:"%USERPROFILE%\rmq_srv_gc.log" -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
set "JAVA_OPT=%JAVA_OPT% -XX:-OmitStackTraceInFastThrow"
set "JAVA_OPT=%JAVA_OPT% -XX:-UseLargePages"
set "JAVA_OPT=%JAVA_OPT% %JAVA_OPT_EXT% -cp "%CLASSPATH%""

"%JAVA%" %JAVA_OPT% %*

这里需要配置JAVA_HOME环境变量,java所在的目录

第二、检查jdk版本

rocketmq推荐的版本是java8,而我本机配置的环境变量是jdk17的,如果用17去启动会报错,因此为了简单点,

if not exist "D:\Program Files\jdk1.8\bin\java.exe" echo Please set the JAVA_HOME variable in your environment, We need java(x64)! & EXIT /B 1
set "JAVA=D:\Program Files\jdk1.8\bin\java.exe"

然后打开cmd,运行

E:\rocketmq-all-5.1.0\bin>mqnamesrv.cmd

如果出现以下的信息,就说明启动成功

The Name Server boot success. serializeType=JSON, address 0.0.0.0:9876

1.1.3启动broker

同样的,在mqnamesrv.cmd所在的目录里面我们会发现有mqbroker.cmd和runbroker.cmd这个两个文件,我们只需注意ROCKETMQ_HOME和JAVA_HOME两个环境变量即可。

ROCKETMQ_HOME已经在上面一步完成了,接下来只需把涉及到JAVA_HOME的地方替换成“D:\Program Files\jdk1.8”即可

我们打开cmd,运行以下命令

E:\rocketmq-all-5.1.0\bin>mqbroker.cmd -n localhost:9876
The broker[DESKTOP-ED6R0PI, 198.18.0.1:10911] boot success. serializeType=JSON and name server is localhost:9876

说明broker也启动成功了

这样window下启动rocketmq就完成了

1.2linux下启动rocketmq

linux下启动rocketmq主要的问题也是两个:

  1. jdk版本,上面已经提到过了
  2. 内存大小的问题

1.2.1检查jdk版本

[root@centos7 ~]# java -version
java version "1.8.0_212"
Java(TM) SE Runtime Environment (build 1.8.0_212-b10)
Java HotSpot(TM) 64-Bit Server VM (build 25.212-b10, mixed mode)

jdk是1.8满足安装rocketmq的要求

1.2.2下载rocketmq

[root@centos7 ~]# cd /opt/software/
[root@centos7 software]# wget https://archive.apache.org/dist/rocketmq/5.1.0/rocketmq-all-5.1.0-bin-release.zip

我把下载的压缩包放到software目录下,然后解压到module目录下

[root@centos7 software]# unzip rocketmq-all-5.1.0-bin-release.zip -d /opt/module
[root@centos7 software]# cd /opt/module/
[root@centos7 module]# ll
drwxr-xr-x.  6 root root 120 410 16:28 rocketmq-all-5.1.0-bin-release

1.2.3启动mqnamesrv

启动之前要修改一个文件,打开runserver.sh

在这里插入图片描述

因为虚拟机的内容没有那么大,上面是我修改后的值,改完之后使用以下命令启动

[root@centos7 rocketmq-all-5.1.0-bin-release]# nohup sh bin/mqnamesrv &
[1] 13155
[root@centos7 rocketmq-all-5.1.0-bin-release]# nohup: 忽略输入并把输出追加到"nohup.out"

1.2.4启动broker

同样,启动之前也要修改一个文件,打开runbroker.sh

同样的,框起来的值是我修改后的值

然后用以下命令启动

[root@centos7 rocketmq-all-5.1.0-bin-release]# nohup sh bin/mqbroker -n localhost:9876 &
[2] 13258
[root@centos7 rocketmq-all-5.1.0-bin-release]# nohup: 忽略输入并把输出追加到"nohup.out"

最后我们使用jps命令看看刚才的是否成功启动服务

[root@centos7 rocketmq-all-5.1.0-bin-release]# jps
13314 Jps
13289 BrokerStartup
13182 NamesrvStartup

可以看到服务已经启动成功

这样linux下启动rocketmq就完成了

1.3 RocketMQ仪表板

RocketMQ Dashboard是一个Web界面应用程序,用于管理和监控Apache RocketMQ消息中间件的运行状态。RocketMQ Dashboard提供了一系列的监控指标和管理功能,包括Broker配置、集群概览、主题状态、消费者状态、生产者状态和消息跟踪等。

使用RocketMQ Dashboard,用户可以方便地查看RocketMQ集群的整体状态和各个组件的运行情况,以及执行一些管理操作,如创建和删除主题、查看消费者组订阅信息、发送测试消息等。

打开https://rocketmq.apache.org/download#rocketmq-dashboard,直接下载源码到本地部署
在这里插入图片描述

用idea打开,修改配置文件application.yml

rocketmq:
  config:
    namesrvAddrs:
      - 192.168.10.100:9876
      - 192.168.10.100:9876

namesrvAddrs地址就是服务的地址,192.168.10.100是我虚拟主机的ip,如果是window环境下,改成localhost

然后启动程序,访问http://localhost:8080/#/

在这里插入图片描述

可以在上面添加主题,查看发送的消息

2.准备工作

2.1引入依赖

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.3</version>
</dependency>

2.2配置文件

server:
  port: 9002

rocketmq:
  # 服务地址,多个用逗号分开
  name-server: 192.168.10.100:9876
  producer:
    # 发送消息超时时间,默认3000
    send-message-timeout: 30000
    # 生产者组
    group: group1
    # 发送消息失败重试次数,默认2
    retryTimesWhenSendFailed: 2
    # 异步消息重试此处,默认2
    retryTimesWhenSendAsyncFailed: 2
    # 消息最大长度,默认1024 * 1024 * 4(默认4M)
    maxMessageSize: 4096
    # 压缩消息阈值,默认4k(1024 * 4)
    compressMessageBodyThreshold: 4096
    # 是否在内部发送失败时重试另一个broker,默认false
    retryNextServer: false

2.3创建RocketMQTemplate

封装了RocketMQ的生产者API,提供了一系列简单易用的方法,用于发送不同类型的消息, 后面的例子都会用到它

我们要新建一个ExtRocketMQTemplate类去继承RocketMQTemplate

@ExtRocketMQTemplateConfiguration(nameServer = "${rocketmq.name-server}")
public class ExtRocketMQTemplate extends RocketMQTemplate {
}

后面发送消息都用这个类

2.4创建一个主题

利用上面的RocketMQ仪表盘新建一个主题,后面用得到

在这里插入图片描述

3.发送消息

3.1普通消息

普通消息为 Apache RocketMQ 中最基础的消息 , 普通消息一般应用于微服务解耦、事件驱动、数据集成等场景,这些场景大多数要求数据传输通道具有可靠传输的能力,且对消息的处理时机、处理顺序没有特别要求。

3.1.1单向消息

指发送消息后,不需要等待Broker的响应,直接返回。这种方式适用于不需要关注消息发送结果的场景,如日志记录、统计信息等。

@Resource(name = "extRocketMQTemplate")
private RocketMQTemplate extRocketMQTemplate;

public void sendOneWay(){
    HeroDTO heroDTO = new HeroDTO();
    heroDTO.setId("jialuo");
    heroDTO.setName("伽罗");

    Message<String> msgs = MessageBuilder.withPayload(JSON.toJSONString(heroDTO))
        //设置消息KEYS,一般是数据的唯一ID,主要用于在仪表盘中方便搜索
        .setHeader("KEYS", heroDTO.getId())
        .build();
	
    //给消息打上射手的标签。主题+tag,中间用“:”分隔,主要是用于消息的过滤,比如说在消费的时候,只消费ESS标签下的消息
    String topic = "kingsTopic".concat(":shooter");
    extRocketMQTemplate.sendOneWay(topic, msgs);
}

3.1.2同步发送消息

syncSend方法会阻塞当前线程,直到消息发送完成并收到了消息服务器的响应。如果消息发送成功,syncSend方法会返回一个SendResult对象,包含了消息的发送状态、消息ID等信息。如果消息发送失败,syncSend方法会抛出一个MessagingException异常。

public void syncSend(){
    HeroDTO heroDTO = new HeroDTO();
    heroDTO.setId("luban");
    heroDTO.setName("小鲁班");

    Message<String> msgs = MessageBuilder.withPayload(JSON.toJSONString(heroDTO))
        .setHeader("KEYS", heroDTO.getId()).build();

    SendResult sendResult = extRocketMQTemplate.syncSend("kingsTopic".concat(":shooter"), msgs);

    log.info(sendResult.toString());
}

3.1.3异步发送消息

asyncSend方法不会阻塞当前线程,而是在另一个线程中异步发送消息。因此,asyncSend方法会立即返回,不会等待消息发送完成。如果需要等待消息发送完成并处理发送结果,可以使用SendCallback回调接口。

public void asyncSend(){
    HeroDTO heroDTO = new HeroDTO();
    heroDTO.setId("sunshangxiang");
    heroDTO.setName("孙尚香");

    Message<String> msgs = MessageBuilder.withPayload(JSON.toJSONString(heroDTO)).setHeader("KEYS", heroDTO.getId()).build();

    extRocketMQTemplate.asyncSend("kingsTopic".concat(":shooter"), msgs, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
            log.info(sendResult.toString());
        }

        @Override
        public void onException(Throwable e) {
            log.info(e.getMessage());
        }
    });
}

3.2顺序消息

顺序消息是 Apache RocketMQ 提供的一种高级消息类型,支持消费者按照发送消息的先后顺序获取消息,从而实现业务场景中的顺序处理。 相比其他类型消息,顺序消息在发送、存储和投递的处理过程中,更多强调多条消息间的先后顺序关系。

3.2.1单向顺序消息

public void sendOneWayOrderly(){
    HeroDTO heroDTO = new HeroDTO();
        heroDTO.setId("zhangfei");
        heroDTO.setName("张飞");

        Message<String> msgs = MessageBuilder.withPayload(JSON.toJSONString(heroDTO))
            .setHeader("KEYS", heroDTO.getId()).build();
		
    //顺序消息比普通消息多一个参数,第三个参数只要唯一就行,比如ID
        extRocketMQTemplate.sendOneWayOrderly("kingsTopic".concat(":tank"), msgs,heroDTO.getId());
}

3.2.2同步发送顺序消息

public void syncSendOrderly(){
    HeroDTO heroDTO = new HeroDTO();
    heroDTO.setId("niumo");
    heroDTO.setName("牛魔");

    Message<String> msgs = MessageBuilder.withPayload(JSON.toJSONString(heroDTO))
        .setHeader("KEYS", heroDTO.getId()).build();

    SendResult sendResult = extRocketMQTemplate.syncSendOrderly("kingsTopic".concat(":tank"), msgs, heroDTO.getId());
    log.info(sendResult.toString());
}

3.2.3异步发送顺序消息

public void asyncSendOrderly(){
    HeroDTO heroDTO = new HeroDTO();
    heroDTO.setId("xiangyu");
    heroDTO.setName("项羽");

    Message<String> msgs = MessageBuilder.withPayload(JSON.toJSONString(heroDTO)).setHeader("KEYS", heroDTO.getId()).build();

    extRocketMQTemplate.asyncSendOrderly("kingsTopic".concat(":tank"), msgs, heroDTO.getId(), new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
            log.info(sendResult.toString());
        }

        @Override
        public void onException(Throwable e) {
            log.info(e.getMessage());
        }
    });
}

3.3定时/延时消息

定时消息是 Apache RocketMQ 提供的一种高级消息类型,消息被发送至服务端后,在指定时间后才能被消费者消费。通过设置一定的定时时间可以实现分布式场景的延时调度触发效果。

3.3.1发送延迟消息

场景一:任务超时处理

public void syncSendDelayTimeSeconds(){
    HeroDTO heroDTO = new HeroDTO();
    heroDTO.setId("baiqi");
    heroDTO.setName("白起");

    Message<String> msgs = MessageBuilder.withPayload(JSON.toJSONString(heroDTO)).setHeader("KEYS", heroDTO.getId()).build();

    //10秒后才能消费这条消息
    SendResult sendResult = extRocketMQTemplate.syncSendDelayTimeSeconds("kingsTopic".concat(":tank"), msgs, 10L);
    //10毫秒后才能消费这条消息
    //extRocketMQTemplate.syncSendDelayTimeMills("kingsTopic".concat(":tank"), msgs, 10L);
    log.info(sendResult.toString());
}

场景二:定时处理

public void syncSendDeliverTimeMills(){
    HeroDTO heroDTO = new HeroDTO();
    heroDTO.setId("liubang");
    heroDTO.setName("刘邦");

    Message<String> msgs = MessageBuilder.withPayload(JSON.toJSONString(heroDTO)).setHeader("KEYS", heroDTO.getId()).build();

    //每天凌晨处理
    long time = LocalDate.now().atStartOfDay().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
    SendResult sendResult = extRocketMQTemplate.syncSendDeliverTimeMills("kingsTopic".concat(":tank"), msgs, time);

    log.info(sendResult.toString());
}

所有任务都在每天凌晨0点统一处理

3.4事务消息

事务消息是 Apache RocketMQ 提供的一种高级消息类型,支持在分布式场景下保障消息生产和本地事务的最终一致性

 public void sendMessageInTransaction() throws InterruptedException {
     HeroDTO heroDTO = new HeroDTO();
     heroDTO.setId("houyi");
     heroDTO.setName("后裔");

     Message<String> msgs = MessageBuilder.withPayload(JSON.toJSONString(heroDTO))
         .setHeader("KEYS", heroDTO.getId())
         //设置事务ID
         .setHeader(RocketMQHeaders.TRANSACTION_ID,"KEY_"+heroDTO.getId())
         .build();

     TransactionSendResult transactionSendResult = extRocketMQTemplate.sendMessageInTransaction("kingsTopic".concat(":shooter"), msgs, null);
     log.info(transactionSendResult.toString());
 }

另外还需要创建监听器

@RocketMQTransactionListener(rocketMQTemplateBeanName = "extRocketMQTemplate")
@Slf4j
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);

    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<String, Integer>();

    /**
     *执行事务
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        //事务ID
        String transId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);

        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        assert transId != null;
        localTrans.put(transId, status);

        if (status == 0) {
            log.info("success");
            //成功,提交事务
            return RocketMQLocalTransactionState.COMMIT;
        }

        if (status == 1) {
            log.info("failure");
            //失败,回滚事务
            return RocketMQLocalTransactionState.ROLLBACK;
        }

        log.info("unknown");
        //中间状态
        return RocketMQLocalTransactionState.UNKNOWN;
    }

    /**
     *检查事务状态
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        String transId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        RocketMQLocalTransactionState retState = RocketMQLocalTransactionState.COMMIT;
        Integer status = localTrans.get(transId);
        if (null != status) {
            switch (status) {
                case 0:
                    retState = RocketMQLocalTransactionState.COMMIT;
                    break;
                case 1:
                    retState = RocketMQLocalTransactionState.ROLLBACK;
                    break;
                case 2:
                    retState = RocketMQLocalTransactionState.UNKNOWN;
                    break;

                default: break;
            }
        }
        log.info("msgTransactionId:{},TransactionState:{},status:{}",transId,retState,status);
        return retState;
    }
}

3.5批量发送

public void syncSendBatchMessage() throws InterruptedException {
    List<HeroDTO> heroList = new ArrayList<>();
    HeroDTO heroDTO = new HeroDTO();
    heroDTO.setId("geya");
    heroDTO.setName("戈娅");
    heroList.add(heroDTO);

    HeroDTO heroDTO1 = new HeroDTO();
    heroDTO.setId("direnjie");
    heroDTO.setName("狄仁杰");
    heroList.add(heroDTO1);

    List<Message> msgs = new ArrayList<Message>();
    for (HeroDTO hero : heroList){
        Message<String> message = MessageBuilder.withPayload(JSON.toJSONString(heroDTO))
            .setHeader("KEYS", heroDTO.getId())
            .build();
        msgs.add(message);
    }

    SendResult sendResult = extRocketMQTemplate.syncSend("kingsTopic".concat(":shooter"), msgs);
    log.info(sendResult.toString());
}

4.消费消息

消费消息的代码就很简单,新建一个HeroConsumer,继承RocketMQListener监听器

@Service
@Slf4j
@RocketMQMessageListener(topic = "kingsTopic", consumerGroup = "hero",selectorExpression = "shooter")
public class HeroConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        log.info(message);
    }
}

接下来了解一下@RocketMQMessageListener这个注解

topic:主题,指消费者组订阅的消息服务

consumerGroup:消费者组,一个组可以有多个消费者,主要的作用是集群模式负载均衡的实现,广播模式的通知的实现

consumeModel:控制消费模式,你可以选择并发或有序接收消息

messageModel:控制消息模式,广播模式-所有消费者都能接收到信息, 集群模式:无论有多少个消费者,只有一个消费者能够接收到信息,也就是说消息一旦被消费了,其它消费者就不能消费该条消息

selectorExpression:选择哪个标签(tag)下的信息,默认是消费该主题下的所有信息

源码:https://gitee.com/myha/spring-cloud-study-demo/tree/master/rocketmq

Logo

开放原子开发者工作坊旨在鼓励更多人参与开源活动,与志同道合的开发者们相互交流开发经验、分享开发心得、获取前沿技术趋势。工作坊有多种形式的开发者活动,如meetup、训练营等,主打技术交流,干货满满,真诚地邀请各位开发者共同参与!

更多推荐