IDEA 界面主题
IDEA初次启动时可以选择黑或白主题:
以后也可以根据个人偏好,自由的设置IDEA的界面主题:
编辑器的颜色方案、字体设置
针对编辑器,可以单独设置它的颜色方案:
如果要设置字体,可以在这里设置:
IDEA起步(三) - Maven配置以及在pom文件中快速添加依赖
Maven配置
- 搜索
maven
配置 - 选择
maven
- 选择
maven
文件夹 - 可以设置自定义的
maven
配置文件
Maven项目在pom.xml中快速添加依赖
- 在
pom.xml
中需要添加依赖的位置,使用快捷键alt+insert
插入内容 - 选择
Dependency
- 根据
artifact
名查找 - 输入查找的关键词
- 选择依赖版本
也可以根据类名来查找包含这个类的依赖
- 选择根据类名查找
- 输入类名,可以看到,包名类名可以支持模糊查询
- 选择依赖
IDEA起步(四) - 创建Springboot项目,以及使用EditStarters插件快速添加依赖
创建Springboot项目
使用Springboot Initializr
创建Springboot项目
- 新建
Project
或Module
- 选择
Spring Initializr
- 选择使用
start.spring.io
start.spring.io
有时访问缓慢,或甚至无法访问,可以选择使用国内的镜像服务器:
spring.wht6.cn
start.aliyun.com
(支持Spring Cloud Alibaba)
与 STS 的 Boot Dashboard
类似的Spring boot 项目控制台
STS 的 Boot Dashboard
工具可以方便地控制Spring boot 项目的启停,IDEA提供了类似的工具:Services
工具窗口。
执行下面操作开启Spring boot控制台:
- 打开 View | Tool Windows |
Services
工具窗口
2. 找到 Services
工具窗口
3. 点击添加按钮
4. 选择Run Configuration Type
5. 选择添加Spring Boot
这里列出所有 Spring boot 项目的启动类,可以方便的控制项目启停。
Springboot 插件 - EditStarters
STS开发工具中提供的 Edit Starters
功能非常好用,可以方便的设置 Springboot 和 Spring Cloud 的依赖。IDEA中可以使用 EditStarters
插件来添加此功能。
- 安装
EditStarters
插件
- 在设置中选择
Plugins
- 选择插件市场来安装插件
- 搜索
EditStarters
- 点击
Install
安装,安装后重启IDEA
- 使用
EditStarters
添加Springboot依赖
- 在pom.xml中,按快捷键
Alt+Insert
- 选择
Edit Starters
- 填写
Spring Initializr
路径,
可以使用官方的start.spring.io
,或者也可以填写国内镜像地址。
- 选择要添加的依赖
RocketMQ (一) 安装
文件下载
安装
安装配置 jdk8
1. 上传jdk压缩文件
将文件jdk-8u212-linux-x64.tar.gz
上传到 /root 目录
2. 解压缩
执行解压命令
# 将jdk解压到 /usr/local/ 目录
tar -xf jdk-8u212-linux-x64.tar.gz -C /usr/local/
# 切换到 /usr/local/ 目录, 显示列表, 查看解压缩的jdk目录
cd /usr/local
ll
3. 配置环境变量
修改 /etc/profile 配置文件, 配置环境变量
vim /etc/profile
# 在文件末尾添加以下内容:
export JAVA_HOME=/usr/local/jdk1.8.0_212
export PATH=$JAVA_HOME/bin:$PATH
修改完后, 让环境变量立即生效
source /etc/profile
4. 验证
java -version
----------------------------------------------------------------
java version "1.8.0_212"
Java(TM) SE Runtime Environment (build 1.8.0_212-b10)
Java HotSpot(TM) 64-Bit Server VM (build 25.212-b10, mixed mode)
javac -version
---------------
javac 1.8.0_212
安装 RocketMQ
1. 下载 rocketmq 二进制文件
wget https://mirror.bit.edu.cn/apache/rocketmq/4.7.0/rocketmq-all-4.7.0-bin-release.zip
2. 解压缩 rocketmq
将 rocketmq 解压到 /usr/local/ 目录
unzip rocketmq-all-4.7.0-bin-release.zip -d /usr/local/
# 修改一下文件夹名,改成 rocketmq 方便使用
mv /usr/local/rocketmq-all-4.7.0-bin-release /usr/local/rocketmq
3. 配置环境变量 ROCKETMQ_HOME 和 PATH
为了后续操作方便可以配置环境变量,之后在任意位置都可以执行rocketmq的操作命令。
vim /etc/profile
# 在文件末尾添加以下内容:
export ROCKETMQ_HOME=/usr/local/rocketmq
export PATH=$ROCKETMQ_HOME/bin:$PATH
修改完后, 让环境变量立即生效
source /etc/profile
4. 减小 rocketmq 使用的内存
rocketmq需要启动两个服务: name server
和 broker
, name server
默认配置JVM使用的内存是4g, broker
默认配置JVM使用的内存是8g.
开发环境中如果内存不足, 服务可能会无法启动, 可以通过降低两个服务的内存, 使服务可以正常启动, 也可以节省内存.
修改 name server
内存改为 256m
cd /usr/local/rocketmq/
# 编辑 bin/runserver.sh
vim bin/runserver.sh
# 找到文件中下面这一行:
JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
# 将 -Xms4g -Xmx4g -Xmn2g 修改为 -Xms256m -Xmx256m -Xmn128m
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
修改 broker
内存改为 256m
# 编辑 bin/runbroker.sh
vim bin/runbroker.sh
# 找到文件中下面这一行:
JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"
# 将 -Xms8g -Xmx8g -Xmn4g 修改为 -Xms256m -Xmx256m -Xmn128m
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"
5. 启动 rocketmq
先启动 name server
# 进入 rocketmq 目录
cd /usr/local/rocketmq/
# 启动 name server
nohup sh bin/mqnamesrv &
# 查看运行日志, 看到"The Name Server boot success."表示启动成功
tail -f ~/logs/rocketmqlogs/namesrv.log
再启动 broker
# 启动 broker, 连接name server: localhost:9876
nohup sh bin/mqbroker -n localhost:9876 &
# 查看运行日志, 看到"The broker[......:10911] boot success."表示启动成功
tail -f ~/logs/rocketmqlogs/broker.log
6. 关闭防火墙
rocketmq的通信会用到多个端口, 为了方便测试我们关闭防火墙
# 关闭防火墙
systemctl stop firewalld.service
# 禁止防火墙开机启动
systemctl disable firewalld.service
测试
运行测试, 启动生产者发送消息, 启动消费者接收消息
# 通过环境变量, 告诉客户端程序name server的地址
export NAMESRV_ADDR=localhost:9876
# 启动生产者来测试发送消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
# 启动消费者来测试接收消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
RocketMQ 的关闭命令
关闭 broker
mqshutdown broker
关闭 nameserver
mqshutdown namesrv
管理界面
在开源项目 rocketmq-externals
中提供了rocketmq 的管理界面: 地址为: https://github.com/apache/rocketmq-externals
github 在国内访问缓慢, 也可以使用码云的镜像项目, 地址为: https://gitee.com/mirrors/RocketMQ-Externals
1. 克隆项目
cd /usr/local/rocketmq/
# 克隆 rocketmq-externals 项目
git clone https://gitee.com/mirrors/RocketMQ-Externals
2. maven打包管理界面项目
如果没有安装 maven, 请先执行 maven 安装命令
yum install -y maven
打包管理界面项目 rocketmq-console
.
打包过程中会下载各种依赖,比较缓慢,请耐心等待
# 进入管理界面项目的文件夹
cd RocketMQ-Externals/rocketmq-console
# 执行maven 打包命令, 执行时间较长, 请耐心等待
mvn clean package -Dmaven.test.skip=true
3. 运行启动管理界面
打包的 jar 文件在 target 目录, 进入目录执行jar文件
# 进入 target 目录
cd target
# 运行管理界面
nohup java -jar rocketmq-console-ng-1.0.1.jar --server.port=8080 --rocketmq.config.namesrvAddr=localhost:9876 &
访问管理界面:
RocketMQ (二) 双主双从同步复制集群方案
部署环境
作为测试环境,我们使用两台虚拟机来部署双主双从环境,具体结构如下:
整个集群由两个 name server 实例和四个 broker 实例组成
name server:
- 两台服务器分别启动两个name server
broker A 主从:
- 服务器1部署 broker A 主服务
- 服务器2部署 broker A 从服务
broker B 主从:
- 服务器2部署 broker B 主服务
- 服务器1部署 broker B 从服务
安装 Rocketmq
首先参照《RocketMQ (一) 安装》笔记,在两台虚拟机上安装 Rocketmq。或在一台虚拟机上装好后进行克隆。
建文件夹
在一台服务器上启动两个 broker 实例,需要为不同实例设置单独的数据存储目录。
为了方便起见,我们在两台服务器上都创建这四个实例所需要的的目录。
mkdir /usr/local/rocketmq/store/
mkdir /usr/local/rocketmq/store/broker-a
mkdir /usr/local/rocketmq/store/broker-a/commitlog
mkdir /usr/local/rocketmq/store/broker-b
mkdir /usr/local/rocketmq/store/broker-b/commitlog
mkdir /usr/local/rocketmq/store/broker-as
mkdir /usr/local/rocketmq/store/broker-as/commitlog
mkdir /usr/local/rocketmq/store/broker-bs
mkdir /usr/local/rocketmq/store/broker-bs/commitlog
配置
在 rocketmq/conf
目录下提供了四种集群方案的配置样例
- 2m-2s-async:双主双从异步复制
- 2m-2s-sync:双主双从同步复制
- 2m-noslave:双主
- dledger: raft主从切换
这里我们选择双主双从同步复制方案。
1. broker-a,a主服务器配置
在服务器1修改样例配置文件:rocketmq/conf/2m-2s-sync/broker-a.properties
在样例配置文件中,添加三项配置:
listenPort
:我们在一台服务器上要运行两个broker实例,所以两个实例的端口要有所区分。这里broker-a主服务器的端口使用默认的10911。storePathRootDir
:数据存储目录storePathCommitLog
:提交日志存储目录
brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0
deleteWhen=04
fileReservedTime=48
brokerRole=SYNC_MASTER
flushDiskType=ASYNC_FLUSH
listenPort=10911
storePathRootDir=/usr/local/rocketmq/store/broker-a
storePathCommitLog=/usr/local/rocketmq/store/broker-a/commitlog
2. broker-a slave,a从服务器配置
在服务器2修改样例配置文件:rocketmq/conf/2m-2s-sync/broker-a-s.properties
在样例配置文件中,添加三项配置:
listenPort
:我们在一台服务器上要运行两个broker实例,所以两个实例的端口要有所区分。这里broker-a slave从服务器的端口使用11911。storePathRootDir
:数据存储目录storePathCommitLog
:提交日志存储目录
brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=1
deleteWhen=04
fileReservedTime=48
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
listenPort=11911
storePathRootDir=/usr/local/rocketmq/store/broker-as
storePathCommitLog=/usr/local/rocketmq/store/broker-as/commitlog
3. broker-b,b主服务器配置
在服务器2修改样例配置文件:rocketmq/conf/2m-2s-sync/broker-b.properties
在样例配置文件中,添加三项配置:
listenPort
:我们在一台服务器上要运行两个broker实例,所以两个实例的端口要有所区分。这里broker-b主服务器的端口使用默认的10911。storePathRootDir
:数据存储目录storePathCommitLog
:提交日志存储目录
brokerClusterName=DefaultCluster
brokerName=broker-b
brokerId=0
deleteWhen=04
fileReservedTime=48
brokerRole=SYNC_MASTER
flushDiskType=ASYNC_FLUSH
listenPort=10911
storePathRootDir=/usr/local/rocketmq/store/broker-b
storePathCommitLog=/usr/local/rocketmq/store/broker-b/commitlog
4. broker-b slave,b从服务器配置
在服务器1修改样例配置文件:rocketmq/conf/2m-2s-sync/broker-b-s.properties
在样例配置文件中,添加三项配置:
listenPort
:我们在一台服务器上要运行两个broker实例,所以两个实例的端口要有所区分。这里broker-b slave从服务器的端口使用11911。storePathRootDir
:数据存储目录storePathCommitLog
:提交日志存储目录
brokerClusterName=DefaultCluster
brokerName=broker-b
brokerId=1
deleteWhen=04
fileReservedTime=48
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
listenPort=11911
storePathRootDir=/usr/local/rocketmq/store/broker-bs
storePathCommitLog=/usr/local/rocketmq/store/broker-bs/commitlog
配置要点说明
-
四台服务器的集群名
brokerClusterName
相同。集群名称相同的服务器共同组成服务集群 。 -
从服务器通过名字与主服务器关联在一起,
brokerName
与主服务器相同。 -
brokerId
为0是主服务器。从服务器的值是非零值,例如如果有四个从服务器,他们的brokerId
应该是 1,2,3,4。 -
brokerRole
的值为SYNC_MASTER
是同步复制的主服务器。如果是ASYNC_MASTER
则为异步复制的主服务器。
- 同步复制:消息复制到从服务器后才向生产者发回反馈信息。
- 异步复制:消息发到主服务器就向生产者发回反馈信息,之后再向从服务器复制。
启动
1. 启动两个 name server
在两台服务器上启动两个 name server,它们不用做任何集群的配置,都是作为独立服务运行,它们之间也不会进行数据复制。
所有broker服务启动后,要同时连接这两个 name server,向两个 name server 进行注册。
在两台服务器上都启动 name server:
nohup sh mqnamesrv &
2. 启动 broker a 的主从两台服务器
在服务器1上启动 broker a 主服务器:
参数说明:
- -n参数:指定name server地址列表,多个地址用分号分隔
- -c参数:指定配置文件,使用指定的配置文件启动 broker
nohup sh mqbroker \
-n '192.168.64.151:9876;192.168.64.152:9876' \
-c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a.properties \
&
在服务器2上启动 broker a 从服务器:
nohup sh mqbroker \
-n '192.168.64.151:9876;192.168.64.152:9876' \
-c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a-s.properties \
&
3. 启动 broker b 的主从两台服务器
在服务器2上启动 broker b 主服务器:
nohup sh mqbroker \
-n '192.168.64.151:9876;192.168.64.152:9876' \
-c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b.properties \
&
在服务器1上启动 broker b 从服务器:
nohup sh mqbroker \
-n '192.168.64.151:9876;192.168.64.152:9876' \
-c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b-s.properties \
&
检查启动的服务
在两台服务器上分别查看java进程,确认两台服务器上是否各启动了三个java进程,分别运行name server和两个broker。
# 查看 java 进程
jps
---------------------
12081 NamesrvStartup
15745 BrokerStartup
15595 BrokerStartup
16655 Jps
启动管理界面
# 进入 rocketmq-console 项目打包文件目录
cd /usr/local/rocketmq/RocketMQ-Externals/rocketmq-console/target/
# 启动管理界面
nohup java -jar rocketmq-console-ng-1.0.1.jar \
--server.port=8080 \
--rocketmq.config.namesrvAddr='192.168.64.151:9876;192.168.64.152:9876' \
&
查看集群状态:
java编程基础(一)二进制
二进制
对于任何已知类型的传统计算机而言,均以1和0作为编码形式进行程序计算。
在电路结构中,高电平与低电平对应表示1和0。
计算机内部表示任何数据都是用二进制。而与人进行交互时,可以将数据转换成人可以理解的内容:10进制、文字、图片、音频视频。
不同进制的计算方式是相通的:
不同进制的计算方式是相通的:
十进制 十进制逢十进一 |
二进制 二进制逢二进一 |
---|---|
9+1 --> 10 | 1+1 --> 10 |
99+1 --> 100 | 11+1 --> 100 |
从0开始一直加1:
十进制 | 二进制 |
---|---|
0 | 0 |
1 | 1 |
2 | 10 |
3 | 11 |
4 | 100 |
5 | 101 |
6 | 110 |
7 | 111 |
8 | 1000 |
9 | 1001 |
10 | 1010 |
11 | 1011 |
… | |
128 | 10000000 |
十进制乘10和二进制乘2运算比较:
十进制乘10 | 二进制乘2 |
---|---|
93*10 --> 930 | 1*2 --> 10 |
930*10 --> 9300 | 10*2 --> 100 |
9300*10 --> 93000 | 100*2 --> 1000 |
93000*10 --> 930000 | 1000*2 --> 10000 |
常用的二进制值:
十进制 | 二进制 | 用2的指数表示 |
---|---|---|
1 | 1 | 2^0 |
2 | 10 | 2^1 |
4 | 100 | 2^2 |
8 | 1000 | 2^3 |
16 | 10000 | 2^4 |
32 | 100000 | 2^5 |
64 | 1000000 | 2^6 |
128 | 10000000 | 2^7 |
256 | 100000000 | 2^8 |
512 | 1000000000 | 2^9 |
1024 | 10000000000 | 2^10 |
2048 | 100000000000 | 2^11 |
4096 | 1000000000000 | 2^12 |
8192 | 10000000000000 | 2^13 |
每加一个0都是乘2,所以用2的指数来表示。
例如 10000000,1后面有7位,就可以用2的7次方表示。
用2的指数可以很方便的表示任意的数字值:
10进制 | 2进制 | 2的指数 |
---|---|---|
1 | 00000001 | 2^0 |
8 | 00001000 | 2^3 |
16 | 00010000 | 2^4 |
32 | 00100000 | 2^5 |
相加 | 相加 | |
57 | 00111001 | – |
面试题
1000个苹果,分装在若干箱子中,不管客户要多少个苹果,都可以直接交付给客户几个箱子。
问:这1000个箱子如何进行装箱?
bit 和 byte
- bit - 位
- byte - 字节
一个 bit 表示一个电位,1 或 0;一个 byte 表示 8 个电位。
个人安装的光纤宽带如果是 100 兆宽带,下载最高速度只能到十几兆,这是因为采用的表示单位不同。100兆宽带指的是100兆bit,我们一般说下载速度指的是byte,那么100兆bit换算成byte需要除8,也就是 12.5 兆 byte。
java 中的 byte 类型
java 中的 byte
类型整数是单字节类型,也就是说,它使用 8 位(bit) 来表示整数。
8 位(bit)能表示的数字:
10进制 | 2进制 |
---|---|
0 | 00000000 |
1 | 00000001 |
2 | 00000010 |
3 | 00000011 |
4 | 00000100 |
5 | 00000101 |
6 | 00000110 |
7 | 00000111 |
8 | 00001000 |
… | … |
255 | 11111111 |
但是,java中,byte类型的左侧的一位规定用来表示符号,0表示正数,1表示负数。那么它表示数字的位就只有7位,而不是8位:
java 中 byte 类型表示正数:
10进制 | 2进制 |
---|---|
0 | 00000000 |
1 | 00000001 |
2 | 00000010 |
3 | 00000011 |
4 | 00000100 |
5 | 00000101 |
6 | 00000110 |
7 | 00000111 |
8 | 00001000 |
… | … |
127 | 01111111 |
java 中 byte 类型表示负数:
java 的 byte 类型如何表示负数?我们先从 0-1
这个运算来看:
java 中的负数运算,执行的是与正数相同的加减运算,这样执行 0-1
运算得到的二进制值是 11111111
,表示 -1,那么可以以此类推,继续执行减1运算减到最小:
10进制 | 2进制 |
---|---|
-128 |
10000000 |
… | … |
-8 |
11111000 |
-7 |
11111001 |
-6 |
11111010 |
-5 |
11111011 |
-4 |
11111100 |
-3 |
11111101 |
-2 |
11111110 |
-1 |
11111111 |
-------------------- | -------------------- |
0 | 00000000 |
1 | 00000001 |
2 | 00000010 |
3 | 00000011 |
4 | 00000100 |
5 | 00000101 |
6 | 00000110 |
7 | 00000111 |
8 | 00001000 |
… | … |
127 | 01111111 |
四种整数类型的最小和最大值
与 byte 类型同理,short,int 和 long 也是用相同的方式表示整数。这四种类型的最小值和最大值分别可以这样表示:
byte(1字节,8位):
2进制 | 2的指数 | |
---|---|---|
最小 | 10000000 | -2^7 |
最大 | 01111111 | 2^7 - 1 |
short(2字节,16位):
2进制 | 2的指数 | |
---|---|---|
最小 | 10000000 00000000 | -2^15 |
最大 | 01111111 11111111 | 2^15 - 1 |
int(4字节,32位):
2进制 | 2的指数 | |
---|---|---|
最小 | 10000000 00000000 00000000 00000000 | -2^31 |
最大 | 01111111 11111111 11111111 11111111 | 2^31 - 1 |
long(8字节,64位):
2进制 | 2的指数 | |
---|---|---|
最小 | 10000000 00000000 00000000 00000000 00000000 00000000 00000000 00000000 | -2^63 |
最大 | 01111111 11111111 11111111 11111111 11111111 11111111 11111111 11111111 | 2^63 - 1 |
二进制和十进制的互转
java 中提供了二进制和十进制的互转方法:
- 十进制转二进制
Integer.toBinaryString(int i)
String s = Integer.toBinaryString(257);
System.out.println(s);
// 打印结果: 100000001
- 二进制转十进制
Integer.parseInt(String s, int radix)
- 第二个参数指定进制
- 这里不支持直接使用负数的二进制码来表示负数,而是要用正数的二进制码添加负号来表示
int a = Integer.parseInt("1100101101", 2);
System.out.println(a);
// 打印结果: 813
练一练
输入十进制整数转成2进制;输入二进制整数转成十进制。
package demo;
import java.util.Scanner;
public class Test2 {
public static void main(String[] args) {
while (true) {
System.out.print("输入十进制整数:");
int a = new Scanner(System.in).nextInt();
String s = Integer.toBinaryString(a);
System.out.println(s);
}
}
}
package demo;
import java.util.Scanner;
public class Test3 {
public static void main(String[] args) {
while (true) {
System.out.print("输入整数的二进制码:");
String s = new Scanner(System.in).nextLine();
int a = Integer.parseInt(s, 2);
System.out.println(a);
}
}
}
java 代码中直接写二进制字面值
java 语法中,可以使用 0b
前缀来直接写二进制码表示的整数:
int a = 0b11001;
byte b = 0b1111111;
// 0b前缀语法不允许直接写负数的二进制码,这样写实际表示的是正的128
byte c = 0b10000000; //此处错误
// 要写byte的最小值-128,需要写 128 的二进制码,并加负号
byte d = -0b10000000;
代码仓库
https://gitee.com/benwang6/java-review
java编程基础(二)位运算符
位运算符
针对二进制运算,java提供了位运算符:
运算符 | 说明 |
---|---|
& | 位与 |
| | 位或 |
^ | 异或 |
~ | 求反 |
<< | 左移位 |
>> | 带符号右移位 |
>>> | 不带符号右移位 |
位运算
1. 位与 &
两个整数值的32位,每一位和每一位求与。
两位都是1与得的结果是1;只要有0结果就是0。
00000000 00000000 00000000 01101001
00000000 00000000 00000000 00100011 &
----------------------------------------
00000000 00000000 00000000 00100001
2. 位或 |
只要有1结果就是1;两位都是0结果是0。
00000000 00000000 00000000 01101001
00000000 00000000 00000000 00100011 |
----------------------------------------
00000000 00000000 00000000 01101011
3. 异或 ^
异或运算是“找不同”
不同是1;相同是0。
00000000 00000000 00000000 01101001
00000000 00000000 00000000 00100011 ^
----------------------------------------
00000000 00000000 00000000 01001010
异或还有个特点,就是对同一个值异或两次会得到原值。
试着用上面异或的结果再对第二个值求一次异或,这样可以还原成第一个值。
4. 求反 ~
1变0;0变1。
求反运算只有一个运算项
00000000 00000000 00000000 01101001 ~
----------------------------------------
11111111 11111111 11111111 10010110
移位运算
带符号右移位
所有位整体向右移动,右侧移出部分截断。
侧符号位是0,在左侧补0。
符号位是1,在左侧补1。
不带符号右移位
不管符号位是0还是1,左侧都补0。
1000000000000000000000011101011 >>> 2
结果:
0010000000000000000000000111010
面试题
乘2和除2,除了用 *
/
运算符,还可以如何运算?
练一练
1. 获得任意int
整数的二进制码
代码
package demo2;
import java.util.Scanner;
public class Test1 {
public static void main(String[] args) {
while (true) {
System.out.print("输入整数的二进制码:");
int a = new Scanner(System.in).nextInt();
String code = binaryCode(a);
System.out.println("二进制码: " + code);
}
}
static String binaryCode(int a) {
int p = 1 << 31; //最左侧位是1
String s = "";
for (int i = 0; i < 32; i++) {
int r = a & (p >>> i);
if (r == 0) {
s += "0";
} else {
s += "1";
}
}
return s;
}
}
2. int
整数的32位,分拆成4个byte
值
获取左侧第一个字节可以这样做:
- 把左侧第一个字节移动到最右侧
- 强转成
byte
,这时会截断int
,只保留右侧末尾的一个字节
以此类推,把每个字节都移动到最右侧再强转成 byte
,就可以完成分拆。
代码
package demo2;
import java.util.Scanner;
public class Test2 {
public static void main(String[] args) {
while (true) {
System.out.print("输入整数:");
int n = new Scanner(System.in).nextInt();
// int 整数转成四个 byte 值
byte[] a = split(n);
for (byte b : a) {
System.out.println(b);
}
}
}
static byte[] split(int n) {
byte[] a = new byte[4];
for (int i = 0; i < a.length; i++) {
// n循环右移四次,分别移动 24,16,8,0 位
a[i] = (byte) (n >> (8 * (3 - i)));
}
return a;
}
}
3. 四个byte
值 合并成一个 int
整数
合并第一个字节
首先准备一个整数0:
我们打算把这个字节放到左侧第一个字节:
byte
运算时会被自动转成 int
,变成32位,如果符号位是1,左侧会补1:
我们把这个字节移动到最左侧:
两个值求位或:
合并第二个字节
从上一步的结果开始:
这个个字节会先变成 int
,如果符号位是1,左侧会补1:
把它左移到第二个字节位置:
注意看,左侧8位现在都是1
只留下我们需要的8位,把其他24位都抹成0,像下面这样求个位与运算(两位都是1结果是1,有0就是0)
把其他位抹成0了,接下来再求位或,这样就完成了第二个字节的合并:
以此类推,可以继续完成后面两个字节的合并过程。
代码
package demo2;
import java.util.Scanner;
public class Test3 {
public static void main(String[] args) {
while (true) {
byte[] a = new byte[4];
System.out.println("输入四个byte值:");
for (int i = 0; i < a.length; i++) {
a[i] = new Scanner(System.in).nextByte();
}
int r = toInt(a);
System.out.println("合并结果: "+r);
}
}
static int toInt(byte[] a) {
int r = 0;
for (int i = 0; i < a.length; i++) {
// 先向左移动 24,16,8,0 位
int tmp = a[i] << (8 * (3 - i));
// 再把不需要的位抹成0
tmp = tmp & (0b11111111 << (8 * (3 - i)));
// 再求位或
r |= tmp;
}
return r;
}
}
代码仓库
https://gitee.com/benwang6/java-review
java编程基础(三)冒泡排序
冒泡排序运算逻辑分析
从上面的演示看到,排序运算主要依靠j向前递减来完成,j递减过程做两件事:
- 较小值向前移动
- 最小值移动到i位置
代码
package demo3;
import java.util.Arrays;
import java.util.Random;
public class Test1 {
public static void main(String[] args) {
int[] a = rnd();
System.out.println(Arrays.toString(a));
System.out.println("----------------------------");
sort(a);
System.out.println("----------------------------");
System.out.println(Arrays.toString(a));
}
static void sort(int[] a) {
for (int i = 0; i < a.length; i++) {
for (int j = a.length-1; j > i; j--) {
if (a[j-1]>a[j]) {
int t = a[j-1];
a[j-1] = a[j];
a[j] = t;
}
}
System.out.println(Arrays.toString(a));
}
}
static int[] rnd() {
int[] a = new int[5 + new Random().nextInt(6)];
for (int i = 0; i < a.length; i++) {
a[i] = new Random().nextInt(100);
}
return a;
}
}
冒泡排序优化
观察一下执行的结果:
根据前面的演示,到蓝线位置就应该结束了,为什么i
变量还在继续递增一直到末尾呢?我们能不能在排序已经完成的情况下让排序过程退出呢?
要判断排序已经完成是没有可以判断的条件的!没有条件我们就创造条件,继续看下面的演示:
代码
只需要再增加三处代码,注意看注释编号1,2,3:
package demo3;
import java.util.Arrays;
import java.util.Random;
public class Test1 {
public static void main(String[] args) {
int[] a = rnd();
System.out.println(Arrays.toString(a));
System.out.println("----------------------------");
sort(a);
System.out.println("----------------------------");
System.out.println(Arrays.toString(a));
}
static void sort(int[] a) {
for (int i = 0; i < a.length; i++) {
// 1. j开始之前设置flag的值
boolean flag = false;
for (int j = a.length-1; j > i; j--) {
if (a[j-1]>a[j]) {
int t = a[j-1];
a[j-1] = a[j];
a[j] = t;
// 2. 有交换,flag 修改成 true
flag = true;
}
}
// 3. j结束后检查 flag 的值
if(! flag) {
break; //排序完成,退出循环
}
System.out.println(Arrays.toString(a));
}
}
static int[] rnd() {
int[] a = new int[5 + new Random().nextInt(6)];
for (int i = 0; i < a.length; i++) {
a[i] = new Random().nextInt(100);
}
return a;
}
}
代码仓库
https://gitee.com/benwang6/java-review
RocketMQ (三) 基本原理
Topic 基本原理
在Rocketmq集群中新建 Topic1
在管理界面中新建主题Topic1
,为了方便观察测试效果,这里把写队列和读队列的数量都设置成3。
这样,在 broker-a 和 broker-b 上都创建了 Topic1 主题,并各创建了3写3读队列,共6写6读,如下图所示:
你也可以修改Topic1分别配置 broker-a 和 borker-b 上的队列数量。
perm 参数的含义
perm
参数是设置队列的读写权限,下面表格列出了可配置的值及其含义:
取值 | 含义 |
---|---|
6 | 同时开启读写 |
4 | 禁写 |
2 | 禁读 |
Topic 收发消息原理
生产者将消息发送到 Topic1 的其中一个写队列,消费者从对应的一个读队列接收消息。
生产者的负载均衡
生产者以轮询的方式向所有写队列发送消息,这些队列可能会分布在多个broker实例上。
消费者的负载均衡
一个 group 中的多个消费者,可以以负载均衡的方式来接收消息。
读取队列
被均匀分配给这些消费者,它们从指定的队列来接收消息。队列的分配可以采用不同的策略,这里简略介绍以下三种策略:
AllocateMessageQueueAveragely 平均分配
这是默认策略,它是这样分配队列的:
AllocateMessageQueueAveragelyByCircle 环形分配
如果使用环形分配,在消费者的代码中需要设置分配策略,代码如下:
consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragelyByCircle());
这种分配策略的逻辑很简单,所有0号队列分给0号消费者,所有1号队列分给1号消费者,以此类推。
AllocateMessageQueueConsistentHash 一致性哈希
如果使用一致性哈希算法进行分配,在消费者的代码中需要设置分配策略,代码如下:
consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueConsistentHash());
这种算法依靠一致性哈希算法,看当前消费者可以落到哪个虚拟节点,该虚拟节点对应哪个队列。
问题
思考一下,如果写队列比读队列多会怎样?反之会怎样?
NameServer 基本原理
NameServer 是 rocketmq 自己开发的一个轻型注册中心,他的作用相当于是 zk、eureka等。
rocketmq 为什么不使用 zk 呢?实际上 rocketmq 的早期版本使用的就是 zookeeper。
而 rocketmq 的架构设计决定了只需要一个轻量级的元数据服务器就足够了。杀鸡焉用牛刀?小区里,搞个货架就行了,建个仓库,又占地方,维护成本又高。
甚至,NameServer 都不需要有一个集群的管理者。以至于,NameServer 看起来都不像一个集群。事实上,NameServer 本质上来看,也不是一个集群。因为它的各个节点是独立的,不相关的。每个 NameServer 都是独立和 Producer、Consumer打交道。
基本认识
- NameServer主要用于存储Topic,Broker关系信息,功能简单,稳定性高。
- 各个NameServer节点之间不相关,不需要通信,单台宕机不影响其它节点。
- NameServer集群整体宕机不影响已建立关系的Concumer,Producer,Broker。
Broker、Producer、Consumer 与NameServer的通信
-
每个Borker和所有NameServer保持长连接,心跳间隔为30秒。每次心跳时还会携带当前的Topic信息。当某个Broker两分钟之内没有心跳,则认为该Broker下线,并调整内存中与该Broker相关的Topic信息。
-
Consumer 从 NameServer 获得 Topic 的路由信息,与对应的 Broker 建立长连接。间隔30秒发送心跳至Broker。Broker检查若发现某 Consumer 两分钟内无心跳则认为该Consumer下线,并通知该Consumer所有的消费者集群中的其他实例,触发该消费者集群重新负载均衡。
-
Producer 与消费者一样,也是从 NameServer 获得 Topic 的路由信息,与对应的 Broker 建立长连接,30秒发送一次心跳。Broker 也会认为两分钟内没有心跳的 Producer 下线。
RocketMQ (四) 使用RocketMQ原生API收发消息代码样例
pom文件
新建 maven 项目或 module,添加 rocketmq-client
依赖。
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>cn.tedu</groupId>
<artifactId>demo1</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.7.1</version>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-store</artifactId>
<version>4.7.1</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.8.0</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
</plugins>
</build>
</project>
同步消息
同步消息发送要保证强一致性,发到master的消息向slave复制后,才会向生产者发送反馈信息。
这种可靠性同步地发送方式使用的比较广泛,比如:重要的消息通知,短信通知。
生产者
package demo1;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import java.util.Scanner;
/*
发送同步消息
*/
public class Producer {
public static void main(String[] args) throws Exception {
/*
group 相同的生产者成为一个生产者组
标识发送同一类消息的Producer,通常发送逻辑一致。
发送普通消息的时候,仅标识使用,并无特别用处。
若发送事务消息,发送某条消息的producer-A宕机,
使得事务消息一直处于PREPARED状态并超时,
则broker会回查同一个group的其他producer,
确认这条消息应该commit还是rollback。
但开源版本并不完全支持事务消息(阉割了事务回查的代码)。?????
*/
DefaultMQProducer p = new DefaultMQProducer("producer-demo1");
/*
连接nameserver集群, 获得注册的broker信息
*/
p.setNamesrvAddr("192.168.64.151:9876:192.168.64.152:9876");
p.start();
/*
主题相当于是消息的分类, 一类消息使用一个主题
*/
String topic = "Topic1";
/*
tag 相当于是消息的二级分类, 在一个主题下, 可以通过 tag 再对消息进行分类
*/
String tag = "TagA";
while (true) {
System.out.print("输入消息,用逗号分隔多条消息: ");
String[] a = new Scanner(System.in).nextLine().split(",");
for (String s : a) {
Message msg = new Message(topic, tag, s.getBytes()); //一级分类, 二级分类, 消息内容
SendResult r = p.send(msg);// 发送消息后会得到服务器反馈, 包含: smsgId, sendStatus, queue, queueOffset, offsetMsgId
System.out.println(r);
}
}
}
}
消费者
消费者的要点:
1. push 和 pull
消费者有两种模式:push 和 pull。
push 模式由服务器主动向消费者发送消息;pull 模式由消费者主动向服务器请求消息。
在消费者处理能力有限时,为了减轻消费者的压力,可以采用pull模式。多数情况下都采用 pull 模式。
2. NameServer
消费者需要向 NameServer 询问 Topic 的路由信息。
3. Topic
从指定的Topic接收消息。Topic相当于是一级分类。
4. Tag
Topic 相当于是一级分类,Tag 相当于是2级分类。
- 多个 Tag 可以这样写:
TagA || TagB || TagC
- 不指定 Tag,或者说接收所有的 Tag,可以写星号:
*
package demo1;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class Consumer {
public static void main(String[] args) throws Exception {
/*
标识一类Consumer的集合名称,
这类Consumer通常消费一类消息,且消费逻辑一致。
同一个Consumer Group下的各个实例将共同消费
topic的消息,起到负载均衡的作用。
消费进度以Consumer Group为粒度管理,不同
Consumer Group之间消费进度彼此不受影响,
即消息A被Consumer Group1消费过,也会再
给Consumer Group2消费。
注: RocketMQ要求同一个Consumer Group的
消费者必须要拥有相同的注册信息,即必须要听一样
的topic(并且tag也一样)。
*/
DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo1");
c.setNamesrvAddr("192.168.64.151:9876:192.168.64.152:9876");
c.subscribe("Topic1", "TagA");
c.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext ctx) {
for (MessageExt msg : list) {
System.out.println(new String(msg.getBody()) + " - " + msg);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
c.start();
System.out.println("开始消费数据");
}
}
异步消息
master 收到消息后立即向生产者进行反馈。之后再以异步方式向 slave 复制消息。
异步消息通常用在对响应时间敏感的业务场景,即发送端不能容忍长时间地等待Broker的响应。
生产者
package demo2;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.Scanner;
/*
异步发送消息
一条消息送出后, 不必暂停等待服务器针对这条消息的反馈, 而是可以立即发送后续消息.
使用监听器, 以异步的方式接收服务器的反馈
*/
public class Producer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException {
DefaultMQProducer p = new DefaultMQProducer("producer-demo2");
p.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
p.start();
p.setRetryTimesWhenSendAsyncFailed(0);
String topic = "Topic2";
String tag = "TagA";
String key = "Key-demo2";
while (true) {
System.out.print("输入消息,用逗号分隔多条消息: ");
String[] a = new Scanner(System.in).nextLine().split(",");
for (String s : a) {
Message msg = new Message(topic, tag, key, s.getBytes());
p.send(msg, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.println("\n\n消息发送成功 : "+sendResult);
}
@Override
public void onException(Throwable throwable) {
System.out.println("\n\n消息发送失败");
}
});
System.out.println("--------------------消息已送出-----------------------");
}
}
}
}
消费者
package demo2;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
/*
与 demo1.Consumer 完全相同
*/
public class Consumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo2");
c.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
c.subscribe("Topic2", "TagA");
c.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt msg : list) {
System.out.println(new String(msg.getBody()) + " - " + msg);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
c.start();
System.out.println("开始消费数据");
}
}
单向消息
这种方式主要用在不特别关心发送结果的场景,例如日志发送。
生产者
package demo3;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.Scanner;
/*
单向消息
消息发出后, 服务器不会返回结果
*/
public class Producer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException {
DefaultMQProducer p = new DefaultMQProducer("producer-demo3");
p.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
p.start();
String topic = "Topic3";
String tag = "TagA";
while (true) {
System.out.print("输入消息,用逗号分隔多条消息: ");
String[] a = new Scanner(System.in).nextLine().split(",");
for (String s : a) {
Message msg = new Message(topic, tag, s.getBytes());
p.sendOneway(msg);
}
System.out.println("--------------------消息已送出-----------------------");
}
}
}
消费者
package demo3;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
/*
与 demo1.Consumer 完全相同
*/
public class Consumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo2");
c.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
c.subscribe("Topic3", "TagA");
c.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt msg : list) {
System.out.println(new String(msg.getBody()) + " - " + msg);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
c.start();
System.out.println("开始消费数据");
}
}
顺序消息
上图演示了 Rocketmq 顺序消息的基本原理:
- 同一组有序的消息序列,会被发送到同一个队列,按照 FIFO 的方式进行处理
- 一个队列只允许一个消费者线程接收消息,这样就保证消息按顺序被接收
下面以订单为例:
一个订单的顺序流程是:创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中。消费时,从同一个队列接收同一个订单的消息。
生产者
package demo4;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.List;
import java.util.Scanner;
/*
以下消息, 相同id的消息按顺序发送到同一个队列,
消费时也从同一个队列按顺序消费
topic
======================= queue1
======================= queue2
111,消息1 111,消息2 111,消息3 ------->======================= queue3
======================= queue4
222,消息1 222,消息2 222,消息3 ------->======================= queue5
======================= queue6
333,消息1 333,消息2 333,消息3 ------->======================= queue7
======================= queue8
......
*/
public class Producer {
static String[] msgs = {
"15103111039,创建",
"15103111065,创建",
"15103111039,付款",
"15103117235,创建",
"15103111065,付款",
"15103117235,付款",
"15103111065,完成",
"15103111039,推送",
"15103117235,完成",
"15103111039,完成"
};
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
DefaultMQProducer p = new DefaultMQProducer("producer-demo4");
p.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
p.start();
String topic = "Topic4";
String tag = "TagA";
for (String s : msgs) {
System.out.println("按回车发送此消息: "+s);
new Scanner(System.in).nextLine();
Message msg = new Message(topic, tag, s.getBytes());
String[] a = s.split(",");
long orderId = Long.parseLong(a[0]);
/*
MessageQueueSelector用来选择发送的队列,
这里用订单的id对队列数量取余来计算队列索引
send(msg, queueSelector, obj)
第三个参数会传递到queueSelector, 作为它的第三个参数
*/
SendResult r = p.send(msg, new MessageQueueSelector() {
/*
三个参数的含义:
queueList: 当前Topic中所有队列的列表
message: 消息
o: send()方法传入的orderId
*/
@Override
public MessageQueue select(List<MessageQueue> queueList, Message message, Object o) {
Long orderId = (Long) o;
//订单id对队列数量取余, 相同订单id得到相同的队列索引
long index = orderId % queueList.size();
System.out.println("消息已发送到: "+queueList.get((int) index));
return queueList.get((int) index);
}
}, orderId);
System.out.println(r+"\n\n");
}
}
}
消费者
package demo4;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class Consumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo4");
c.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
c.subscribe("Topic4", "*");
c.registerMessageListener(new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
String t = Thread.currentThread().getName();
for (MessageExt msg : list) {
System.out.println(t+" - "+ msg.getQueueId() + " - " +new String(msg.getBody()));
}
return ConsumeOrderlyStatus.SUCCESS;
}
});
c.start();
System.out.println("开始消费数据");
}
}
延时消息
消息发送到 Rocketmq 服务器后, 延迟一定时间再向消费者进行投递。
延时消息的使用场景:
比如电商里,提交了一个订单就可以发送一个延时消息,1h后去检查这个订单的状态,如果还是未付款就取消订单释放库存。
生产者发送消息时,对消息进行延时设置:
msg.setDelayTimeLevel(3);
其中 3
代表级别而不是一个具体的时间值,级别和延时时长对应关系是在 MessageStoreConfig
类种进行定义的:
this.messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
对应关系表:
级别 | 延时时长 |
---|---|
1 | 1s |
2 | 5s |
3 | 10s |
4 | 30s |
5 | 1m |
6 | 2m |
7 | 3m |
8 | 4m |
9 | 5m |
10 | 6m |
11 | 7m |
12 | 8m |
13 | 9m |
14 | 10m |
15 | 20m |
16 | 30m |
17 | 1h |
18 | 2h |
生产者
package demo5;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.Scanner;
/*
延时消息
延时消息的使用场景
比如电商里,提交了一个订单就可以发送一个延时消息,1h后去检查这个订单的状态,如果还是未付款就取消订单释放库存。
*/
public class Producer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
DefaultMQProducer p = new DefaultMQProducer("producer-demo5");
p.setNamesrvAddr("192.168.64.151:9876:192.168.64.152:9876");
p.start();
while (true) {
System.out.print("输入消息,用逗号分隔多条消息: ");
String[] a = new Scanner(System.in).nextLine().split(",");
for (String s : a) {
Message msg = new Message("Topic5", s.getBytes());
/*
设置消息的延迟时间,这里不支持任意的时间,只支持18个固定的延迟时长,
分别用Leven 1到18 来表示:
org/apache/rocketmq/store/config/MessageStoreConfig.java
this.messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
*/
msg.setDelayTimeLevel(3);
p.send(msg);
}
}
}
}
消费者
package demo5;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class Consumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo5");
c.setNamesrvAddr("192.168.64.151:9876:192.168.64.152:9876");
c.subscribe("Topic5", "*");
c.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext ctx) {
System.out.println("------------------------------");
for (MessageExt msg : list) {
long t = System.currentTimeMillis() - msg.getBornTimestamp();
System.out.println(new String(msg.getBody()) + " - 延迟: "+t);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
c.start();
System.out.println("开始消费数据");
}
}
批量消息
批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的topic,相同的waitStoreMsgOK,而且不能是延时消息。此外,这一批消息的总大小不应超过4MB。
生产者
package demo6;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.ArrayList;
import java.util.Scanner;
/*
批量发送消息能显著提高传递小消息的性能。限制是:
- 这些批量消息应该有相同的topic,
- 相同的waitStoreMsgOK,
- 而且不能是延时消息。
- 这一批消息的总大小不应超过4MB。
如果超出4M需要进行数据分割, 请参考官方代码样例https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md
*/
public class Producer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
DefaultMQProducer p = new DefaultMQProducer("producer-demo6");
p.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
p.start();
String topic = "Topic6";
while (true) {
System.out.print("输入消息,用逗号分隔多条消息: ");
String[] a = new Scanner(System.in).nextLine().split(",");
ArrayList<Message> messages = new ArrayList<>();
for (String s : a) {
messages.add(new Message(topic, s.getBytes()));
}
p.send(messages);
System.out.println("批量消息已发送");
}
}
}
消费者
package demo6;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class Consumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo6");
c.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
c.subscribe("Topic6", "*");
c.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt msg :
list) {
System.out.println("收到: "+new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
c.start();
System.out.println("开始消费数据");
}
}
消息过滤
Tag 过滤
Tag 可以满足大多数消息过滤的需求。使用 Tag 过滤非常简单,例如:
consumer.subscribe("Topic1", "TagA || TagB || TagC");
对自定义属性过滤
生产者可以在消息中添加自定义的属性:
msg.putUserProperty("prop1", "1");
msg.putUserProperty("prop2", "2");
消费者接收数据时,可以根据属性来过滤消息:
consumer.subscribe("Topic7", MessageSelector.bySql("prop1=1 or prop2=2"));
可以看到,自定义属性的过滤语法是 Sql 语法,RocketMQ只定义了一些基本语法来支持这个特性,支持的 Sql 过滤语法如下:
- 数值比较,比如:>,>=,<,<=,BETWEEN,=;
- 字符比较,比如:=,<>,IN;
- IS NULL 或者 IS NOT NULL;
- 逻辑符号 AND,OR,NOT;
生产者
package demo7;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.Random;
import java.util.Scanner;
/*
发送的消息中包含 tag 和 userProperty
消费者接收时,可以选择用 tag 或 userProperty 进行过滤
*/
public class Producer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
DefaultMQProducer p = new DefaultMQProducer("producer-demo7");
p.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
p.start();
String topic = "Topic7";
while (true) {
System.out.print("输入消息,用逗号分隔多条消息: ");
String[] a = new Scanner(System.in).nextLine().split(",");
System.out.print("输入Tag: ");
String tag = new Scanner(System.in).nextLine();
for (String s : a) {
Message msg = new Message(topic, tag, s.getBytes());
msg.putUserProperty("rnd", ""+new Random().nextInt(4));
p.send(msg);
}
}
}
}
消费者
package demo7;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
import java.util.Scanner;
/*
如果使用sql过滤,需要在 broker.properties 中添加配置来启用 sql 过滤:
enablePropertyFilter=true
*/
public class Consumer {
public static void main(String[] args) throws MQClientException {
System.out.print("使用Tag过滤还是使用Sql过滤(tag/sql): ");
String ts = new Scanner(System.in).nextLine();
DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo7");
c.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
if (ts.equalsIgnoreCase("tag")) {
System.out.println("使用Tag过滤: TagA || TagB || TagC");
c.subscribe("Topic7", "TagA || TagB || TagC");
} else {
System.out.println("使用Sql过滤: rnd=1 or rnd > 2");
c.subscribe("Topic7", MessageSelector.bySql("rnd=1 or rnd > 2"));
}
c.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt msg : list) {
System.out.println(new String(msg.getBody()) + " - " + msg.getUserProperty("rnd"));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
c.start();
System.out.println("开始消费数据");
}
}
事务消息
RocketMQ 提供了可靠性消息,也叫事务消息。下面分析一下其原理。
事务消息的原理
下面来看 RocketMQ 的事务消息是如何来发送“可靠消息”的,只需要以下三步:
- 发送半消息(半消息不会发送给消费者)
- 执行本地事务
- 提交消息
完成事务消息发送后,消费者就可以以正常的方式来消费数据。
RocketMQ 的自动重发机制在绝大多数情况下,都可以保证消息被正确消费。
假如消息最终消费失败了,还可以由人工处理进行托底。
上面分析的是正常情况下的执行流程。下面再来看两种错误情况:
- 事务执行失败时回滚消息
- 服务器无法得知消息状态时,需要主动回查消息状态
回滚:
消息回查:
生产者
package demo8;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.Scanner;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
public class Producer {
public static void main(String[] args) throws MQClientException {
TransactionMQProducer p = new TransactionMQProducer("producer-demo8");
p.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
p.setExecutorService(Executors.newFixedThreadPool(5));
p.setTransactionListener(new TransactionListener() {
ConcurrentHashMap<String, LocalTransactionState> localTx = new ConcurrentHashMap<>();
/*
在这里执行本地事务
*/
@Override
public LocalTransactionState executeLocalTransaction(Message message, Object o) {
System.out.println("执行本地事务");
if (Math.random()<0.333) {
System.out.println("本地事务执行成功, 按回车提交事务消息");
new Scanner(System.in).nextLine();
localTx.put(message.getTransactionId(), LocalTransactionState.COMMIT_MESSAGE);
return LocalTransactionState.COMMIT_MESSAGE;
} else if (Math.random()<0.666) {
System.out.println("本地事务执行失败, 按回车回滚事务消息");
new Scanner(System.in).nextLine();
localTx.put(message.getTransactionId(), LocalTransactionState.ROLLBACK_MESSAGE);
return LocalTransactionState.ROLLBACK_MESSAGE;
} else {
System.out.println("本地事务执行情况未知, 按回车继续");
new Scanner(System.in).nextLine();
localTx.put(message.getTransactionId(), LocalTransactionState.UNKNOW);
return LocalTransactionState.UNKNOW;
}
}
/*
回查方法
检测频率默认1分钟,可通过在broker.conf文件中设置transactionCheckInterval的值来改变默认值,单位为毫秒。
*/
@Override
public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
System.out.println("服务器正在回查消息状态");
LocalTransactionState s = localTx.get(messageExt.getTransactionId());
if (s == null || s == LocalTransactionState.UNKNOW) {
s = LocalTransactionState.ROLLBACK_MESSAGE;
}
return s;
}
});
p.start();
String topic = "Topic8";
while (true) {
System.out.print("输入消息,用逗号分隔多条消息: ");
String[] a = new Scanner(System.in).nextLine().split(",");
for (String s : a) {
Message msg = new Message(topic, s.getBytes());
System.out.println("---------发送半消息-----------");
TransactionSendResult r = p.sendMessageInTransaction(msg, null);
System.out.println("事务消息发送结果: "+ r.getLocalTransactionState().name());
}
}
}
}
消费者
package demo8;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
/*
如果返回 RECONSUME_LATER, 服务器会等待一会再重试发送消息
消息属性默认设置 DELAY=6, 等待时间为 2 分钟,
org/apache/rocketmq/store/config/MessageStoreConfig.java
this.messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
*/
public class Consumer {
public static void main(String[] args) throws Exception {
DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo8");
c.setNamesrvAddr("192.168.64.151:9876:192.168.64.152:9876");
c.subscribe("Topic8", "*");
c.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext ctx) {
for (MessageExt msg : list) {
System.out.println(new String(msg.getBody()) + " - " + msg);
}
if (Math.random()<0.5) {
System.out.println("消息处理完成");
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
} else {
System.out.println("消息处理失败, 要求服务器稍后重试发送消息");
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
});
c.start();
System.out.println("开始消费数据");
}
}
RocketMQ 发送事务消息原理分析和代码实现
事务消息
RocketMQ 提供了可靠性消息,也叫事务消息。下面分析一下其原理。
事务消息的原理
下面来看 RocketMQ 的事务消息是如何来发送“可靠消息”的,只需要以下三步:
- 发送半消息(半消息不会发送给消费者)
- 执行本地事务
- 提交消息
完成事务消息发送后,消费者就可以以正常的方式来消费数据。
RocketMQ 的自动重发机制在绝大多数情况下,都可以保证消息被正确消费。
假如消息最终消费失败了,还可以由人工处理进行托底。
上面分析的是正常情况下的执行流程。下面再来看两种错误情况:
- 事务执行失败时回滚消息
- 服务器无法得知消息状态时,需要主动回查消息状态
回滚:
消息回查:
生产者
package demo8;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.Scanner;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
public class Producer {
public static void main(String[] args) throws MQClientException {
TransactionMQProducer p = new TransactionMQProducer("producer-demo8");
p.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
p.setExecutorService(Executors.newFixedThreadPool(5));
p.setTransactionListener(new TransactionListener() {
ConcurrentHashMap<String, LocalTransactionState> localTx = new ConcurrentHashMap<>();
/*
在这里执行本地事务
*/
@Override
public LocalTransactionState executeLocalTransaction(Message message, Object o) {
System.out.println("执行本地事务");
if (Math.random()<0.333) {
System.out.println("本地事务执行成功, 按回车提交事务消息");
new Scanner(System.in).nextLine();
localTx.put(message.getTransactionId(), LocalTransactionState.COMMIT_MESSAGE);
return LocalTransactionState.COMMIT_MESSAGE;
} else if (Math.random()<0.666) {
System.out.println("本地事务执行失败, 按回车回滚事务消息");
new Scanner(System.in).nextLine();
localTx.put(message.getTransactionId(), LocalTransactionState.ROLLBACK_MESSAGE);
return LocalTransactionState.ROLLBACK_MESSAGE;
} else {
System.out.println("本地事务执行情况未知, 按回车继续");
new Scanner(System.in).nextLine();
localTx.put(message.getTransactionId(), LocalTransactionState.UNKNOW);
return LocalTransactionState.UNKNOW;
}
}
/*
回查方法
检测频率默认1分钟,可通过在broker.conf文件中设置transactionCheckInterval的值来改变默认值,单位为毫秒。
*/
@Override
public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
System.out.println("服务器正在回查消息状态");
LocalTransactionState s = localTx.get(messageExt.getTransactionId());
if (s == null || s == LocalTransactionState.UNKNOW) {
s = LocalTransactionState.ROLLBACK_MESSAGE;
}
return s;
}
});
p.start();
String topic = "Topic8";
while (true) {
System.out.print("输入消息,用逗号分隔多条消息: ");
String[] a = new Scanner(System.in).nextLine().split(",");
for (String s : a) {
Message msg = new Message(topic, s.getBytes());
System.out.println("---------发送半消息-----------");
TransactionSendResult r = p.sendMessageInTransaction(msg, null);
System.out.println("事务消息发送结果: "+ r.getLocalTransactionState().name());
}
}
}
}
消费者
package demo8;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
/*
如果返回 RECONSUME_LATER, 服务器会等待一会再重试发送消息
消息属性默认设置 DELAY=6, 等待时间为 2 分钟,
org/apache/rocketmq/store/config/MessageStoreConfig.java
this.messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
*/
public class Consumer {
public static void main(String[] args) throws Exception {
DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo8");
c.setNamesrvAddr("192.168.64.151:9876:192.168.64.152:9876");
c.subscribe("Topic8", "*");
c.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext ctx) {
for (MessageExt msg : list) {
System.out.println(new String(msg.getBody()) + " - " + msg);
}
if (Math.random()<0.5) {
System.out.println("消息处理完成");
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
} else {
System.out.println("消息处理失败, 要求服务器稍后重试发送消息");
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
});
c.start();
System.out.println("开始消费数据");
}
}
RocketMQ (五) Springboot 整合 RocketMQ 收发消息样例
pom文件
创建 springboot 项目,添加rocketmq-spring-boot-starter
依赖。
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.1.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>cn.tedu</groupId>
<artifactId>demo2-springboot-rocketmq</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>demo2-springboot-rocketmq</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.1.0</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
yml 配置
application.yml
rocketmq:
name-server: 192.168.64.151:9876;192.168.64.152:9876
application-demo1.yml
使用 demo1 profile 指定生产者组组名
rocketmq:
producer:
group: producer-demo1
application-demo2.yml
使用 demo2 profile 指定生产者组组名
rocketmq:
producer:
group: producer-demo2
demo 1
- 发送普通消息
- 发送 Spring 的通用 Message 对象
- 发送异步消息
- 发送顺序消息
package cn.tedu.demo2.m1;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
@Component
public class Producer {
@Autowired
private RocketMQTemplate t;
public void send() {
//发送消息
t.convertAndSend("Topic1:TagA", "Hello world! ");
//发送spring的Message
t.send("Topic1:TagA", MessageBuilder.withPayload("Hello world! ").build());
//发送异步消息
t.asyncSend("Topic1:TagA", "Hello world!", new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.println("发送成功");
}
@Override
public void onException(Throwable throwable) {
System.out.println("发送失败");
}
});
//发送顺序消息
t.syncSendOrderly("Topic1", "98456237,创建", "98456237");
t.syncSendOrderly("Topic1", "98456237,支付", "98456237");
t.syncSendOrderly("Topic1", "98456237,完成", "98456237");
}
}
package cn.tedu.demo2.m1;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
@Component
@RocketMQMessageListener(topic = "Topic1", consumerGroup = "consumer-demo1")
public class Consumer implements RocketMQListener<String> {
@Override
public void onMessage(String s) {
System.out.println("收到: "+s);
}
}
package cn.tedu.demo2.m1;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class Main {
public static void main(String[] args) {
SpringApplication.run(Main.class, args);
}
}
测试类,需要放在 test 文件夹
这个测试激活 demo1 profile
package cn.tedu.demo2.m1;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.ActiveProfiles;
@SpringBootTest
@ActiveProfiles("demo1")
public class Test1 {
@Autowired
private Producer p;
@Test
public void test() {
p.send();
}
}
demo 2
发送事务消息
package cn.tedu.demo2.m2;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
@Component
public class Producer {
@Autowired
private RocketMQTemplate t;
public void send() {
Message<String> msg = MessageBuilder.withPayload("Hello world!").build();
t.sendMessageInTransaction("Topic2:TagA", msg, null);
}
@RocketMQTransactionListener
class Lis implements RocketMQLocalTransactionListener {
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
System.out.println("执行本地事务");
return RocketMQLocalTransactionState.UNKNOWN;
}
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
System.out.println("执行事务回查");
return RocketMQLocalTransactionState.COMMIT;
}
}
}
package cn.tedu.demo2.m2;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
@Component
@RocketMQMessageListener(topic = "Topic2", consumerGroup = "consumer-demo2")
public class Consumer implements RocketMQListener<String> {
@Override
public void onMessage(String s) {
System.out.println("收到: "+s);
}
}
package cn.tedu.demo2.m2;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class Main {
public static void main(String[] args) {
SpringApplication.run(Main.class, args);
}
}
测试类,需要放在 test 文件夹
这个测试激活 demo2 profile
package cn.tedu.demo2.m2;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.ActiveProfiles;
import java.util.Scanner;
@SpringBootTest
@ActiveProfiles("demo2")
public class Test1 {
@Autowired
private Producer p;
@Test
public void test() {
p.send();
System.out.println("按回车结束");
new Scanner(System.in).nextLine();
}
}
分布式事务(一)Mysql本地事务和事务隔离级别
数据库事务
数据库事务由一组sql语句组成。
所有sql语句执行成功则事务整体成功;任一条sql语句失败则事务整体失败,数据恢复到事务之前的状态。
下面以转账为例进一步说明。
A 账户向 B 账户转账,需要更新两个账户的记录:
- A 账户减金额
update user set money=money-100 where id='A'
- B 账户加金额
update user set money=money+100 where id='B'
- 两条sql语句都成功则转账成功。
- 任意一条sql语句失败,恢复以前的状态。
数据操作的最小单元是事务,而不是一条sql语句!
Mysql 事务操作
开始事务
start transaction;
- 或
begin;
事务开始后,对数据的增删改操作不直接修改数据表,而是被记录在日志文件中。
提交事务
commit;
将日志中记录的操作,永久保存到数据表,并清空日志文件。
回滚事务
rollback;
直接清空日志文件
Mysql 事务操作测试
1.准备测试表
CREATE TABLE USER (
id INT PRIMARY KEY AUTO_INCREMENT,
username VARCHAR(128),
`password` CHAR(32)
) CHARSET=utf8;
# 插入一条数据
INSERT INTO `user`(username,`password`) VALUES('aaa', 'aaa');
2. 测试
打开两个终端,分别登录数据库。
# 在终端登录 mysql
mysql -uroot -p
# 切换到你的数据库
mysql> use 数据库名;
测试一
步骤 | 终端A | 终端B |
---|---|---|
1 | begin; |
begin; |
2 | insert into user(username) values('A'); |
|
3 | update user set username='X' where id=1; |
|
4 | select * from user; B查询不到A未提交的数据 |
|
5 | select * from user; A未提交的数据对于自己是可见的 |
|
6 | commit; |
|
7 | select * from user; 即使A已经提交,B也查询不到A提交的数据 |
|
8 | commit; |
|
9 | select * from user; B的事务结束后才能查询到A提交的数据 |
测试二
步骤 | 终端A | 终端B |
---|---|---|
1 | rollback; 为避免未结束的事务影响,先结束前面的事务 |
rollback; 为避免未结束的事务影响,先结束前面的事务 |
2 | begin; |
begin; |
3 | insert into user(username) values('B'); |
|
4 | update user set username='Y' where id=1; |
|
5 | select * from user; |
|
6 | commit; |
|
7 | select * from user; |
|
8 | insert into user(username) values('C'); |
|
9 | update user set username='Z' where id=1; |
|
10 | select * from user; |
|
11 | rollback; 回滚事务,B的数据修改全部丢弃,恢复到以前的状态。注意自增主键不会回滚 |
|
12 | select * from user; |
事务特性 ACID
A - 原子性 Atomic
一个事务是一个不可分割的工作单元,事务中包括的操作要么都做,要么都不做。
数据操作的最小单元是事务,而不是SQL语句 。
C - 一致性 Consistency
事务必须是使数据库从一个一致性状态变到另一个一致性状态。一致性与原子性是密切相关的。
例如:
- 转账前 a+b = 100
- 转帐后 a+b = 100
I - 隔离性 Isolation
一个事务的执行不能被其他事务干扰。
即一个事务内部的操作及使用的数据对并发的其他事务是隔离的,并发执行的各个事务之间不能互相干扰。
D - 持久性 Durancy
一个事务一旦提交,它对数据库中数据的改变就应该是永久性的。接下来的其他操作或故障不应该对其有任何影响。
数据库并发访问冲突问题
脏读
读取到其他事务未提交的数据。
不可重复读
- 重复读取同一数据时,与之前读取的数据不一致。
- 一个事务提交的数据,可以被另一个事务立即读取。
幻读
- 读取到已经被删除的数据。
- 读取不到新插入的数据。
Mysql 的四种事务隔离级别
事务之间为了避免互相干扰,执行时要进行隔离。也就是A执行时B要等待。但严格的隔离会造成性能的下降。
数据库为了兼顾数据安全和性能,可以在一定程度上允许多个事务并行执行。
Mysql 提供了四种隔离级别从低到高:
READ-UNCOMMITTED
READ-COMMITTED
REPEATABLE-READ
SERIALIZABLE
隔离级别越高数据越安全;越低性能越好,但会造成数据访问的问题:
可能引发的问题 | READ-UNCOMMITTED |
READ-COMMITTED |
REPEATABLE-READ |
SERIALIZABLE |
---|---|---|---|---|
幻读 | √ |
√ |
√ |
× |
不可重复读 | √ |
√ |
× | × |
脏读 | √ |
× | × | × |
Mysql 设置隔离级别
set tx_isolation='read-uncommitted';
set tx_isolation='read-committed';
# repeatable-read 是Mysql默认的隔离级别
set tx_isolation='repeatable-read';
set tx_isolation='serializable';
oracle mysql 8 使用 transaction_isolation
系统变量:
set transaction_isolation='read-uncommitted';
set transaction_isolation='read-committed';
# repeatable-read 是Mysql默认的隔离级别
set transaction_isolation='repeatable-read';
set transaction_isolation='serializable';
注意:set
设置的变量只对当前会话有效。需要进行全局设置使用 set global
Mysql 隔离级别测试
打开两个终端,分别登录数据库。
# 在终端登录 mysql
mysql -uroot -p
# 切换到你的数据库
mysql> use 数据库名;
测试一
步骤 | 终端A | 终端B |
---|---|---|
1 | set tx_isolation='read-uncommitted'; |
set tx_isolation='read-uncommitted'; |
2 | rollback; |
rollback; |
3 | begin; |
begin; |
4 | insert into user(username) values('D'); |
|
5 | update user set username='R' where id=1; |
|
6 | select * from user; 可以读取A未提交的数据,这些数据在真实数据表中并不存在 |
|
7 | rollback; |
|
8 | select * from user; A回滚后,B不能再重复读取这些数据 |
测试二
步骤 | 终端A | 终端B |
---|---|---|
1 | set tx_isolation='read-committed'; |
set tx_isolation='read-committed'; |
2 | rollback; |
rollback; |
3 | begin; |
begin; |
4 | `` | select * from user; |
5 | insert into user(username) values('E'); |
`` |
6 | update user set username='S' where id=1; |
`` |
7 | `` | select * from user; 不能读取A未提交的数据 |
8 | commit; |
`` |
9 | `` | select * from user; A提交后,可以立即读取 |
测试三
步骤 | 终端A | 终端B |
---|---|---|
1 | set tx_isolation='repeatable-read'; |
set tx_isolation='repeatable-read'; |
1 | rollback; |
rollback; |
1 | begin; |
begin; |
1 | `` | select * from user; |
1 | insert into user(username) values('F'); |
`` |
1 | update user set username='T' where id=1; |
`` |
1 | commit; |
`` |
1 | `` | select * from user; 即使A已经提交,这里也不能读取A已提交的数据。这里读取的结果要和前一次的结果一致 |
1 | `` | update user set password='111'; 可以修改A已提交的数据 |
1 | `` | select * from user; 修改后这些数据又变成可见的 |
测试四
步骤 | 终端A | 终端B |
---|---|---|
1 | rollback; |
rollback; |
1 | begin; |
begin; |
1 | `` | select * from user; |
1 | delete from user where id<5; |
`` |
1 | commit; |
`` |
1 | `` | select * from user; 仍然可以查询到已被A删除的数据 |
1 | `` | update user set password='222'; 但是不可能去修改这些已删除的数据 |
1 | `` | select * from user; 现在被删掉的数据还是可见的 |
1 | `` | commit; |
1 | `` | select * from user; 事务结束后这些数据不再可见 |
分布式事务(二)分布式事务方案
什么是分布式事务
首先这是普通事务:
下面是分布式事务:
在微服务系统中,每个微服务应用都可能会有自己的数据库,它们首先需要控制自己的本地事务。
一项业务操作可能会调用执行多个微服务。如何保证多个服务执行的多个数据库的操作整体成功或整体失败?这就是分布式事务要解决的问题。
理论部分
CAP 和 BASE 是对大规模互联网系统分布式实践的理论总结,如果没有实践为基础理论则难以理解。
这里建议先对分布式事务进行实践,之后再来阅读理论来互相印证。
CAP
请参考 百度百科 - CAP原则。
在分布式系统中,由于网络原因出现子系统之间无法通信的情况,就会造成分区。一般分布式系统中必须容忍这种情况,那么就需要在A和C之间进行取舍。
在分布式事务中,
- 如果保证CP,就意味着要让所有子系统的数据操作要么全部成功,要么全部失败,不允许有不一致的情况发生。但是强一致性会造成性能下降。
- 如果保证AP,就意味着可以牺牲一定的一致性,允许在各个子系统中存在有的数据操作成功,有的数据操作失败的情况,只要通过后续处理,能够达到最终一致即可。
BASE
请参考 百度百科 - BASE
分布式事务方案
分布式事务有以下解决方案:
- XA
- TCC
- Seata 框架 AT 事务
- SAGA
- 可靠消息最终一致性
- 最大努力通知
后面我们会对 Seata 框架 AT 事务
、TCC
和 可靠消息最终一致性
三个方案进行实践。