Fork me on GitHub

RocketMQ搭建

字数统计: 2.2k字   |   阅读时长≈ 8分

消息中间件

中间件这个词一听就是 处于 多个系统中间一个软件，那么它的存在必然有它的好处，也必然要与其他的系统进行关联，如果没有其他其他系统，那么就没有中间件存在的意义。

常见的中间件有 文件存储中间件，缓存中间件，搜索中间件，消息中间件等等，还有其他的

那么消息中间件少不了消息队列

消息队列

消息队列是《数据结构》中先进先出的一种数据结构

消息中间件功能

应用系统解耦

多个应用系统之间不再互相依赖，都跟中间件通讯，只要中间件保证了可用和一致，那么多个应用系统就可以降低可用性

以前A应用需要通过http请求或者RMI 同步调用方式来获取B应用的结果，是一个强依赖关系，因为B挂了，A所有需要B来支撑的服务接口也就相当于什么都干不了，此时就需要对B做高可用，分布式情况下，一个系统像B这样的服务有很多，每一个系统做高可用浪费资源

消息中间件做高可用

那么A应用的服务 想干什么事变成一个消息 放到消息队列

而B应用 订阅消息来进行处理消费

A和B解开了强依赖的关系

此时A应用调用B的服务相当于写了一个消息放入容器，而这个容器是高可用的，那么就相当于异步调用，因为是异步，所以事务这一块无法保证

流量削峰

对于流量很大的系统，往往需要做限流操作，避免系统被大流量击垮，但是限流操作意味着有一部分请求不能立马得到处理，相当于损失了请求，而往往有一些请求很重要不能丢失

这个时候可以使用消息中间件来存入大量的请求，（就可以降低限流的大小，来保证可用），匀速消费，当消息队列中堆积消息过多时，也可以动态上线增加消费服务系统，来保证不丢失重要请求。

大数据处理

消息中间件可以把各个模块中产生的管理员操作日志、用户行为、系统状态等数据文件作为消息收集到主题中

数据使用方可以订阅自己感兴趣的数据内容互不影响，进行消费

异构系统

可以把多个系统进行关联

RocketMQ

RocketMQ是基于（kafka）MetaQ的内核，对其进行了优化让其更满足互联网公司的特点。

纯Java开发，具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。

RocketMQ目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。

角色

首先rocketMQ不遵循JMS的标准

Name Server

底层由netty实现，提供了路由管理、服务注册、服务发现的功能，是一个无状态节点

服务发现者，集群中各个角色（producer、broker、consumer等）都需要定时想nameserver上报自己的状态，以便互相发现彼此，超时不上报的话，nameserver会把它从列表中剔除

部署多个，当多个nameserver存在的时候，其他角色同时向他们上报信息，以保证高可用，

集群间互不通信，没有主备的概念

内存式存储，nameserver中的broker、topic等信息默认不会持久化

Broker

  • Broker面向producer和consumer接受和发送消息

  • 向nameserver提交自己的信息

  • 是消息中间件的消息存储、转发服务器。

  • 每个Broker节点，在启动时，都会遍历NameServer列表，与每个NameServer建立长连接，注册自己的信息，之后定时上报。

  • Broker集群高可用，可以配成Master/Slave结构，Master可写可读，Slave只可以读，Master将写入的数据同步给Slave。Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义BrokerId为0表示Master，非0表示Slave

producer
  • 消息的生产者
  • 通过nameserver集群中的其中一个节点（随机选择）建立长连接，获得Topic的路由信息，包括Topic下面有哪些Queue，这些Queue分布在哪些Broker上等
  • 接下来向提供Topic服务的Master建立长连接，且定时向Master发送心跳
consumer

消息的消费者，通过NameServer集群获得Topic的路由信息，连接到对应的Broker上消费消息。

注意，由于Master和Slave都可以读取消息，因此Consumer与Master和Slave可以建立连接。

消息存储形式

Message是在每个Broker上以Queue的形式记录。

Topic是一个逻辑概念

一个Topic可以包含多个queue

多个queue可以分布在不同的broker

而每一个queue都能记录对应消息

广播形式需要客户端设置消费的属性

官方网站

http://rocketmq.apache.org

GitHub

https://github.com/apache/rocketmq

基本环境

CentOS7 Xshell Xftp 6

Java JDK环境
1
2
3
4
5
6
7
8
9
10
11
12
13
14
yum install wget -y
wget http://download.oracle.com/otn-pub/java/jdk/8u181-b13/96a7b8442fe848ef90c96a2fad6ed6d1/jdk-8u181-linux-x64.rpm
#地址可能会变，小伙伴注意一下


rpm -ivh jdk-8u181-linux-x64.rpm

vi /etc/profile
####
export JAVA_HOME=/usr/java/default
export PATH=$PATH:$JAVA_HOME/bin
####
#检测
java -version
maven环境
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
wget https://mirror.bit.edu.cn/apache/maven/maven-3/3.6.3/binaries/apache-maven-3.6.3-bin.tar.gz
#建议先下好，然后传输到虚拟机


tar -zxvf apache-maven-3.6.3-bin.tar.gz

#换个地方，按自己喜欢来
mv apache-maven-3.6.3 /opt/wyz/

#修改配置文件，换成aliyun的镜像
cd apache-maven-3.6.3/conf/

vi settings.xml
#####
#找到mirrors节点，在里面加入
<mirror>
		<!--This sends everything else to /public -->
		<id>nexus-aliyun</id>
		<mirrorOf>*</mirrorOf>
		<name>Nexus aliyun</name>
		<url>http://maven.aliyun.com/nexus/content/groups/public</url>
	</mirror>
#####

#去修改环境变量
vi /etc/profile
####
#添加环境变量
export PATH=$PATH:/opt/wyz/apache-maven-3.6.3/bin
####
source /etc/profile
#检查
mvn -v

RocketMQ安装

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
#建议先下好
wget https://github.com/apache/rocketmq/archive/master.zip
#下面是MQ扩展包，以后会用到
wget https://github.com/apache/rocketmq-externals/archive/master.zip


yum install unzip -y
unzip rocketmq-master.zip 
cd rocketmq-master
#ls查看目录会发现他是一个聚合项目，maven编译父工程就行
mvn -Prelease-all -DskipTests clean install -U

#此时的target会出现在distribute目录下
cd distribution/
cd target/rocketmq-4.7.1-SNAPSHOT
mv rocketmq-4.7.1-SNAPSHOT/ /opt/wyz/

#此时大体的安装流程走完了
启动
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
cd /opt/wyz/rocketmq-4.7.1-SNAPSHOT/
cd bin/

#先启动mqnamesrv  相当于一个注册中心，先启动
#启动会调起runserver.sh 
./mqnamesrv

#此时没有配置后台运行，会卡住，新开一个虚拟机连接
#启动mqbroker  服务器，做消息业务的
#启动会调起runbroker.sh
cd /opt/wyz/rocketmq-4.7.1-SNAPSHOT/bin/
./mqbroker
#此时会报错，不能分配内存,看看runbroker.sh
vi runbroker.sh
####
#可以看到jvm的一些参数,
#-Xms8g表示分配最小内存为8g
#-Xmx8g表示分配最大内存为8g
#这两个参数相等是为了不要动态扩容，扩容需要耗时
#-Xmn4g 设置年轻代分配4g内存
JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"
#改成
#我虚拟机没有分配这么大内存，所以改小
JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn128m"
####

#此时启动
./mqbroker 
#./mqbroker -n localhost:9876 指定连接哪个mqnamesrv

#会报错，意思是无法获取主机名，但是不影响运行
# BaseNode: BaseNode: Name or service not known
#BaseNode是我当前虚拟机的名字
#但是hosts里面没有配置这个名字的ip地址
ping BaseNode#是ping不通的

vi /etc/hosts
####
#添加
192.168.47.100 BaseNode
####


#此时启动，不会报错
./mqbroker
测试
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
#新开一个虚拟机连接

cd /opt/wyz/rocketmq-4.7.1-SNAPSHOT/bin/

./tools.sh org.apache.rocketmq.example.quickstart.Producer
#此时会抛出远程连接失败的异常

vi tools.sh 
####
#添加一个注册中心的环境变量，位置不要放错，里面有export的位置的
export NAMESRV_ADDR=localhost:9876
####

./tools.sh org.apache.rocketmq.example.quickstart.Producer

#此时还是报错，
#MQClientException: No route info of this topic: TopicTest
#没有客户端的路由信息，把mqnamesrv当成注册中心，其他的服务就相当于客户端，这里的意思就是broker没有把服务注册到注册中心

#重新启动mqbroker
./mqbroker -n localhost:9876 

#这一次执行，正常来说就没有错误了

#这一句话的意思是 把测试的消息放入mq
./tools.sh org.apache.rocketmq.example.quickstart.Producer
#这一句话的意思是 把mq里面的消息进行消费
./tools.sh org.apache.rocketmq.example.quickstart.Consumer

#到这里就测试成功
赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
Java后端开发
开源爱好者、Linux

联系QQ:2960571342

很惭愧

只做了一点微小的工作
谢谢大家