微服务架构设计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
在磁盘上保存消息的时长,单位是小时,自动删除超时的消息。

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 依赖

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

修改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<String> {
    @Override
    public void onMessage(String message) {
        log.info("received message is {}", message);
    }
}

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

编写单元测试发送消息

@RunWith(SpringRunner.class)
@SpringBootTest
public class MessageProduceTest {
    @Autowired
    private MessageProduce messageProduce;

    @Test
    public void testSendMessage() {
        messageProduce.sendMessage("test-topic","Hello,JAVA日知录");
    }
}

测试

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

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

(0)

相关推荐

  • java RocketMQ快速入门基础知识

    如何使用 1.引入 rocketmq-client <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>4.1.0-incubating</version> </dependency> 2.编写Producer DefaultMQProducer produce

  • Window搭建部署RocketMQ步骤详解

    序 以前简单用过ActiveMQ但是公司项目上使用的是RocketMQ,所以准备多花点时间在这上面,搞懂项目的配置使用. 看了很多资料,先说说我自己对RocketMQ的简单理解.不管是我们写的消费者还是生产者都属于客户端,而我们需要安装RocketMQ,这是属于服务端.和ActivieMQ.zookeeper类似,消费者.生成者.服务端(NameServer)之间是采取观察者模式实现. 在操作系统上安装RocketMQ,启动服务端NameServer.启动Broker,书写Consumer代码,

  • springBoot整合RocketMQ及坑的示例代码

    版本: JDK:1.8 springBoot:1.5.10 rocketMQ:4.2.0 pom 配置: <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>1.5.10.RELEASE</version> </parent> <d

  • idea创建Spring项目的方法步骤(图文)

    Spring介绍 Spring概述 Spring是一个开源框架,Spring是2003年兴起的轻量级java开发框架,由Rod Johnson 在其著作 Expert One-On-One J2EE Development and Design 中阐述的部分理念和原形衍生而来.它是为了解决企业级开发的复杂性而创建的.Spring使用基本的javaBaen来完成以前只可能由EJB完成的事情,然而Spring的用途不仅限于服务器端的开发,从简单性.可测试性.低耦合的角度而言任何java应用都可以在s

  • 浅谈Springboot整合RocketMQ使用心得

    一.阿里云官网---帮助文档 https://help.aliyun.com/document_detail/29536.html?spm=5176.doc29535.6.555.WWTIUh 按照官网步骤,创建Topic.申请发布(生产者).申请订阅(消费者) 二.代码 1.配置: public class MqConfig { /** * 启动测试之前请替换如下 XXX 为您的配置 */ public static final String PUBLIC_TOPIC = "test"

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

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

  • 微服务架构设计RocketMQ进阶事务消息原理详解

    目录 前言 RocketMQ事务流程概要 RocketMQ事务流程关键 实现 基础配置 引入组件 添加配置 发送半消息 执行本地事务与回查 消费消息 测试 总结 前言 分布式消息选型的时候是否支持事务消息是一个很重要的考量点,而目前只有RocketMQ对事务消息支持的最好.今天我们来唠唠如何实现RocketMQ的事务消息! Apache RocketMQ在4.3.0版中已经支持分布式事务消息,这里RocketMQ采用了2PC的思想来实现了提交事务消息,同时增加一个补偿逻辑来处理二阶段超时或者失败

  • 微服务架构拆分策略详解

    目录 1 微服务迁移准备 2 微服务颗粒的拆分策略 2.1 基于业务逻辑拆分 2.1.1 领域模型拆分 2.1.2 用户群体拆分 2.2 基于可扩展拆分 2.3 基于可靠性拆分 2.3.1 核心模块拆分 2.3.2 主次链路拆分 2.4 基于性能需求拆分 3 总结拆分原则 微服务架构及其演进史 微服务全景架构全面瓦解 前面我们学习了微服务的全景架构,了解到相对于传统单体架构,微服务的优势,以及系统服务化的发展趋势. 对于新启动的项目,我们在权衡之后可以大方的使用微服务架构.但其实大部分情况下,我

  • 微服务架构之服务注册与发现功能详解

    目录 微服务的注册与发现 1.服务注册 2.服务发现 3.注册中心 4.现下的主流注册中心 4.1 Eureka 4.1.1 介绍 4.1.2 整体架构 4.1.3 接入Spring Cloud 4.2 ZooKeeper 4.2.1 介绍 4.2.2 整体架构 4.2.3 接入Dubbo生态 4.3 Consul 4.3.1 介绍 4.3.2 整体架构 4.3.3 生态对接 4.4 总结对比 详解微服务架构及其演进史 微服务全景架构全面瓦解 微服务架构拆分策略详解 微服务的注册与发现 我们前面

  • Springcloud微服务架构基础知识解析

    一 前言 学习微服务要从基础的架构学起,首先你要有个微服务的概念才能学习对吧!!如果你都不知道啥是微服务,就一头扎进去学习,你自己也觉得自己也学不会对吧.本篇文章主要让大家快速了解基础的架构分格,以便于微服务入门. 二 单体架构 单体架构是传统架构,其发展了几十年,我们今天任然还在用单体架构开发,存在即合理:单体架构也就是通常的表现层跟UI界面交互,业务层写业务逻辑,数据DAO层访问数据库.其部署方式也很简单,直接将项目打包成war包放进web服务器(如tomcat,jetty)中运行: 其优点

  • 详解Java 微服务架构

    一.传统的整体式架构 传统的整体式架构都是模块化的设计逻辑,如展示(Views).应用程序逻辑(Controller).业务逻辑(Service)和数据访问对象(Dao),程序在编写完成后被打包部署为一个具体的应用.如图所示: 系统的水平扩展 如果要对系统进行水平扩展,通常情况下,只需要增加服务器的数量,并将打包好的应用拷贝到不同的服务器,然后通过负载均衡器(Nginx)就可以轻松实现应用的水平扩展. 整体式架构的缺点 应用复杂度增加,更新.维护困难. 易造成系统资源浪费. 影响开发效率. 应用

  • 解析SpringCloud简介与微服务架构

    1. 微服务架构 1.1 微服务架构理解 微服务架构(Microservice Architecture)是一种架构概念,旨在通过将功能分解到各个离散的服务中以实现对解决方案的解耦.你可以将其看作是在架构层次而非获取服务的类上应用很多SOLID原则.微服务架构是个很有趣的概念,它的主要作用是将功能分解到离散的各个服务当中,从而降低系统的耦合性,并提供更加灵活的服务支持. 概念:把一个大型的单个应用程序和服务拆分为数个甚至数十个的支持微服务,它可扩展单个组件而不是整个的应用程序堆栈,从而满足服务等

  • java开发微服务架构设计消息队列的水有多深

    目录 消息队列的作用 消息队列的设计难题 处理并发和顺序消息 处理重复消息 编写幂等消息处理器 跟踪消息并丢弃重复消息 处理事务性消息 使用数据库表作为消息队列 使用事务日志发布事件 RocketMQ事务消息解决方案 很多人在做架构设计时往往会"过度设计",简单问题复杂化,上来就引一堆中间件,我想大概原因主要有下面两点: 为了秀(学)技术而架构 我们常说技术是为业务服务的,不能为了技术而技术,为了秀技术引入一堆复杂架构这是要不得的. 考虑问题不全面,或者说广度不够,不知道如何简单化 举

  • SpringCloud微服务架构实战之微服务治理功能的实现

    微服务治理 Spring Cloud 工具套件为微服务治理提供了全面的技术支持.这些治理工具主要包括服务的注册与发现.负载均衡管理.动态路由.服务降级和故障转移.链路跟踪.服务监控等.微服务治理的主要功能组件如下: 注册管理服务组件Eureka,提供服务注册和发现的功能. 负载均衡服务组件Ribbon,提供负载均衡调度管理的功能. 边缘代理服务组件Zuul,提供网关服务和动态路由的功能. 断路器组件Hystrix,提供容错机制.服务降级.故障转移等功能. 聚合服务事件流组件Turbine,可用来

  • SpringCloud微服务架构升级汇总

    一.背景 1.1 应用系统的架构历史 1.2 什么是微服务? 起源:微服务的概念源于 2014 年 3 月 Martin Fowler 所写的一篇文章"Microservices".文中内容提到:微服务架构是一种架构模式,它提倡将单一应用程序划分成一组小的服务,服务之间互相协调.互相配合,为用户提供最终价值. 通信方式:每个服务运行在其独立的进程中,服务与服务间采用轻量级的通信机制互相沟通(通常是基于 HTTP 的 RESTful API). 微服务的常规定义:微服务是一种架构风格,一

随机推荐