文章目录

前言
mq介绍
一 rocketmq简单安装[docker]
二 springboot集成rocketmq
三进阶--各消息类型测试
四进阶--rocketmq高可用集群搭建
参考文档

前言

记录rocketmq 4.8.0 docker环境安装及springboot集成, 消息功能测试持续集成,
验证MQ在不同场景的使用,以及由于引入mq带来的问题,rockermq都是怎么解决的

mq介绍

本质上是先进先出的队列
MQ作用及优点
- 系统解耦: 同步调用改成异步调用，下游服务故障不影响上游服务
- 流量削峰: 解决请求激增超过应用最大的qps,导致服务故障
- 数据分发: 不用关心谁使用数据，关注自己服务，谁要使用谁订阅
MQ缺点
- 系统可用性降低(mq服务挂了怎么办–>MQ的高可用怎么保证)
- 系统复杂性提高(重复消费,消息丢失,消息顺序性怎么解决)
- 一致性问题(A发送消息, BC成功，D失败了,怎么保证消息处理一致性–>怎么处理事务呢)

一 rocketmq简单安装[docker]

虚拟机安装docker环境可参考:centos7.4安装docker

1. 获取镜像

#创建RocketMQ使用的共有网络，便于相互访问
docker network create rocketmq_network

#foxiswho/rocketmq 4.7.0以后不再分别创建broker及nameserver的镜像，统一使用rocketmq镜像，只是在启动命令上区分
docker pull foxiswho/rocketmq:4.8.0

#rocketmq控制台2.0.0版本，源码来自于官方仓库https://github.com/apache/rocketmq-externals#rocketmq-console
docker pull 56553655/rocketmq-console-ng:2.0.0

2. 启动容器

#启动rocketmq nameserver
docker run -d --network rocketmq_network --network-alias rmqnamesrv --name rmqnamesrv -e "JAVA_OPT_EXT=-Xms512M -Xmx512M -Xmn128m" -p 9876:9876 foxiswho/rocketmq:4.8.0 sh mqnamesrv

#启动rocketmq broker
docker run -d --network rocketmq_network --network-alias rmqbroker --name rmqbroker -e "NAMESRV_ADDR=rmqnamesrv:9876" -e "JAVA_OPT_EXT=-Xms512M -Xmx512M -Xmn128m" -p 10911:10911 -p 10912:10912 -p 10909:10909 foxiswho/rocketmq:4.8.0 sh mqbroker -c ../conf/broker.conf

#启动rocketmq-console-ng
docker run -d --network rocketmq_network --network-alias rocketmq-console-ng -p 8080:8080 --name rocketmq-console-ng 56553655/rocketmq-console-ng:2.0.0

注意以下三个端口都要开放(简单起见关闭防火墙)
- nameServer对应的9876端口
- broker对应的10911端口
- rocketmq-console-ng对应的8080端口
broker启动指定了配置文件: …/conf/broker.conf

3. 访问控制台

http://192.168.100.70:8080
成功效果
在这里插入图片描述

4. 踩坑处理

broker的默认ip地址，使用docker内部网络地址,需要修改成docker宿主机的IP地址(192.168.100.70),否则后面springboot集成对接时候,会无法连接上.

进入rmqbroker 容器内部修改配置

broker.conf 增加配置brokerIP1 = 192.168.100.70

[root@localhost ~]# docker exec -it rmqbroker bash
[rocketmq@ad5cc337a328 bin]$  cat ../conf/broker.conf
brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
brokerIP1 = 192.168.100.70

退出容器,重启rockermq的broker

docker restart rmqbroker

二 springboot集成rocketmq

1. maven依赖


    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.18.RELEASE</version>
        <relativePath/>
    </parent>
    
    <!--rocketmq-->
     <dependency>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-spring-boot-starter</artifactId>
         <version>2.2.0</version>
     </dependency>

2. application.properties配置

rocketmq.name-server=192.168.100.70:9876
rocketmq.producer.group=my-group
# 默认3000 太小会报RemotingTooMuchRequestException: sendDefaultImpl call timeout错误
rocketmq.producer.send-message-timeout=10000

3. 简单消息测试

基本springboot框架搭建后增加上述,rocketmq配置。测试时候，生产者项目和消费者项目application.properties的rockermq都配都一样, 不同项目端口设置不同即可。

3.1 消息生产者

@Api(tags = "mq接口测试")
@RestController
@RequestMapping("/rocketMq")
public class RocketTestController {
    private static final Logger LOGGER = LoggerFactory.getLogger(RocketTestController.class);

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @PostMapping("/send")
    public ApiResult<String> sendMessage(@RequestParam(defaultValue = "TOPIC_TEST") String topic, String message) {
        rocketMQTemplate.convertAndSend(topic, message);
        return new ApiResult<>();
    }

}

3.2 消息消费者

@Component
@RocketMQMessageListener(topic = "TOPIC_TEST",consumerGroup = "group_consumer")
public class Consumer implements RocketMQListener<String> {
    private static final Logger LOGGER = LoggerFactory.getLogger(Consumer.class);
    @Override 
    public void onMessage(String message) {
        LOGGER.info("Receive message："+message);
    } 
}

3.3 测试

连续往“TOPIC_TEST”topic发送多条消息,发现消息分别被两个消费者随机消费
在这里插入图片描述
结果

三进阶–各消息类型测试

TODO

四进阶–rocketmq高可用集群搭建