微服务架构设计RocketMQ基础及环境整合

网友投稿 870 2022-11-27

微服务架构设计RocketMQ基础及环境整合

微服务架构设计RocketMQ基础及环境整合

目录概述&选型单机安装配置双机主从高可用搭建启动多个NameServer 和 Broker重要参数说明可视化管理平台SpringBoot整合RocketMQ引入组件rocketmq-spring-boot-starter 依赖修改application.yml,添加RocketMQ相关配置编写消息生产者 MessageProduce编写消息消费者 MessageConsumer编写单元测试发送消息测试

概述&选型

消息队列作为高并发系统的核心组件之一,能够帮助业务系统解构提升开发效率和系统稳定性。主要用于三种典型场景:应用解耦、流量消峰、消息分发。

目前主流的MQ主要是Rocketmq、kafka、Rabbitmq,Rocketmq相比于Rabbitmq、kafka具有主要优势特性有:

支持事务型消息(消息发送和DB操作保持两方的最终一致性,rabbitmq和kafka不支持)

支持结合rocketmq的多个系统之间数据最终一致性(多方事务,二方事务是前提)

支持18个级别的延迟消息(rabbitmq和kafka不支持)

支持指定次数和时间间隔的失败消息重发(kafka不支持,rabbitmq需要手动确认)

支持consumer端tag过滤,减少不必要的网络传输(rabbitmq和kafka不支持)

支持重复消费(rabbitmq不支持,kafka支持)

本文主要介绍RocketMQ的单机安装、双机主从高可用安装配置、运维管理平台搭建、与SpringBoot整合几个知识点,具备相关知识技能的同学请直接拉到最后点个 “在看” 即可。

文章开始之前需要先准备好JDK1.8或以上的服务器环境以及从rocketmq官网-好二进制安装包,-http://rocketmq.apache.org/dowloading/releases/

单机安装配置

工欲善其事必先利其器,要想深入了解RocketMQ得先把环境安装好,咱们先开始单机版RocketMQ的安装!

解压安装

unzip rocketmq-all-4.7.0-bin-release.zip

启动 Name Server

> nohup sh bin/mqnamesrv &

查看 Name Server启动日志

> tail -f ~/logs/rocketmqlogs/namesrv.log

启动 Broker Server

> nohup sh bin/mqbroker -n localhost:9876 &

查看 Broker Server 启动日志

> tail -f ~/logs/rocketmqlogs/broker.log

单机情况下安装使用RocketMQ很简单,只需要分别启动NameServer和Broker Server即可!

关闭RockerMQ需要使用下面的命令:

# 先关闭Broker Server> sh bin/mqshutdown broker

# 再关闭NameServer> sh bin/mqshutdown namesrv

双机主从高可用搭建

为了消除单机故障,增加可靠性或增大吞吐量,可以在多台服务器上部署多个NameServer和Broker,并为每个Broker部署一个或多个Slave。本节将说明使用两台机器,搭建双主、双从、无单点故障的高可用RocketMQ集群。假设现在有两台服务器,IP地址分别为:192.168.100.43和192.168.100.44,部署架构如下:

启动多个NameServer 和 Broker

首先需要在两台服务器上分别启动NameServer(nohup sh bin/mqnamesrv &),这样我们就得到了一个无单点的NameServer服务,服务地址为192.168.100.43:9876和192.168.100.44:9876。

然后在两台服务器中RocketMQ的conf目录分别建立两个文件 broker-master.properties,broker-slave.properties,下面是不同服务器的配置说明:

192.168.100.43 机器上的broker-master.properties文件:

namesrvAddr = 192.168.100.43:9876;192.168.100.44:9876

brokerClusterName = DefaultCluster

brokerName = broker-a

brokerId = 0

deleteWhen = 04

fileReservedTime = 48

brokerRole = SYNC_MASTER

flushDiskType = ASYNC_FLUSH

listenPort = 10911

storePathRootDir = /app/rocketmq/store-a

192.168.100.43 机器上的broker-slave.properties文件:

namesrvAddr = 192.168.100.43:9876;192.168.100.44:9876

brokerClusterName = DefaultCluster

brokerName = broker-b

brokerId = 1

deleteWhen = 04

fileReservedTime = 48

brokerRole = SLAVE

flushDiskType = ASYNC_FLUSH

listenPort = 11011

storePathRootDir = /app/rocketmq/store-b

192.168.100.44 机器上的broker-master.properties文件:

namesrvAddr = 192.168.100.43:9876;192.168.100.44:9876

brokerClusterName = DefaultCluster

brokerName = broker-b

brokerId = 0

deleteWhen = 04

fileReservedTime = 48

brokerRole = SYNC_MASTER

flushDiskType = ASYNC_FLUSH

listenPort = 10911

storePathRootDir = /app/rocketmq/store-b

192.168.100.44 机器上的broker-slave.properties文件:

namesrvAddr = 192.168.100.43:9876;192.168.100.44:9876

brokerClusterName = DefaultCluster

brokerName = broker-a

brokerId = 1

deleteWhen = 04

fileReservedTime = 48

brokerRole = SLAVE

flushDiskType = ASYNC_FLUSH

listenPort = 11011

storePathRootDir = /app/rocketmq/store-a

然后分别使用如下命令启动两台服务器的主节点和从节点

nohup sh bin/mqbroker -c conf/broker-master.properties &

nohup sh bin/mqbroker -c conf/broker-slave.properties &

这样一个高可用的RockerMQ集群就搭建好了,我们登陆可视化运维管理界面查看集群状态,集群正常启动。

重要参数说明

本节主要是对Broker的配置文件中用到的参数进行说明

namesrvAddr = 192.168.100.43:9876;192.168.100.44:9876

指定NameServer的地址,可以是多个。

brokerClusterName = DefaultCluster

Cluster地址,如果集群数量比较多,可以分成多个Cluster,每个Cluster供一个业务群使用。

brokerName = broker-a

Broker的名称,Master 和Slave 通过使用相同的 Broker 名称来表明相互关系,以说明某个Slave 是哪个Master 的 Slave。

brokerId = 1

一个Master可以有多个Slave,0表示Master,大于0的表示不同Slave的ID。

fileReservedTime = 48

在磁盘上保存消息的时长,单位是小时,自动http://删除超时的消息。

deleteWhen = 04

与 fileReservedTime 参数对应,表明在几点做消息删除动作,默认是凌晨4点。

brokerRole = SYNC_MASTER

brokerRole的可选参数有SYNC_MASTER,ASYNC_MASTER,SLAVE三种。SYNC 和ASYNC 表示MASTER 和SLAVE 之间同步消息的机制,SYNC的意思是当Slave 和 Master 的消息同步完成后再返回发送成功的状态。

flushDiskType = ASYNC_FLUSH

flushDiskType 表示刷盘策略,可选值有ASYNC_FLUSH 和 SYNC_FLUSH两种,分别代表同步刷盘和异步刷盘。同步情况下,消息只有真正写入磁盘才返回成功状态;异步情况下,消息写入page_cache后就返回成功状态。

listenPort = 11011

Broker监听的端口,一台服务器启动多个Broker,需要设置不同的监听端口避免端口冲突。

storePathRootDir = /app/rocketmq/store-a

存储消息以及配置信息的根目录。

可视化管理平台

RocketMQ可以使用rocketmq-externals作为运维管理平台,github地址https://github.com/apache/rocketmq-externals,我们需要将源码-下来后再进行手动编译,过程如下:

-

从github(https://github.com/apache/rocketmq-externals) -RocketMQ可视化管理工具 rocketmq-externals 的源码;

打包

-完成后切换进rocketmq-console目录,使用maven命令对其打包 mvn clean package -Dmaven.test.skip=true

打包完成后生成可执行文件rocketmq-console-ng-1.0.1.jar

运行

使用 java -jar rocketmq-console-ng-1.0.1.jar --server.port=8080 --rocketmq.config.namesrvAddr=xxxx.xxx.xxx.xxx:9876 命令启动

这里注意需要设置两个参数:

--server.port 为运行的这个web应用的端口,如果不设置的话默认为8080;

--rocketmq.config.namesrvAddr 为RocketMQ命名服务地址,若NameServer为集群则使用英文 ; 分割

访问

浏览器访问 xxx.xxx.xxx.xxx:8080 进入控制台界面,效果如下

SpringBoot整合RocketMQ

在SpringBoot中整合RocketMQ主要用到 rocketmq-spring-boot-starter 组件,下面是详细整合过程。

引入组件rocketmq-spring-boot-starter 依赖

org.apache.rocketmq

rocketmq-spring-boot-starter

2.1.0

修改application.yml,添加RocketMQ相关配置

rocketmq:

name-server: 192.168.100.43:9876;192.168.100.44:9876

producer:

group: test-group

send-message-timeout: 3000

如果是集群,多个name-server使用英文 ; 分割。

编写消息生产者 MessageProduce

/**

* Description:

* rocketMQ消息发送方法

* @author javadaily

*/

@Component

public class MessageProduce {

@Autowired

private RocketMQTemplate rocketMQTemplate;

/**

* 发送消息

* @param topic 主题

* @param message 消息体

*/

public void sendMessage(String topic,String message){

this.rocketMQTemplate.convertAndSend(topic,message);

}

}

使用RocketMQTemplate发送消息

编写消息消费者 MessageConsumer

@Slf4j

@Component

@RocketMQMessageListener(

topic = "test-topic",

consumerGroup = "test-group",

selectorExpression = "*"

)

public class MessageConsumer implements RocketMQListener {

@Override

public void onMessage(String message) {

log.info("received message is {}", message);

}

}

消费者只需要继承RocketMQListener类即可,主要关注实现类上的 @RocketMQMessageListener 注解,配置的 topic 和 consumerGroup 需要跟消息生产者的配置保持一致。

编写单元测试发送消息

@RunWith(SpringRunner.class)

@SpringBootTest

public class MessageProduceTest {

@Autowired

private MessageProduce messageProduce;

@Test

public void testSendMessage() {

messageProduce.sendMessage("test-topic","Hello,JAVA日知录");

}

}

测试

先启动springboot应用,再执行测试用例。

以上就是微服务架构设计入门RocketMQ基础及环境整合的详细内容,更多关于微服务架构设计RocketMQ环境整合的资料请关注我们其它相关文章!

版权声明:本文内容由网络用户投稿,版权归原作者所有,本站不拥有其著作权,亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容,请联系我们jiasou666@gmail.com 处理,核实后本网站将在24小时内删除侵权内容。

上一篇:Eclipse IDE for Java Developers - 2020-06 安装maven
下一篇:Redis (error) NOAUTH Authentication required.解决方法
相关文章

 发表评论

暂时没有评论,来抢沙发吧~