提问5要素

主题 + 数量 + 细节 + 方式 + 格式

有用的提示语

给我一点惊喜

一步一步思考

输出格式

1
2
3
4
5
6
7
8
9
10
11
12
13

用简洁的，通俗易懂的语言告诉新手：
## 💡 TIPS：
1.定义「明确、具体的问题」，并给出 3 组实例和反例，附上讲解。格式：
  **✅ Do：**
  实例。
  **❌ Don't：**
  反例。
  你的讲解。
2.如何验证你输出的准确性和前后一致性？
3.什么是 prompt 工程？
4.使用 ChatGPT 的过程中，prompt 工程的重要性？
5.掌握 ChatGPT 的过程中，和普通人快速拉开差距的秘诀？
谢谢 ❤️

正则表达式

基础

语法汇总：

另外，注意区分正则表达式和通配符的区别。强调*这个字符

在正则中表示重复前面的字符0次或多次

在通配符中表示任意字符

举例来说。(ls命令是不支持正则的，grep，sed支持)

1

ls -l a*

是取出以a开头的所有文件

1

grep 'a*' 某文件

会把文件中所有的数据全部取出，因为正则中*可以表示0次，所以a*的含义就是所有字符，而不是以a开头的意思

正则表达式延申

上面是正则的基本语法，除此之外，正则还有延申的特殊参数

要使用的话用grep -E，不过更常用的是egrep

sed

sed当管道使用

sed是一个管道命令。

sed后接动作要用单引号括住

1

sed '2,5d'

2,5表示对第二行到第五行进行动作。d表示删除动作

可以接的动作如下

• d删除。前面加上行号。如上所示，如果要实现从某一行删除到文件结束sed '2,$d'

• a 新增。后面加上内容。在当前行的下一行新增内容

  1 2 3 4 5

  [api@kfxqtyglpt ~]$ cat test.txt 456 [api@kfxqtyglpt ~]$ cat test.txt | sed '1a okokok' 456 okokok

  sed也是管道操作，所以直接输出到了控制台上

• i在前面新增。接的内容插入在上一行

  1 2 3

  [api@kfxqtyglpt ~]$ cat test.txt | sed '1i helphelp' helphelp 456

• p 显示

  我们要显示一个文件的第10到20行，可以用

  1	head -n 20 | tail -n 10

  现在有了sed，可以直接

  1	sed -n '10,20p'

• s 搜索并替换，这是sed很常用的功能

  1	sed 's/old/new/g'

  old部分可以用正则表达式

• c按行取代

  1 2 3 4 5 6

  [api@kfxqtyglpt ~]$ cat test.txt 456 the last line [api@kfxqtyglpt ~]$ cat test.txt | sed '1c the first line' the first line the last line

sed直接修改文件

比较危险，使用时先用管道场景进行测试

1

sed -i '$a the last line' test.txt

加上-i参数就是直接修改文件，为不是输出到console了

$是最后一行，$a就是在最后以后后新增的意思了

awk

将一行数据分成多个段来处理

格式

1

awk '条件1{动作1} 条件2{动作2}...' 文件名

awk的处理流程是：

1. 将文件第一行读入，并赋给变量$0,$1,$2…，awk默认的分割符是空格或tab
2. 根据条件判断，执行动作
3. 执行完所有动作
4. 继续下一行

变量$0是一整行的数据，$1是第一分段数据

awk的内建变量：

NF：每一行（$0）拥有的分段总数

NR：当前awk是在处理第几行

FS：当前的分隔符

自定义分隔符

通过FS=”:”可以设置分隔符为：

如

1
2
3
4
5
6

$ last -n 5 |egrep -v '^$|wtmp' | awk '{FS=":"} {print $1 "\t all:" NF "\t now: " NR}'
api      all:10  now: 1
api      pts/12       1x.xx.0.66      Fri Feb  3 11     all:4   now: 2
root     pts/4        1x.xxx.7.21      Fri Feb  3 11     all:4   now: 3
api      pts/1        1x.xx.179.212   Fri Feb  3 11     all:4   now: 4
api      pts/4        1x.xx.179.212   Fri Feb  3 10     all:4   now: 5

但是这样有个问题，就是第一行在执行的时候，分隔符还是空格，所以第一行的结果有问题

解决办法是使用关键字BEGIN

1
2
3
4
5
6

[api@kfxqtyglpt ~]$ last -n 5 |egrep -v '^$|wtmp' | awk 'BEGIN {FS=":"} {print $1 "\t all:" NF "\t now: " NR}'
api      pts/1        1x.xx.179.212   Fri Feb  3 14     all:2   now: 1
api      pts/12       1x.xx.0.66      Fri Feb  3 11     all:4   now: 2
root     pts/4        1x.xxx.7.21      Fri Feb  3 11     all:4   now: 3
api      pts/1        1x.xx.179.212   Fri Feb  3 11     all:4   now: 4
api      pts/4        1x.xx.179.212   Fri Feb  3 10     all:4   now: 5

条件

awk中可以使用条件运算符：>,<,>=,<=,==,!=

1
2
3
4
5
6

[api@kfxqtyglpt ~]$ last -n 5 |egrep -v '^$|wtmp' | awk 'BEGIN {FS=":"} {printf $1 "\t all:" NF "\t now: " NR} NR>3 {printf "\t more than 3 \n"} NR<=3 {printf "\n"}'
api      pts/4        1x.xx.181.65    Fri Feb  3 14     all:2   now: 1
api      pts/1        1x.xx.179.212   Fri Feb  3 14     all:2   now: 2
api      pts/12       1x.xx.0.66      Fri Feb  3 11     all:4   now: 3
root     pts/4        1x.xxx.7.21      Fri Feb  3 11     all:4   now: 4  more than 3
api      pts/1        1x.xx.179.212   Fri Feb  3 11     all:4   now: 5  more than 3

对大于3的行添加了一个输出。注意我这里使用了printf，要自己加上\n换行

第一个sh脚本

1

vim hello.sh

1
2
3
4

#!/bin/bash
# 注释
echo -e "Hello World! \a \n"
exit 0

1. 第一行 #!/bin/bash，作用是宣告这个文件内的语法使用bash语法，在这个脚本执行时，就能够加载bash相关的环境配置文件（~/.bashrc）
2. # 用来注解
3. echo这里就是我们写的程序体部分
4. exit 回传结果。一般 0 表示执行成功。其他的数字是执行失败，可以用不同的数字表示不同的失败场景

执行

1

sh hello.sh

注意，现在用

1

./hello.sh

会执行失败。

因为我们创建的文件并没有x可执行权限。而sh hello.sh中我们的sh是作为sh的参数传入，因此只要hello.sh有r读权限即可

1
2
3

[api@kfxqtyglpt bin]$ ll
总用量 4
-rw-rw-r--. 1 api api 57 2月   7 16:48 hello.sh

可以加上x权限，再执行

1

chmod a+x hello.sh

必会功能

变量

定义变量

使用变量

（见上一篇文章）

执行命令

sh脚本中也是使用$()来执行linux的命令

1

$(seq 1 100)

执行seq命令，生产1到100的数组，可用在for循环中

交互式输入

很多场景下需要用户输入一些信息，此时可使用read命令

1
2
3
4
5
6
7

#!/bin/bash
# 注释
read -p "输入用户名：" name
read -p "输入密码：" passwd
echo -e "\n ${name} ${passwd}"
exit 0

日期相关

根据当天的日期创建文件，使用到date命令

1
2
3
4
5

#!/bin/bash
today=$(date +%Y%m%d)
filename="log_"${today}
touch "${filename}"
exit 0

$()用于执行命令

测试一个文件是否存在

我们需要根据一个文件是否存在，从而执行不同的动作

用到test命令

1
2
3
4
5

#!/bin/bash
# 注释
read -p "输入文件名：" filename
test -e ${filename} && echo "文件存在" || echo "文件不存在"
exit 0

test命令有很多参数，只记录一些常用的，

判断符[]

上面的test命令也可以用判断符[]来代替。bash中的判断符就是中括号

1

[ "${param}" == "xxx" ]

格式如上，表示判断param变量和xxx字符串是否相等

需要特别注意的是[]里必须跟着空格，==这类判断符号两边也必须跟着空格

变量${}要用双引号包裹

常量最好用单引号或双引号包裹

[] 判断符常用在if…then…if的条件语法中

脚本接收参数

这基本是必用的功能，我们执行脚本时经常在脚本后直接跟上参数，那我们自己编写的脚本怎么接收参数？

默认参数变量

在shell脚本中已经默认使用如$0,$1作为接收的参数变量，具体如下

$0,$1等也可以加上大括号${0},${1}

1
2
3
4
5
6
7
8

[api@kfxqtyglpt bin]$ cat param.sh
#!/bin/bash
echo "script name is $0"
echo "first param is $1"
echo "second param is $2"
echo "total param num is $#"
echo "all params are $@"
exit 0

1
2
3
4
5
6

[api@kfxqtyglpt bin]$ ./param.sh one two
script name is ./param.sh
first param is one
second param is two
total param num is 2
all params are one two

条件判断if

sh脚本中条件判断的写法

1
2
3

if [ 判断表达式 ]; then
	为true进行动作
fi

这是虽简单的单层判断

[ 判断表达式 ] 可以使用&&或||多个相连

1
2
3
4
5

if [ 判断表达式 ]; then
	为true进行动作
else
	为false进行工作
fi

也是使用else关键字

再复杂一点

1
2
3
4
5
6
7

if [ 判断表达式1 ]; then
	判断表达式1为true进行动作
elif [ 判断表达式2 ]; then 
	判断表达式2为true进行动作
else
	都不成立
fi

条件判断case

sh脚本中也有case语法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

case $变量 in 
  "匹配内容1")
  		程序
  		;;
  "匹配内容2")
  		程序
  		;;
  "匹配内容3")
  		程序
  		;;
  *)
    	最后都不匹配的程序
    	exit 1
  		;;
esac

注意，

1. 匹配内容后的小括号）
2. 每个匹配后的程序结束后，要用两个分号结束;;
3. 最后要用*）匹配所有，做一个兜底

循环

sh脚本中的循环也有两种形式，while和for

while do形式

语法格式

1
2
3
4

while [ 条件 ]
do
	代码段
done

sh脚本里还有一个until…do…done的语法，

while的含义是当条件成立就do

until的含义是do直到条件成立，也就是说当条件成立了就不做

for循环

for in的语法

1
2
3
4

for var in 多个变量，比如数组
do
	代码段
done

举个例子

1
2
3
4
5

#!/bin/bash
for i in $(seq 1 10)
do
        echo "the num is ${i}"
done

效果

1
2
3
4
5
6
7
8
9
10
11
12

[api@kfxqtyglpt bin]$ ./for.sh
the num is 1
the num is 2
the num is 3
the num is 4
the num is 5
the num is 6
the num is 7
the num is 8
the num is 9
the num is 10

上面的$(seq 1 10)也可以用{1..10}来代替。这是bash的内置机制。

两个..表示连续出现的意思

{a..d}就表示a,b,c,d

sh脚本中只要是列表的都可进行for循环

1

for finename in $(ls 目录)

和其他编程语言一样，除了for in，还有

for 数字

1
2
3
4

for ((初始值;限制条件;步进))
do
	代码段
done

举个例子

1
2
3
4
5
6
7
8

#!/bin/bash
sum=0
for (( i=1;i<=100;i=i+1))
do
        sum=$((${sum}+${i}))
done
echo "sum is ${sum}"

$(())双括号是数值计算

定义函数

定义函数的语法

1
2
3

function 函数名() {
	代码段
}

很标准的语法。

需要注意的是：

函数的定义必须放在函数的使用之前，否则找不到

自定义函数的使用

1
2
3
4
5
6

#!/bin/bash
function myprint() {
  echo "your input is:"
}

myprint; echo ${1}

使用函数时直接用函数名，不需要加括号

1
2
3

[api@kfxqtyglpt bin]$ ./func.sh one
your input is:
one

myprint后面的；，只是为了分开两个命令，后面的echo是另一个命令

完全也可以将echo语句换一行,效果是一样的

1
2
3
4
5
6
7

#!/bin/bash
function myprint() {
  echo "your input is:"
}

myprint
echo ${1}

函数接收参数

函数也可以接收参数，参数也是$0，$1，$2等，$0代表函数名，$1是第一个参数。

注意这个参数和整个sh脚本接收的参数不是一回事。

这个是函数的参数。

比如将上面的例子改一下

1
2
3
4
5
6

#!/bin/bash
function myprint() {
  echo "your input is: ${1}"
}

myprint "one"

sh脚本debug小技巧

编写sh脚本经常会遇到语法问题，有一个参数可以帮我们进行语法校验

1

$sh -n xxx.sh

sh使用-n参数可以进行sh脚本的语法校验

还有其他参数

1

sh -x xxx.sh

-x参数可以将执行过程全部打印出来

Redis分享

[TOC]

数据结构

一共有10种，常用的有5种

• String字符串

• Hash 字典

• List 列表

• Set集合

• ZSet 有序集合

• Pubsub 发布订阅 （不推荐使用，坑很多）

• Bitmap 位图

• GEO 地理位置 （有限使用，附近的人）

• Stream 流(5.0) （与Kafka非常像）

• Hyperloglog 基数统计，HyperLogLog 提供不精确的去重计数方案

集群模式

redis提供了非常丰富的集群模式：主从、哨兵、cluster，满足服务高可用的需求。

其中cluster模式为目前大多数公司所采用的方式。redis cluster是redis亲生的集群方案，它的主要特点就是去中心化，无需proxy代理。其中一个主要设计目标就是达到线性可扩展性。

优点

1. 不再需要额外的Sentinel集群，为使用者提供了一致的方案，减少了学习成本。

2. 去中心架构，节点对等，集群可支持上千个节点。

3. 副本功能能够实现自动故障转移，大部分情况下无需人工介入。

4. 可水平线性扩展

缺点

1. 从库是完全的冷备，无法分担读操作
2. 数据是通过异步复制的，不能保证数据的强一致性。(其实不单是cluster模式的问题)
3. 资源隔离困难，经常流量不均衡，尤其是多个业务共用集群的时候。可能存在热点数据问题。
4. 数据迁移是基于key而不是基于slot的，过程较慢。
5. 一些批量操作。由于key被hash到多台机器上，所以mget、hmset、sunion等操作就非常的不友好，经常发生性能问题。

基本原理

redis cluster采用虚拟槽的概念，把所有的key映射到 0～16383一共16384个整数槽内，当需要在其中存取一个key时，redis客户端会首先对这个key采用crc16算法算出一个值，然后对这个值进行mod操作。

1

crc16(key)mod 16384

redis cluster所有的集群操作都是围绕slot进行。所以必须存储slot的存储关系。

redis节点发送心跳包时，需要把所有的槽信息放在这个心跳包里，所以必须优化使数据量最小。

使用使用bitmap来存储是最节省空间的。这个数组的长度为 16384/8=2048 Byte，2K字节。

请求随机落到一个主节点，该节点检查key是否在自己负责的slot上，是就处理。不是就根据信息将请求转发到对应的节点上。所以，客户端连接集群中的任意一台机器，都能够完成操作。

当数据库中的16384个槽都有节点在处理时，集群处于上线状态（ok）；相反地，如果数据库中有任何一个槽没有得到处理，那么集群处于下线状态（fail）。（可以配置，使集群强制可读）

主从切换

redis-cluster可以自动完成一定程度的故障转移。

集群中的每个节点都会定期地向集群中的其他节点发送ping消息，以此来检测对方是否在线，如果接收ping消息的节点没有在规定的时间内返回pong消息，那么发送ping消息的节点就会将接收ping消息的节点标记为疑似下线（PFAIL）。

如果在一个集群里面，半数以上节点都将某个主节点x报告为疑似下线，那么这个主节点x将被标记为已下线（FAIL），将x标记为FAIL的节点会向集群广播一条关于x的FAIL消息，所有收到这条FAIL消息的节点都会立即将x标记为FAIL。

大家可以注意到这个过程，与es和zk的节点判断类似，都是半数以上才进行判断，所以主节点的数量一般都是奇数。

当一个节点发现自己的主节点进入fail状态，将会从这个主节点的从节点当中，选出一台，执行slaveof no one命令，变身为主节点。

新的节点完成自己的槽指派以后，会向集群广播一条pong消息，以便让其他节点立即知道自己的这些变化。它告诉别人：我已经是主节点了，我已经接管了有问题的节点，成为了新的主节点。

主从数据同步

当一台从机连接到master之后，会发送一个sync指令。master在收到这个指令后，会在后台启动存盘进程。执行完毕后，master将整个数据库文件传输到slave，这样就完成了第一次全量同步。

接下来，master会把自己收到的变更指令，依次传送给slave，从而达到数据的最终同步。从redis 2.8开始，就支持主从复制的断点续传，如果主从复制过程中，网络连接断掉了，那么可以接着上次复制的地方，继续复制下去，而不是从头开始复制一份。

数据丢失问题

redis cluster中节点之间使用异步复制，并没有类似kafka这种ack的概念。节点之间通过gossip协议交换状态信息，用投票机制完成Slave到Master的角色提升，完成这个过程注定了需要时间。在发生故障的过程中就容易存在窗口，导致丢失写入的数据。比如以下两种情况。

一、命令已经到到master，此时数据并没有同步到slave，master会对客户端回复ok。如果这个时候主节点宕机，那么这条数据将会丢失。redis这样做会避免很多问题，但对一个对数据可靠性要求较高的系统，是不可忍受的。

二、由于路由表是在客户端存放的，存在一个时效问题。如果分区导致一个节点不可达，提升了某个从节点，但原来的主节点在这个时候又可以用了（并未完成failover）。这个时候一旦客户端的路由表并没有更新，那么它将会把数据写到错误的节点，造成数据丢失。

所以redis cluster在通常情况下运行的很好，在极端情况下某些值丢失问题，目前无解。

针对cluster模式的坑

1、redis cluster号称能够支持1k个节点，但你最好不要这么做。当节点数量增加到10，就能够感受到集群的一些抖动。

2、一定要避免产生热点，如果流量全部打到了某个节点，后果一般很严重。

3、大key不要放redis，它会产生大量的慢查询，影响正常的查询。

4、如果你不是作为存储，缓存一定要设置过期时间。占着茅坑不拉屎的感觉是非常讨厌的。

5、大流量，不要开aof，开rdb即可。

6、redis cluster的操作，少用pipeline，少用multi-key，它们会产生大量不可预料的结果。

其他模式

主从模式

redis最早支持的，就是M-S模式，也就是一主多从。redis单机qps可达到10w+，但是在某些高访问量场景下，依然不太够用。一般通过读写分离来增加slave，减少主机的压力。

数据同步问题

先进行全量同步，再进行增量同步。主从同步是异步进行的。

要注意的一个点是：

run_id是master唯一标示，slave连接master时会传runid，master每次重启runid都发生变化，当slave发现master的runid变化时都会触发全量复制流程。

哨兵模式

哨兵模式是主从的升级版，因为主从的出现故障后，不会自动恢复，需要人为干预。所以在主从的基础上，实现哨兵模式就是为了监控主从的运行状况，对主从的健壮进行监控。并且当master出现故障的时候，会自动选举一个slave作为master顶上去。

故障恢复

当master被认为客观下线后，首先需要在哨兵中选出一个老大哨兵进行故障恢复。选举老大哨兵的算法还是Raft算法

选出大佬哨兵后，大佬哨兵就会对故障进行自动恢复，即从slave中选出一名slave作为master

1. 所有的slave中 slave-priority优先级最高的会被选中。
2. 若是优先级相同，会选择偏移量最大的，因为偏移量记录着数据的复制的增量，越大表示数据越完整。
3. 若是以上两者都相同，选择RunID最小的。

代理模式

代理模式在redis cluster出现之前，非常流行，比如codis，predixy。

持久化机制

redis提供了两种持久化方式：aof和rdb，常用的是rdb。RDB 和 AOF 两种方式也可以同时使用，也可以都不用，也可以只用一个。

RDB

Redis DataBase。简而言之，就是在不同的时间点，将 redis 存储的数据生成快照并存储到磁盘上；

AOF

则是换了一个角度来实现持久化，那就是将 redis 执行过的所有写指令记录下来，在下次 redis 重新启动时，只要把这些写指令从前到后再重复执行一遍，就可以实现数据恢复了

对于 RDB 方式，redis 会单独创建（fork）一个子进程来进行持久化，而主进程是不会进行任何 IO 操作的，这样就确保了 redis 极高的性能。

RDB 有不少优点，但它的缺点也是不容忽视的。

即使你每 5 分钟都持久化一次，当 redis 故障时，仍然会有近 5 分钟的数据丢失

默认的 AOF 持久化策略是每秒钟 fsync 一次（fsync 是指把缓存中的写指令记录到磁盘中），

因为在这种情况下，redis 仍然可以保持很好的处理性能，即使 redis 故障，也只会丢失最近 1 秒钟的数据。

也可以配置每次执行命令都fsync一次。

因为采用了追加方式，如果不做任何处理的话，AOF 文件会变得越来越大，为此，redis 提供了 AOF 文件重写（rewrite）机制，即当 AOF 文件的大小超过所设定的阈值时，redis 就会启动 AOF 文件的内容压缩

rdb的优点是文件小，恢复快。但是存在数据丢失较多的风险。

aof的有点是数据丢失风险较小，缺点是文件较大。

Redis使用场景

• 缓存 （缓存一致性 缓存穿透 缓存击穿 缓存雪崩）
• 分布式锁 （redlock）
• 分布式限流
• Session共享

缓存一致性

为什么有一致性问题?

• 写入。缓存和数据库是两个不同的组件，只要涉及到双写，就存在只有一个写成功的可能性，造成数据不一致。
• 更新。更新的情况类似，需要更新两个不同的组件。
• 读取。读取要保证从缓存中读到的信息是最新的，是和数据库中的是一致的。
• 删除。当删除数据库记录的时候，如何把缓存中的数据也删掉？

建议使用：Cache Aside Pattern

读请求：

• 先读cache，再读db

变更操作：

• 不要更新缓存，直接淘汰缓存

• 先操作数据库，再 淘汰 缓存

涉及到复杂的事务和回滚操作，可以把淘汰放在finally里。

缓存击穿

影响，轻微。

高流量下 大量请求读取一个失效的Key -> Redis Miss -> 穿透到DB

解决方式：采用分布式锁，只有拿到锁的第一个线程去请求数据库，然后插入缓存

缓存穿透

影响，一般。

故意访问一个不存在的Key（恶意攻击）-> Redis Miss -> 穿透到DB

解决方式：

1. 给相应的Key设置一个Null值，放在缓存中
2. BloomFilter预先判断。当布隆过滤器说某个值存在时，这个值可能不存在；当它说不存在时，那就肯定不存在。

缓存雪崩

影响：严重。

大量Key同时失效 ｜ Redis当机 -> Redis Miss -> 压力打到DB

解决方式：

1. 给失效时间加上相对的随机数
2. 保证Redis的高可用

运维注意

问题排查

• monitor指令 回显所有执行的指令。可以使用grep配合过滤
• keyspace-events 订阅某些Key的事件。比如，删除某条数据的事件，底层实现基于pubsub
• slow log 顾名思义，慢查询，非常有用
• --bigkeys启动参数 Redis大Key健康检查。使用的是scan的方式执行， 不用担心阻塞
• memory usage key、memory stats 指令
• info指令，关注instantaneous_ops_per_sec、used_memory_human、connected_clients
• redis-rdb-tools rdb线下分析

常见的运维故障

• 使用 keys * 把库堵死，——使用别名把这个命令改名

  rename

• 超过最大内存（maxmemory）后，部分数据被删除——这个有删除策略的，选择适合自己的即可

  1. volatile-lru 从设置过期数据集里查找最近最少使用

  2. volatile-ttl 从设置过期的数据集里面优先删除剩余时间短的Key

  3. volatile-random 从设置过期的数据集里面任意选择数据淘汰

  4. volatile-lfu 从过期的数据集里删除 最近不常使用 的数据淘汰

  5. allkeys-lru

  6. allkeys-lfu

  7. allkeys-random

  8. no-enviction 禁止删除，默认策略

• 没开持久化，却重启了实例，数据全掉

  记得非缓存的信息需要打开持久化
  RDB 的持久化需要 vm.overcommit_memory=1，否则有可能会持久化失败
  没有持久化情况下，主从，主重启太快，从还没认为主挂的情况下，从会清空自己的数据——人为重启主节点前，先关闭从节点的同步

• 如果不设置maxmemory,Redis将一直使用内存，直到触发操作系统的OOM-KILLER。

规范

冷热数据区分

虽然 Redis支持持久化，但将所有数据存储在 Redis 中，成本非常昂贵。建议将热数据 (如 QPS超过 5k) 的数据加载到 Redis 中。低频数据可存储在 Mysql、 ElasticSearch中。

业务数据分离

不要将不相关的数据业务都放到一个 Redis中。一方面避免业务相互影响，另一方面避免单实例膨胀，并能在故障时降低影响面，快速恢复。

消息大小限制

由于 Redis 是单线程服务，消息过大会阻塞并拖慢其他操作。保持消息内容在 1KB 以下是个好的习惯。严禁超过 50KB 的单条记录。消息过大还会引起网络带宽的高占用，持久化到磁盘时的 IO 问题。

连接数限制

连接的频繁创建和销毁，会浪费大量的系统资源，极限情况会造成宿主机当机。请确保使用了正确的 Redis 客户端连接池配置。

缓存 Key 设置失效时间

作为缓存使用的 Key，必须要设置失效时间。失效时间并不是越长越好，请根据业务性质进行设置。注意，失效时间的单位有的是秒，有的是毫秒，这个很多同学不注意容易搞错。

缓存不能有中间态

缓存应该仅作缓存用，去掉后业务逻辑不应发生改变，万不可切入到业务里。第一，缓存的高可用会影响业务；第二，产生深耦合会发生无法预料的效果；第三，会对维护行产生负效果。

扩展方式首选客户端 hash

小应用就算了

如单 redis 集群并不能为你的数据服务，不要着急扩大你的 redis 集群（包括 M/S 和 Cluster)，集群越大，在状态同步和持久化方面的性能越差。 优先使用客户端 hash 进行集群拆分。如：根据用户 id 分 10 个集群，用户尾号为 0 的落在第一个集群。

Key 规范

Redis 的 Key 一定要规范，这样在遇到问题时，能够进行方便的定位。Redis 属于无 scheme 的 KV 数据库，所以，我们靠约定来建立其 scheme 语义。其好处：

1、能够根据某类 key 进行数据清理

2、能够根据某类 key 进行数据更新

3、能够方面了解到某类 key 的归属方和应用场景

4、为统一化、平台化做准备，减少技术变更

一般，一个 key 需要带以下维度：业务、key 用途、变量等，各个维度使用 : 进行分隔

操作限制

严禁使用 Keys

Keys 命令效率极低，属于 O(N)操作，会阻塞其他正常命令，在 cluster 上，会是灾难性的操作。严禁使用，DBA 应该 rename 此命令，从根源禁用。

严禁使用 Flush

flush 命令会清空所有数据，属于高危操作。严禁使用，DBA 应该 rename 此命令，从根源禁用，仅 DBA 可操作。

严禁作为消息队列使用

如没有非常特殊的需求，严禁将 Redis 当作消息队列使用。Redis 当作消息队列使用，会有容量、网络、效率、功能方面的多种问题。如需要消息队列，可使用高吞吐的 Kafka 或者高可靠的 RocketMQ。

严禁不设置范围的批量操作

redis 那么快，慢查询除了网络延迟，就属于这些批量操作函数。大多数线上问题都是由于这些函数引起。

1、[zset] 严禁对 zset 的不设范围操作

ZRANGE、 ZRANGEBYSCORE等多个操作 ZSET 的函数，严禁使用 ZRANGE myzset 0 -1 等这种不设置范围的操作。请指定范围，如 ZRANGE myzset 0 100。如不确定长度，可使用 ZCARD 判断长度

2、[hash] 严禁对大数据量 Key 使用 HGETALL

HGETALL会取出相关 HASH 的所有数据，如果数据条数过大，同样会引起阻塞，请确保业务可控。如不确定长度，可使用 HLEN 先判断长度

3、[key] Redis Cluster 集群的 mget 操作

Redis Cluster 的 MGET 操作，会到各分片取数据聚合，相比传统的 M/S架构，性能会下降很多，请提前压测和评估

4、[其他] 严禁使用 sunion, sinter, sdiff等一些聚合操作

禁用 select 函数

select函数用来切换 database，对于使用方来说，这是很容易发生问题的地方，cluster 模式也不支持多个 database，且没有任何收益，禁用。

禁用事务

redis 本身已经很快了，如无大的必要，建议捕获异常进行回滚，不要使用事务函数，很少有人这么干。

禁用 lua 脚本扩展

lua 脚本虽然能做很多看起来很 cool 的事情，但它就像是 SQL 的存储过程，会引入性能和一些难以维护的问题，禁用。

禁止长时间 monitor

monitor函数可以快速看到当前 redis 正在执行的数据流，但是当心，高峰期长时间阻塞在 monitor 命令上，会严重影响 redis 的性能。此命令不禁止使用，但使用一定要特别特别注意。

Redis为什么这么快

基于内存

单线程，没有cpu的上下文切换，无锁

io多路复用

1. MQ介绍

##1.1 为什么要用MQ

消息队列是一种“先进先出”的数据结构

其应用场景主要包含以下3个方面

• 应用解耦

系统的耦合性越高，容错性就越低。以电商应用为例，用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障或者因为升级等原因暂时不可用，都会造成下单操作异常，影响用户使用体验。

使用消息队列解耦合，系统的耦合性就会提高了。比如物流系统发生故障，需要几分钟才能来修复，在这段时间内，物流系统要处理的数据被缓存到消息队列中，用户的下单操作正常完成。当物流系统回复后，补充处理存在消息队列中的订单消息即可，终端系统感知不到物流系统发生过几分钟故障。

• 流量削峰

应用系统如果遇到系统请求流量的瞬间猛增，有可能会将系统压垮。有了消息队列可以将大量请求缓存起来，分散到很长一段时间处理，这样可以大大提到系统的稳定性和用户体验。

一般情况，为了保证系统的稳定性，如果系统负载超过阈值，就会阻止用户请求，这会影响用户体验，而如果使用消息队列将请求缓存起来，等待系统处理完毕后通知用户下单完毕，这样总不能下单体验要好。

处于经济考量目的：

业务系统正常时段的QPS如果是1000，流量最高峰是10000，为了应对流量高峰配置高性能的服务器显然不划算，这时可以使用消息队列对峰值流量削峰

• 数据分发

通过消息队列可以让数据在多个系统更加之间进行流通。数据的产生方不需要关心谁来使用数据，只需要将数据发送到消息队列，数据使用方直接在消息队列中直接获取数据即可

1.2 MQ的优点和缺点

优点：解耦、削峰、数据分发

缺点包含以下几点：

• 系统可用性降低

  系统引入的外部依赖越多，系统稳定性越差。一旦MQ宕机，就会对业务造成影响。

  如何保证MQ的高可用？

• 系统复杂度提高

  MQ的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过MQ进行异步调用。

  如何保证消息没有被重复消费？怎么处理消息丢失情况？那么保证消息传递的顺序性？

• 一致性问题

  A系统处理完业务，通过MQ给B、C、D三个系统发消息数据，如果B系统、C系统处理成功，D系统处理失败。

  如何保证消息数据处理的一致性？

1.3 各种MQ产品的比较

常见的MQ产品包括Kafka、ActiveMQ、RabbitMQ、RocketMQ。

2. RocketMQ快速入门

RocketMQ是阿里巴巴2016年MQ中间件，使用Java语言开发，在阿里内部，RocketMQ承接了例如“双11”等高并发场景的消息流转，能够处理万亿级别的消息。

2.1 准备工作

2.1.1 下载RocketMQ

RocketMQ最新版本：4.5.1

下载地址

2.2.2 环境要求

• Linux64位系统

• JDK1.8(64位)

• 源码安装需要安装Maven 3.2.x

2.2 安装RocketMQ

2.2.1 安装步骤

本教程以二进制包方式安装

1. 解压安装包
2. 进入安装目录

2.2.2 目录介绍

• bin：启动脚本，包括shell脚本和CMD脚本
• conf：实例配置文件 ，包括broker配置文件、logback配置文件等
• lib：依赖jar包，包括Netty、commons-lang、FastJSON等

2.3 启动RocketMQ

1. 启动NameServer

1
2
3
4

# 1.启动NameServer
nohup sh bin/mqnamesrv &
# 2.查看启动日志
tail -f ~/logs/rocketmqlogs/namesrv.log

2. 启动Broker

1
2
3
4

# 1.启动Broker
nohup sh bin/mqbroker -n localhost:9876 &
# 2.查看启动日志
tail -f ~/logs/rocketmqlogs/broker.log 

• 问题描述：

  RocketMQ默认的虚拟机内存较大，启动Broker如果因为内存不足失败，需要编辑如下两个配置文件，修改JVM内存大小

1
2
3

# 编辑runbroker.sh和runserver.sh修改默认JVM大小
vi runbroker.sh
vi runserver.sh

• 参考设置：

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

2.4 测试RocketMQ

2.4.1 发送消息

1
2
3
4

# 1.设置环境变量
export NAMESRV_ADDR=localhost:9876
# 2.使用安装包的Demo发送消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer

2.4.2 接收消息

1
2
3
4

# 1.设置环境变量
export NAMESRV_ADDR=localhost:9876
# 2.接收消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

2.5 关闭RocketMQ

1
2
3
4

# 1.关闭NameServer
sh bin/mqshutdown namesrv
# 2.关闭Broker
sh bin/mqshutdown broker

3. RocketMQ集群搭建

3.1 各角色介绍

• Producer：消息的发送者；举例：发信者
• Consumer：消息接收者；举例：收信者
• Broker：暂存和传输消息；举例：邮局
• NameServer：管理Broker；举例：各个邮局的管理机构
• Topic：区分消息的种类；一个发送者可以发送消息给一个或者多个Topic；一个消息的接收者可以订阅一个或者多个Topic消息
• Message Queue：相当于是Topic的分区；用于并行发送和接收消息

3.2 集群搭建方式

3.2.1 集群特点

• NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。

• Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。

• Producer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。

• Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。

3.2.3 集群模式

1）单Master模式

这种方式风险较大，一旦Broker重启或者宕机时，会导致整个服务不可用。不建议线上环境使用,可以用于本地测试。

2）多Master模式

一个集群无Slave，全是Master，例如2个Master或者3个Master，这种模式的优缺点如下：

• 优点：配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10时，即使机器宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢（异步刷盘丢失少量消息，同步刷盘一条不丢），性能最高；
• 缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅，消息实时性会受到影响。

3）多Master多Slave模式（异步）

每个Master配置一个Slave，有多对Master-Slave，HA采用异步复制方式，主备有短暂消息延迟（毫秒级），这种模式的优缺点如下：

• 优点：即使磁盘损坏，消息丢失的非常少，且消息实时性不会受影响，同时Master宕机后，消费者仍然可以从Slave消费，而且此过程对应用透明，不需要人工干预，性能同多Master模式几乎一样；
• 缺点：Master宕机，磁盘损坏情况下会丢失少量消息。

4）多Master多Slave模式（同步）

每个Master配置一个Slave，有多对Master-Slave，HA采用同步双写方式，即只有主备都写成功，才向应用返回成功，这种模式的优缺点如下：

• 优点：数据与服务都无单点故障，Master宕机情况下，消息无延迟，服务可用性与数据可用性都非常高；
• 缺点：性能比异步复制模式略低（大约低10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机后，备机不能自动切换为主机。

3.3 双主双从集群搭建

3.3.1 总体架构

消息高可用采用2m-2s（同步双写）方式

3.3.2 集群工作流程

1. 启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。
2. Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。
3. 收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。
4. Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。
5. Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息。

3.3.3 服务器环境

3.3.4 Host添加信息

1

vim /etc/hosts

配置如下:

1
2
3
4
5
6
7
8

# nameserver
192.168.25.135 rocketmq-nameserver1
192.168.25.138 rocketmq-nameserver2
# broker
192.168.25.135 rocketmq-master1
192.168.25.135 rocketmq-slave2
192.168.25.138 rocketmq-master2
192.168.25.138 rocketmq-slave1

配置完成后, 重启网卡

1

systemctl restart network

3.3.5 防火墙配置

宿主机需要远程访问虚拟机的rocketmq服务和web服务，需要开放相关的端口号，简单粗暴的方式是直接关闭防火墙

1
2
3
4
5
6

# 关闭防火墙
systemctl stop firewalld.service 
# 查看防火墙的状态
firewall-cmd --state 
# 禁止firewall开机启动
systemctl disable firewalld.service

或者为了安全，只开放特定的端口号，RocketMQ默认使用3个端口：9876 、10911 、11011 。如果防火墙没有关闭的话，那么防火墙就必须开放这些端口：

• nameserver 默认使用 9876 端口
• master 默认使用 10911 端口
• slave 默认使用11011 端口

执行以下命令：

1
2
3
4
5
6
7
8

# 开放name server默认端口
firewall-cmd --remove-port=9876/tcp --permanent
# 开放master默认端口
firewall-cmd --remove-port=10911/tcp --permanent
# 开放slave默认端口 (当前集群模式可不开启)
firewall-cmd --remove-port=11011/tcp --permanent 
# 重启防火墙
firewall-cmd --reload

3.3.6 环境变量配置

1

vim /etc/profile

在profile文件的末尾加入如下命令

1
2
3
4

#set rocketmq
ROCKETMQ_HOME=/usr/local/rocketmq/rocketmq-all-4.4.0-bin-release
PATH=$PATH:$ROCKETMQ_HOME/bin
export ROCKETMQ_HOME PATH

输入:wq! 保存并退出， 并使得配置立刻生效：

1

source /etc/profile

3.3.7 创建消息存储路径

1
2
3
4

mkdir /usr/local/rocketmq/store
mkdir /usr/local/rocketmq/store/commitlog
mkdir /usr/local/rocketmq/store/consumequeue
mkdir /usr/local/rocketmq/store/index

3.3.8 broker配置文件

1）master1

服务器：192.168.25.135

1

vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-a.properties

修改配置如下：

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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0 表示 Master，>0 表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=SYNC_MASTER
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=SYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128

2）slave2

服务器：192.168.25.135

1

vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-b-s.properties

修改配置如下：

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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-b
#0 表示 Master，>0 表示 Slave
brokerId=1
#nameServer地址，分号分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=11011
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=SLAVE
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128

3）master2

服务器：192.168.25.138

1

vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-b.properties

修改配置如下：

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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-b
#0 表示 Master，>0 表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=SYNC_MASTER
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=SYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128

4）slave1

服务器：192.168.25.138

1

vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-a-s.properties

修改配置如下：

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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0 表示 Master，>0 表示 Slave
brokerId=1
#nameServer地址，分号分割
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=11011
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/usr/local/rocketmq/store
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#Broker 的角色
#- ASYNC_MASTER 异步复制Master
#- SYNC_MASTER 同步双写Master
#- SLAVE
brokerRole=SLAVE
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128

3.3.9 修改启动脚本文件

1）runbroker.sh

1

vi /usr/local/rocketmq/bin/runbroker.sh

需要根据内存大小进行适当的对JVM参数进行调整：

1
2
3

#===================================================
# 开发环境配置 JVM Configuration
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"

####2）runserver.sh

1

vim /usr/local/rocketmq/bin/runserver.sh

1

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

3.3.10 服务启动

1）启动NameServe集群

分别在192.168.25.135和192.168.25.138启动NameServer

1
2

cd /usr/local/rocketmq/bin
nohup sh mqnamesrv &

2）启动Broker集群

• 在192.168.25.135上启动master1和slave2

master1：

1
2

cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-syncbroker-a.properties &

slave2：

1
2

cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-b-s.properties &

• 在192.168.25.138上启动master2和slave2

master2

1
2

cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-b.properties &

slave1

1
2

cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-a-s.properties &

3.3.11 查看进程状态

启动后通过JPS查看启动进程

3.3.12 查看日志

1
2
3
4

# 查看nameServer日志
tail -500f ~/logs/rocketmqlogs/namesrv.log
# 查看broker日志
tail -500f ~/logs/rocketmqlogs/broker.log

3.4 mqadmin管理工具

3.4.1 使用方式

进入RocketMQ安装位置，在bin目录下执行./mqadmin {command} {args}

###3.4.2 命令介绍

####1）Topic相关

####2）集群相关

####3）Broker相关

####4）消息相关

5）消费者、消费组相关

6）连接相关

7）NameServer相关

8）其他

3.4.3 注意事项

• 几乎所有命令都需要配置-n表示NameServer地址，格式为ip:port
• 几乎所有命令都可以通过-h获取帮助
• 如果既有Broker地址（-b）配置项又有clusterName（-c）配置项，则优先以Broker地址执行命令；如果不配置Broker地址，则对集群中所有主机执行命令

3.5 集群监控平台搭建

3.5.1 概述

RocketMQ有一个对其扩展的开源项目incubator-rocketmq-externals，这个项目中有一个子模块叫rocketmq-console，这个便是管理控制台项目了，先将incubator-rocketmq-externals拉到本地，因为我们需要自己对rocketmq-console进行编译打包运行。

3.5.2 下载并编译打包

1
2
3

git clone https://github.com/apache/rocketmq-externals
cd rocketmq-console
mvn clean package -Dmaven.test.skip=true

注意：打包前在rocketmq-console中配置namesrv集群地址：

1

rocketmq.config.namesrvAddr=192.168.25.135:9876;192.168.25.138:9876

启动rocketmq-console：

1

java -jar rocketmq-console-ng-1.0.0.jar

启动成功后，我们就可以通过浏览器访问http://localhost:8080进入控制台界面了，如下图：

集群状态：

4. 消息发送样例

• 导入MQ客户端依赖

1
2
3
4
5

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.4.0</version>
</dependency>

• 消息发送者步骤分析r

1
2
3
4
5
6

1.创建消息生产者producer，并制定生产者组名
2.指定Nameserver地址
3.启动producer
4.创建消息对象，指定主题Topic、Tag和消息体
5.发送消息
6.关闭生产者producer

• 消息消费者步骤分析

1
2
3
4
5

1.创建消费者Consumer，制定消费者组名
2.指定Nameserver地址
3.订阅主题Topic和Tag
4.设置回调函数，处理消息
5.启动消费者consumer

4.1 基本样例

4.1.1 消息发送

1）发送同步消息

这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

public class SyncProducer {
	public static void main(String[] args) throws Exception {
    	// 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    	// 设置NameServer的地址
    	producer.setNamesrvAddr("localhost:9876");
    	// 启动Producer实例
        producer.start();
    	for (int i = 0; i < 100; i++) {
    	    // 创建消息，并指定Topic，Tag和消息体
    	    Message msg = new Message("TopicTest" /* Topic */,
        	"TagA" /* Tag */,
        	("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
        	);
        	// 发送消息到一个Broker
            SendResult sendResult = producer.send(msg);
            // 通过sendResult返回消息是否成功送达
            System.out.printf("%s%n", sendResult);
    	}
    	// 如果不再发送消息，关闭Producer实例。
    	producer.shutdown();
    }
}

2）发送异步消息

异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。

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

public class AsyncProducer {
	public static void main(String[] args) throws Exception {
    	// 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    	// 设置NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
    	// 启动Producer实例
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(0);
    	for (int i = 0; i < 100; i++) {
                final int index = i;
            	// 创建消息，并指定Topic，Tag和消息体
                Message msg = new Message("TopicTest",
                    "TagA",
                    "OrderID188",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                // SendCallback接收异步返回结果的回调
                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.printf("%-10d OK %s %n", index,
                            sendResult.getMsgId());
                    }
                    @Override
                    public void onException(Throwable e) {
      	              System.out.printf("%-10d Exception %s %n", index, e);
      	              e.printStackTrace();
                    }
            	});
    	}
    	// 如果不再发送消息，关闭Producer实例。
    	producer.shutdown();
    }
}

3）单向发送消息

这种方式主要用在不特别关心发送结果的场景，例如日志发送。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

public class OnewayProducer {
	public static void main(String[] args) throws Exception{
    	// 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    	// 设置NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
    	// 启动Producer实例
        producer.start();
    	for (int i = 0; i < 100; i++) {
        	// 创建消息，并指定Topic，Tag和消息体
        	Message msg = new Message("TopicTest" /* Topic */,
                "TagA" /* Tag */,
                ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
        	);
        	// 发送单向消息，没有任何返回结果
        	producer.sendOneway(msg);

    	}
    	// 如果不再发送消息，关闭Producer实例。
    	producer.shutdown();
    }
}

4.1.2 消费消息

1）负载均衡模式

消费者采用负载均衡方式消费消息，多个消费者共同消费队列消息，每个消费者处理的消息不同

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

public static void main(String[] args) throws Exception {
    // 实例化消息生产者,指定组名
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
    // 指定Namesrv地址信息.
    consumer.setNamesrvAddr("localhost:9876");
    // 订阅Topic
    consumer.subscribe("Test", "*");
    //负载均衡模式消费
    consumer.setMessageModel(MessageModel.CLUSTERING);
    // 注册回调函数，处理消息
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                        ConsumeConcurrentlyContext context) {
            System.out.printf("%s Receive New Messages: %s %n", 
                              Thread.currentThread().getName(), msgs);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    //启动消息者
    consumer.start();
    System.out.printf("Consumer Started.%n");
}

2）广播模式

消费者采用广播的方式消费消息，每个消费者消费的消息都是相同的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

public static void main(String[] args) throws Exception {
    // 实例化消息生产者,指定组名
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
    // 指定Namesrv地址信息.
    consumer.setNamesrvAddr("localhost:9876");
    // 订阅Topic
    consumer.subscribe("Test", "*");
    //广播模式消费
    consumer.setMessageModel(MessageModel.BROADCASTING);
    // 注册回调函数，处理消息
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                        ConsumeConcurrentlyContext context) {
            System.out.printf("%s Receive New Messages: %s %n", 
                              Thread.currentThread().getName(), msgs);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    //启动消息者
    consumer.start();
    System.out.printf("Consumer Started.%n");
}

4.2 顺序消息

消息有序指的是可以按照消息的发送顺序来消费(FIFO)。RocketMQ可以严格的保证消息有序，可以分为分区有序或者全局有序。

顺序消费的原理解析，在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列)；而消费消息的时候从多个queue上拉取消息，这种情况发送和消费是不能保证顺序。但是如果控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序。当发送和消费参与的queue只有一个，则是全局有序；如果多个queue参与，则为分区有序，即相对每个queue，消息都是有序的。

下面用订单进行分区有序的示例。一个订单的顺序流程是：创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中，消费时，同一个OrderId获取到的肯定是同一个队列。

4.2.1 顺序消息生产

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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

/**
* Producer，发送顺序消息
*/
public class Producer {

   public static void main(String[] args) throws Exception {
       DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");

       producer.setNamesrvAddr("127.0.0.1:9876");

       producer.start();

       String[] tags = new String[]{"TagA", "TagC", "TagD"};

       // 订单列表
       List<OrderStep> orderList = new Producer().buildOrders();

       Date date = new Date();
       SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
       String dateStr = sdf.format(date);
       for (int i = 0; i < 10; i++) {
           // 加个时间前缀
           String body = dateStr + " Hello RocketMQ " + orderList.get(i);
           Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i, body.getBytes());

           SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
               @Override
               public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                   Long id = (Long) arg;  //根据订单id选择发送queue
                   long index = id % mqs.size();
                   return mqs.get((int) index);
               }
           }, orderList.get(i).getOrderId());//订单id

           System.out.println(String.format("SendResult status:%s, queueId:%d, body:%s",
               sendResult.getSendStatus(),
               sendResult.getMessageQueue().getQueueId(),
               body));
       }

       producer.shutdown();
   }

   /**
    * 订单的步骤
    */
   private static class OrderStep {
       private long orderId;
       private String desc;

       public long getOrderId() {
           return orderId;
       }

       public void setOrderId(long orderId) {
           this.orderId = orderId;
       }

       public String getDesc() {
           return desc;
       }

       public void setDesc(String desc) {
           this.desc = desc;
       }

       @Override
       public String toString() {
           return "OrderStep{" +
               "orderId=" + orderId +
               ", desc='" + desc + '\'' +
               '}';
       }
   }

   /**
    * 生成模拟订单数据
    */
   private List<OrderStep> buildOrders() {
       List<OrderStep> orderList = new ArrayList<OrderStep>();

       OrderStep orderDemo = new OrderStep();
       orderDemo.setOrderId(15103111039L);
       orderDemo.setDesc("创建");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103111065L);
       orderDemo.setDesc("创建");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103111039L);
       orderDemo.setDesc("付款");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103117235L);
       orderDemo.setDesc("创建");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103111065L);
       orderDemo.setDesc("付款");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103117235L);
       orderDemo.setDesc("付款");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103111065L);
       orderDemo.setDesc("完成");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103111039L);
       orderDemo.setDesc("推送");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103117235L);
       orderDemo.setDesc("完成");
       orderList.add(orderDemo);

       orderDemo = new OrderStep();
       orderDemo.setOrderId(15103111039L);
       orderDemo.setDesc("完成");
       orderList.add(orderDemo);

       return orderList;
   }
}

4.2.2 顺序消费消息

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

/**
* 顺序消息消费，带事务方式（应用可控制Offset什么时候提交）
*/
public class ConsumerInOrder {

   public static void main(String[] args) throws Exception {
       DefaultMQPushConsumer consumer = new 
           DefaultMQPushConsumer("please_rename_unique_group_name_3");
       consumer.setNamesrvAddr("127.0.0.1:9876");
       /**
        * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
        * 如果非第一次启动，那么按照上次消费的位置继续消费
        */
       consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

       consumer.subscribe("TopicTest", "TagA || TagC || TagD");

       consumer.registerMessageListener(new MessageListenerOrderly() {

           Random random = new Random();

           @Override
           public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
               context.setAutoCommit(true);
               for (MessageExt msg : msgs) {
                   // 可以看到每个queue有唯一的consume线程来消费, 订单对每个queue(分区)有序
                   System.out.println("consumeThread=" + Thread.currentThread().getName() + "queueId=" + msg.getQueueId() + ", content:" + new String(msg.getBody()));
               }

               try {
                   //模拟业务逻辑处理中...
                   TimeUnit.SECONDS.sleep(random.nextInt(10));
               } catch (Exception e) {
                   e.printStackTrace();
               }
               return ConsumeOrderlyStatus.SUCCESS;
           }
       });

       consumer.start();

       System.out.println("Consumer Started.");
   }
}

4.3 延时消息

比如电商里，提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的状态，如果还是未付款就取消订单释放库存。

4.3.1 启动消息消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

public class ScheduledMessageConsumer {
   public static void main(String[] args) throws Exception {
      // 实例化消费者
      DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ExampleConsumer");
      // 订阅Topics
      consumer.subscribe("TestTopic", "*");
      // 注册消息监听者
      consumer.registerMessageListener(new MessageListenerConcurrently() {
          @Override
          public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
              for (MessageExt message : messages) {
                  // Print approximate delay time period
                  System.out.println("Receive message[msgId=" + message.getMsgId() + "] " + (System.currentTimeMillis() - message.getStoreTimestamp()) + "ms later");
              }
              return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
          }
      });
      // 启动消费者
      consumer.start();
  }
}

4.3.2 发送延时消息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

public class ScheduledMessageProducer {
   public static void main(String[] args) throws Exception {
      // 实例化一个生产者来产生延时消息
      DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
      // 启动生产者
      producer.start();
      int totalMessagesToSend = 100;
      for (int i = 0; i < totalMessagesToSend; i++) {
          Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());
          // 设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel)
          message.setDelayTimeLevel(3);
          // 发送消息
          producer.send(message);
      }
       // 关闭生产者
      producer.shutdown();
  }
}

###4.3.3 验证

您将会看到消息的消费比存储时间晚10秒

4.3.4 使用限制

1
2

// org/apache/rocketmq/store/config/MessageStoreConfig.java
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18

4.4 批量消息

批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的topic，相同的waitStoreMsgOK，而且不能是延时消息。此外，这一批消息的总大小不应超过4MB。

4.4.1 发送批量消息

如果您每次只发送不超过4MB的消息，则很容易使用批处理，样例如下：

1
2
3
4
5
6
7
8
9
10
11

String topic = "BatchTest";
List<Message> messages = new ArrayList<>();
messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));
try {
   producer.send(messages);
} catch (Exception e) {
   e.printStackTrace();
   //处理error
}

如果消息的总长度可能大于4MB时，这时候最好把消息进行分割

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
46
47
48
49
50
51
52
53
54
55

public class ListSplitter implements Iterator<List<Message>> {
   private final int SIZE_LIMIT = 1024 * 1024 * 4;
   private final List<Message> messages;
   private int currIndex;
   public ListSplitter(List<Message> messages) {
           this.messages = messages;
   }
    @Override 
    public boolean hasNext() {
       return currIndex < messages.size();
   }
   	@Override 
    public List<Message> next() {
       int nextIndex = currIndex;
       int totalSize = 0;
       for (; nextIndex < messages.size(); nextIndex++) {
           Message message = messages.get(nextIndex);
           int tmpSize = message.getTopic().length() + message.getBody().length;
           Map<String, String> properties = message.getProperties();
           for (Map.Entry<String, String> entry : properties.entrySet()) {
               tmpSize += entry.getKey().length() + entry.getValue().length();
           }
           tmpSize = tmpSize + 20; // 增加日志的开销20字节
           if (tmpSize > SIZE_LIMIT) {
               //单个消息超过了最大的限制
               //忽略,否则会阻塞分裂的进程
               if (nextIndex - currIndex == 0) {
                  //假如下一个子列表没有元素,则添加这个子列表然后退出循环,否则只是退出循环
                  nextIndex++;
               }
               break;
           }
           if (tmpSize + totalSize > SIZE_LIMIT) {
               break;
           } else {
               totalSize += tmpSize;
           }

       }
       List<Message> subList = messages.subList(currIndex, nextIndex);
       currIndex = nextIndex;
       return subList;
   }
}
//把大的消息分裂成若干个小的消息
ListSplitter splitter = new ListSplitter(messages);
while (splitter.hasNext()) {
  try {
      List<Message>  listItem = splitter.next();
      producer.send(listItem);
  } catch (Exception e) {
      e.printStackTrace();
      //处理error
  }
}

4.5 过滤消息

在大多数情况下，TAG是一个简单而有用的设计，其可以来选择您想要的消息。例如：

1
2

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");

消费者将接收包含TAGA或TAGB或TAGC的消息。但是限制是一个消息只能有一个标签，这对于复杂的场景可能不起作用。在这种情况下，可以使用SQL表达式筛选消息。SQL特性可以通过发送消息时的属性来进行计算。在RocketMQ定义的语法下，可以实现一些简单的逻辑。下面是一个例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

------------
| message  |
|----------|  a > 5 AND b = 'abc'
| a = 10   |  --------------------> Gotten
| b = 'abc'|
| c = true |
------------
------------
| message  |
|----------|   a > 5 AND b = 'abc'
| a = 1    |  --------------------> Missed
| b = 'abc'|
| c = true |
------------

4.5.1 SQL基本语法

RocketMQ只定义了一些基本语法来支持这个特性。你也可以很容易地扩展它。

• 数值比较，比如：>，>=，<，<=，BETWEEN，=；
• 字符比较，比如：=，<>，IN；
• IS NULL 或者 IS NOT NULL；
• 逻辑符号 AND，OR，NOT；

常量支持类型为：

• 数值，比如：123，3.1415；
• 字符，比如：‘abc’，必须用单引号包裹起来；
• NULL，特殊的常量
• 布尔值，TRUE 或 FALSE

只有使用push模式的消费者才能用使用SQL92标准的sql语句，接口如下：

1

public void subscribe(finalString topic, final MessageSelector messageSelector)

4.5.2 消息生产者

发送消息时，你能通过putUserProperty来设置消息的属性

1
2
3
4
5
6
7
8
9
10
11

DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
producer.start();
Message msg = new Message("TopicTest",
   tag,
   ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
);
// 设置一些属性
msg.putUserProperty("a", String.valueOf(i));
SendResult sendResult = producer.send(msg);

producer.shutdown();

4.5.3 消息消费者

用MessageSelector.bySql来使用sql筛选消息

1
2
3
4
5
6
7
8
9
10

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
// 只有订阅的消息有这个属性a, a >=0 and a <= 3
consumer.subscribe("TopicTest", MessageSelector.bySql("a between 0 and 3");
consumer.registerMessageListener(new MessageListenerConcurrently() {
   @Override
   public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
       return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
   }
});
consumer.start();

4.6 事务消息

###4.6.1 流程分析

上图说明了事务消息的大致方案，其中分为两个流程：正常事务消息的发送及提交、事务消息的补偿流程。

1）事务消息发送及提交

(1) 发送消息（half消息）。

(2) 服务端响应消息写入结果。

(3) 根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行）。

(4) 根据本地事务状态执行Commit或者Rollback（Commit操作生成消息索引，消息对消费者可见）

2）事务补偿

(1) 对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起一次“回查”

(2) Producer收到回查消息，检查回查消息对应的本地事务的状态

(3) 根据本地事务状态，重新Commit或者Rollback

其中，补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况。

3）事务消息状态

事务消息共有三种状态，提交状态、回滚状态、中间状态：

• TransactionStatus.CommitTransaction: 提交事务，它允许消费者消费此消息。
• TransactionStatus.RollbackTransaction: 回滚事务，它代表该消息将被删除，不允许被消费。
• TransactionStatus.Unknown: 中间状态，它代表需要检查消息队列来确定状态。

4.6.1 发送事务消息

1) 创建事务性生产者

使用 TransactionMQProducer类创建生产者，并指定唯一的 ProducerGroup，就可以设置自定义线程池来处理这些检查请求。执行本地事务后、需要根据执行结果对消息队列进行回复。回传的事务状态在请参考前一节。

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

public class Producer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        //创建事务监听器
        TransactionListener transactionListener = new TransactionListenerImpl();
        //创建消息生产者
        TransactionMQProducer producer = new TransactionMQProducer("group6");
        producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
        //生产者这是监听器
        producer.setTransactionListener(transactionListener);
        //启动消息生产者
        producer.start();
        String[] tags = new String[]{"TagA", "TagB", "TagC"};
        for (int i = 0; i < 3; i++) {
            try {
                Message msg = new Message("TransactionTopic", tags[i % tags.length], "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                System.out.printf("%s%n", sendResult);
                TimeUnit.SECONDS.sleep(1);
            } catch (MQClientException | UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
        //producer.shutdown();
    }
}

2）实现事务的监听接口

当发送半消息成功时，我们使用 executeLocalTransaction 方法来执行本地事务。它返回前一节中提到的三个事务状态之一。checkLocalTranscation 方法用于检查本地事务状态，并回应消息队列的检查请求。它也是返回前一节中提到的三个事务状态之一。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21