开发者博客 – IT技术 尽在开发者博客

开发者博客 – 科技是第一生产力


  • 首页

  • 归档

  • 搜索

最近遇到的问题记录:UrlEncode、UrlDecode

发表于 2021-01-10

本文阅读前了解知识:什么时候需要使用UrlEncode和UrlDecode函数

=======================================================================================================

作者使用谷歌浏览器,通过按下F12对第三方网站http协议的接口抓包进行分析操作。

场景

运维小哥哥偶尔使用某某外包公司的网站系统,做设备录入工作,流程简单:

录入设备信息

录入设备信息

  1. 录入设备基本信息,有7、8个字段需要输入,然后点击保存按钮;
  2. 基本信息保存成功,进入设备类型选择操作,然后点击生成设备标识按钮;
  3. 设备标识生成成功,录入设备关联的模块信息,简单设备只需要录入2条模块,复杂的设备有6条模块,每个模块有3、4个字段需要输入,最后点击保存。

一条设备录入成功,单身多年的手速可能也花不了几分钟,其实这也没啥。

突然领导说有1000个设备需要搞?运维小哥哥哭了😂,这时就该开发人员上场了:

  1. 运维准备一个Excel模板,输入需要录入的1000个设备基本信息、设备类型信息,这个工作量不大,就半天吧,最多一天工作量;
  2. 开发做个C/S客户端小工具,程序中按业务要求配置模块录入规则;
  3. 程序执行过程中录入一个设备就把生成的设备标识与设备关联;
  4. 全部录入完成,提供一个Excel导出,可将设备基本信息、生成的设备标识全部关联导出,工作完成。

经过几天的开发工作,开发哥哥将精心打磨的小工具交给运维小哥,运维小哥哥使用后投来了赞许的目光…

问题

前面铺垫的话有点啰嗦了,开发这个小工具时,开发小哥遇到一个问题:

xxx接口

xxx接口

这是某个接口的信息,Content-Type 是 application/x-www-form-urlencoded,下面参数使用的Form Data,即参数使用了UrlEncode,比如未编码前的一个参数:

1
ruby复制代码"Content":"{"AP_Name":"HK_7889","IP":"192.168.0.1"}"

编码后(可以使用这个在线URL编码解码工具验证):

1
perl复制代码"Content":"%7B%22AP_Name%22%3A%22HK_7889%22%2C%22IP%22%3A%2292.168.0.1%22%7D"

使用Postman测试时,未对参数使用UrlEncode,接口测试成功,开发这个小工具时,有3个接口都是类似的,未进行UrlEncode操作:

1
vbscript复制代码var client = new RestClient("http://admin.lqclass.com/api/device");client.Timeout = -1;var request = new RestRequest(Method.POST);request.AddHeader("Content-Type", "application/x-www-form-urlencoded");request.AddParameter("Content", "{\"AP_Name\":\"HK_7889\",\"IP\":\"92.168.0.1\"}");IRestResponse response = client.Execute(request);Console.WriteLine(response.Content);

但遇到稍微复杂一点的接口,比如截图中的参数为:

1
sql复制代码"Content":"{"AP_Name":"HK_7889","IP":"192.168.0.1","Module":[{"M_Name":"cameri0","Desc":"cameri0","AP_PUID":"54632325461320320"},{"M_Name":"cameri1","Desc":"cameri1","AP_PUID":"54636325461320320"},{"M_Name":"cameri2","Desc":"cameri2","AP_PUID":"54632325421320320"}]}"

Content值格式化看得清楚一点,Module是设备关联的模块信息:

1
json复制代码{  "AP_Name": "HK_7889",  "IP": "192.168.0.1",  "Module": [    {      "M_Name": "cameri0",      "Desc": "cameri0",      "AP_PUID": "54632325461320320"    },    {      "M_Name": "cameri1",      "Desc": "cameri1",      "AP_PUID": "54636325461320320"    },    {      "M_Name": "cameri2",      "Desc": "cameri2",      "AP_PUID": "54632325421320320"    }  ]}

实际UrlEncode后的参数为:

1
perl复制代码"Content":"%7B%22AP_Name%22%3A%22HK_7889%22%2C%22IP%22%3A%22192.168.0.1%22%2C%22Module%22%3A%22%255B%257B%2522M_Name%2522%253A%2522cameri0%2522%252C%2522Desc%2522%253A%2522cameri0%2522%252C%2522AP_PUID%2522%253A%252254632325461320320%2522%257D%252C%257B%2522M_Name%2522%253A%2522cameri1%2522%252C%2522Desc%2522%253A%2522cameri1%2522%252C%2522AP_PUID%2522%253A%252254636325461320320%2522%257D%252C%257B%2522M_Name%2522%253A%2522cameri2%2522%252C%2522Desc%2522%253A%2522cameri2%2522%252C%2522AP_PUID%2522%253A%252254632325421320320%2522%257D%255D%22%7D"

本来一般接口,如上面成功执行的C#代码那般直接未UrlEncode调用是没问题的。

但这个接口调用,服务器返回错误信息:“xxx解析失败”,调用代码如下:

1
swift复制代码var client = new RestClient("http://admin.lqclass.com/api/device");client.Timeout = -1;var request = new RestRequest(Method.POST);request.AddHeader("Content-Type", "application/x-www-form-urlencoded");request.AddParameter("Content", "{\"AP_Name\":\"HK_7889\",\"IP\":\"192.168.0.1\",\"Module\":[{\"M_Name\":\"cameri0\",\"Desc\":\"cameri0\",\"AP_PUID\":\"54632325461320320\"},{\"M_Name\":\"cameri1\",\"Desc\":\"cameri1\",\"AP_PUID\":\"54636325461320320\"},{\"M_Name\":\"cameri2\",\"Desc\":\"cameri2\",\"AP_PUID\":\"54632325421320320\"}]}");IRestResponse response = client.Execute(request);Console.WriteLine(response.Content);

两处调用代码哪里不同?只是Content值不一样,最后怀疑是不是需要手动进行UrlEncode?又不是url参数,为啥需要编码呢?不管啦,先编码了再说。

问题解决

参数编码后,调用:

1
perl复制代码var client = new RestClient("http://admin.lqclass.com/api/device");client.Timeout = -1;var request = new RestRequest(Method.POST);request.AddHeader("Content-Type", "application/x-www-form-urlencoded");request.AddParameter("Content", "%7B%22AP_Name%22%3A%22HK_7889%22%2C%22IP%22%3A%22192.168.0.1%22%2C%22Module%22%3A%22%255B%257B%2522M_Name%2522%253A%2522cameri0%2522%252C%2522Desc%2522%253A%2522cameri0%2522%252C%2522AP_PUID%2522%253A%252254632325461320320%2522%257D%252C%257B%2522M_Name%2522%253A%2522cameri1%2522%252C%2522Desc%2522%253A%2522cameri1%2522%252C%2522AP_PUID%2522%253A%252254636325461320320%2522%257D%252C%257B%2522M_Name%2522%253A%2522cameri2%2522%252C%2522Desc%2522%253A%2522cameri2%2522%252C%2522AP_PUID%2522%253A%252254632325421320320%2522%257D%255D%22%7D");IRestResponse response = client.Execute(request);Console.WriteLine(response.Content);

其实中间还做了一个参数的UrlEncode操作,即下面的Module参数值:

1
css复制代码"Content":{"AP_Name":"HK_7889","IP":"192.168.0.1","Module":[{"M_Name":"cameri0","Desc":"cameri0","AP_PUID":"54632325461320320"},{"M_Name":"cameri1","Desc":"cameri1","AP_PUID":"54636325461320320"},{"M_Name":"cameri2","Desc":"cameri2","AP_PUID":"54632325421320320"}]}

第一次UrlEncode,即先对Module的值进行UrlEncode:

1
perl复制代码"Content":{"AP_Name":"HK_7889","IP":"192.168.0.1","Module":%5B%7B%22M_Name%22%3A%22cameri0%22%2C%22Desc%22%3A%22cameri0%22%2C%22AP_PUID%22%3A%2254632325461320320%22%7D%2C%7B%22M_Name%22%3A%22cameri1%22%2C%22Desc%22%3A%22cameri1%22%2C%22AP_PUID%22%3A%2254636325461320320%22%7D%2C%7B%22M_Name%22%3A%22cameri2%22%2C%22Desc%22%3A%22cameri2%22%2C%22AP_PUID%22%3A%2254632325421320320%22%7D%5D}

第二次UrlEncode即是上面成功的参数方式了,对整个Content的值进行UrlEncode,看上面成功的参数,不重复贴了。

最后总结

抓别人数据包时,不要凭印象、已有知识判定该怎么怎么做,比如前面的参数,不使用UrlEncode时,调用成功了,其他包我是否也沿用相同的方式使用就正确呢?搞不定时,多尝试猜测的方法。

总结:“管他的,干就是了”。

本文使用的UrlEncode C# 代码:

1
vbnet复制代码public static string UrlEncode(string str){    StringBuilder sb = new StringBuilder();    byte[] byStr = System.Text.Encoding.UTF8.GetBytes(str); //默认是System.Text.Encoding.Default.GetBytes(str)    for (int i = 0; i < byStr.Length; i++)    {        sb.Append(@"%" + Convert.ToString(byStr[i], 16));    }    return (sb.ToString());}

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

数据库篇:mysql表设计原则-三范式 前言 参考文章

发表于 2021-01-10

前言

关系型数据库的表在设计时,我们常常需要考虑哪些业务字段应该放哪张表,字段是否应该拆分,表与表之间该怎么关联。那有没有相应的规范或者原则来指导我们去设计表呢?数据库设计三范式;三范式主要是解决表之间的关联关系,和字段冗余问题

关注公众号,一起交流,微信搜一搜: 潜行前行

第一范式

  • 列都是不可再分,第一范式的目标是确保每列的原子性,每列都是不可再分的最小数据单元
  • 身高体重是两个属性,违反第一范式,不能划分为同一个列
  • 符合第一范式的设计

第二范式

  • 首先满足第一范式,并且表中非主键列不存在对主键不依赖或者部分依赖,确保每个列都和主键相关。一般因为是存在多个主键,或者存在复合主键,因此需要拆表
  • 存在复合主键(学号,学科),而学科学分却只依赖分部主键-学科,不符合第二范式
  • 第二范式的正确示范

第三范式

  • 满足第二范式,并且表中的列不存在对非主键列的传递依赖,每列都和主键列直接相关,而不是间接相关
  • 在成绩表里,爱好是依赖学生的,学生又是依赖主键ID,存在传递依赖应该提取出学生的个人信息为表。
  • 符合第三范式的规范

欢迎指正文中错误

参考文章

  • mysql 数据库的设计三范式

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

k8s1201 集群安装 1 环境准备 2 安装k8s

发表于 2021-01-10

1 环境准备

  1. 1 机器环境

节点CPU核数必须是 :>= 2核 /内存要求必须是:>=2G ,否则k8s无法启动

DNS网络: 最好设置为 本地网络连通的DNS,否则网络不通,无法下载一些镜像

linux内核: linux内核必须是 4 版本以上,因此必须把linux核心进行升级

节点hostname 作用 IP
kmaster master 192.168.8.121
knode1 node1 192.168.8.122
knode2 node2 192.168.8.123

1.2 hostname

1
2
3
bash复制代码[root@base1 ~]# hostnamectl set-hostname kmaster --static
[root@base2 ~]# hostnamectl set-hostname knode1 --static
[root@base3 ~]# hostnamectl set-hostname knode2 --static

1.3 网络设置

1
2
3
4
5
6
7
8
9
10
bash复制代码[root@base1 ~]# vi /etc/sysconfig/network-scripts/ifcfg-ens33
BOOTPROTO="static" #dhcp改为static
ONBOOT="yes" #开机启用本配置
IPADDR=192.168.8.121 #静态IP 192.168.8.122/192.168.8.123
GATEWAY=192.168.8.2 #默认网关
NETMASK=255.255.255.0 #子网掩码
DNS1=114.114.114.114 #DNS 配置
DNS2=8.8.8.8 #DNS 配置

$# reboot

1.4 查看主机名

1
bash复制代码hostname

1.5 配置IP host映射关系

1
2
3
4
bash复制代码vi /etc/hosts
192.168.8.121 kmaster
192.168.8.122 knode1
192.168.8.123 knode2

1.6 安装依赖环境,注意:每一台机器都需要安装此依赖环境

1
bash复制代码yum install -y conntrack ntpdate ntp ipvsadm ipset jq iptables curl sysstatlibseccomp wget vim net-tools git iproute lrzsz bash-completion tree bridge-utils unzip bind-utils gcc

1.7 安装iptables,启动iptables,设置开机自启,清空iptables规则,保存当前规则到默认规则

1
2
3
4
bash复制代码# 关闭防火墙
systemctl stop firewalld && systemctl disable firewalld
# 置空iptables
yum -y install iptables-services && systemctl start iptables && systemctl enable iptables && iptables -F && service iptables save

1.8 关闭selinux

1
2
3
4
bash复制代码# 闭swap分区【虚拟内存】并且永久关闭虚拟内存
swapoff -a && sed -i '/ swap / s/^\(.*\)$/#\1/g' /etc/fstab
# 关闭selinux
setenforce 0 && sed -i 's/^SELINUX=.*/SELINUX=disabled/' /etc/selinux/config

1.9 升级Linux内核为4.44版本

1
2
3
4
5
6
7
8
9
10
11
bash复制代码rpm -Uvh http://www.elrepo.org/elrepo-release-7.0-4.el7.elrepo.noarch.rpm
#安装内核
yum --enablerepo=elrepo-kernel install -y kernel-lt
#设置开机从新内核启动
grub2-set-default 'CentOS Linux (4.4.248-1.el7.elrepo.x86_64) 7 (Core)'
4.4.248-1.el7.elrepo.x86_64

reboot
#注意:设置完内核后,需要重启服务器才会生效。
#查询内核
uname -r

2 安装k8s

2.1 对于k8s,调整内核参数 kubernetes.conf

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
bash复制代码cat > kubernetes.conf <<EOF
net.bridge.bridge-nf-call-iptables=1
net.bridge.bridge-nf-call-ip6tables=1
net.ipv4.ip_forward=1
net.ipv4.tcp_tw_recycle=0
vm.swappiness=0
vm.overcommit_memory=1
vm.panic_on_oom=0
fs.inotify.max_user_instances=8192
fs.inotify.max_user_watches=1048576
fs.file-max=52706963
fs.nr_open=52706963
net.ipv6.conf.all.disable_ipv6=1
net.netfilter.nf_conntrack_max=2310720
EOF
#将优化内核文件拷贝到/etc/sysctl.d/文件夹下,这样优化文件开机的时候能够被调用
cp kubernetes.conf /etc/sysctl.d/kubernetes.conf
#手动刷新,让优化文件立即生效
sysctl -p /etc/sysctl.d/kubernetes.conf
sysctl: cannot stat /proc/sys/net/netfilter/nf_conntrack_max: No such file or directory

错误解决:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
bash复制代码lsmod |grep conntrack
modprobe ip_conntrack
lsmod |grep conntrack
nf_conntrack_ipv4 20480 0
nf_defrag_ipv4 16384 1 nf_conntrack_ipv4
nf_conntrack 114688 1 nf_conntrack_ipv4

sysctl -p /etc/sysctl.d/kubernetes.conf
net.bridge.bridge-nf-call-iptables = 1
net.bridge.bridge-nf-call-ip6tables = 1
net.ipv4.ip_forward = 1
net.ipv4.tcp_tw_recycle = 0
vm.swappiness = 0
vm.overcommit_memory = 1
vm.panic_on_oom = 0
fs.inotify.max_user_instances = 8192
fs.inotify.max_user_watches = 1048576
fs.file-max = 52706963
fs.nr_open = 52706963
net.ipv6.conf.all.disable_ipv6 = 1
net.netfilter.nf_conntrack_max = 2310720

2.2 调整系统临时区

1
2
3
4
5
6
7
bash复制代码#设置系统时区为中国/上海
timedatectl set-timezone Asia/Tokyo
#将当前的UTC 时间写入硬件时钟
timedatectl set-local-rtc 0
#重启依赖于系统时间的服务
systemctl restart rsyslog
systemctl restart crond

2.3 关闭系统不需要的服务

1
bash复制代码systemctl stop postfix && systemctl disable postfix

2.4 设置日志保存方式

2.4.1 创建保存日志的目录

1
bash复制代码mkdir /var/log/journal

2.4.2 创建配置文件存放目录

1
bash复制代码mkdir /etc/systemd/journald.conf.d

2.4.3 创建配置文件

1
2
3
4
5
6
7
8
9
10
11
12
bash复制代码cat > /etc/systemd/journald.conf.d/99-prophet.conf <<EOF
[Journal]
Storage=persistent
Compress=yes
SyncIntervalSec=5m
RateLimitInterval=30s
RateLimitBurst=1000
SystemMaxUse=10G
SystemMaxFileSize=200M
MaxRetentionSec=2week
ForwardToSyslog=no
EOF

2.4.4 重启systemd journald 的配置

1
bash复制代码systemctl restart systemd-journald

2.4.5 打开文件数调整(可忽略,不执行)

1
2
bash复制代码echo "* soft nofile 65536" >> /etc/security/limits.conf
echo "* hard nofile 65536" >> /etc/security/limits.conf

2.4.6 kube-proxy 开启 ipvs 前置条件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
bash复制代码modprobe br_netfilter
cat > /etc/sysconfig/modules/ipvs.modules <<EOF
#!/bin/bash
modprobe -- ip_vs
modprobe -- ip_vs_rr
modprobe -- ip_vs_wrr
modprobe -- ip_vs_sh
modprobe -- nf_conntrack_ipv4
EOF
#使用lsmod命令查看这些文件是否被引导
chmod 755 /etc/sysconfig/modules/ipvs.modules && bash /etc/sysconfig/modules/ipvs.modules && lsmod | grep -e ip_vs -e nf_conntrack_ipv4
ip_vs_sh 16384 0
ip_vs_wrr 16384 0
ip_vs_rr 16384 0
ip_vs 147456 6 ip_vs_rr,ip_vs_sh,ip_vs_wrr
nf_conntrack_ipv4 20480 0
nf_defrag_ipv4 16384 1 nf_conntrack_ipv4
nf_conntrack 114688 2 ip_vs,nf_conntrack_ipv4
libcrc32c 16384 2 xfs,ip_vs

3 docker部署

3.1 安装docker

1
2
3
4
5
6
bash复制代码yum install -y yum-utils device-mapper-persistent-data lvm2

#紧接着配置一个稳定的仓库、仓库配置会保存到/etc/yum.repos.d/docker-ce.repo文件中
yum-config-manager --add-repo https://download.docker.com/linux/centos/docker-ce.repo
#更新Yum安装的相关Docker软件包&安装Docker CE
yum update -y && yum install docker-ce

3.2 设置docker daemon文件

1
2
3
4
5
6
7
8
9
bash复制代码#创建/etc/docker目录
mkdir /etc/docker
#更新daemon.json文件
cat > /etc/docker/daemon.json <<EOF
{"exec-opts":["native.cgroupdriver=systemd"],"log-driver":"json-file","log-opts":{"max-size":"100m"}}
EOF
#注意:一定注意编码问题,出现错误---查看命令:journalctl -amu docker 即可发现错误
#创建,存储docker配置文件
mkdir -p /etc/systemd/system/docker.service.d

3.3 重启docker服务

1
bash复制代码systemctl daemon-reload && systemctl restart docker && systemctl enable docker

4 kubeadm[一键安装k8s]

4.1 yum仓库镜像

国内

1
2
3
4
5
6
7
8
9
10
bash复制代码cat <<EOF > /etc/yum.repos.d/kubernetes.repo
[kubernetes]
name=Kubernetes
baseurl=http://mirrors.aliyun.com/kubernetes/yum/repos/kubernetes-el7-x86_64
enabled=1
gpgcheck=0
repo_gpgcheck=0
gpgkey=http://mirrors.aliyun.com/kubernetes/yum/doc/yum-key.gpg
http://mirrors.aliyun.com/kubernetes/yum/doc/rpm-package-key.gpg
EOF

官网

1
2
3
4
5
6
7
8
9
bash复制代码cat <<EOF > /etc/yum.repos.d/kubernetes.repo
[kubernetes]
name=Kubernetes
baseurl=https://packages.cloud.google.com/yum/repos/kubernetes-el7-x86_64
enabled=1
gpgcheck=1
repo_gpgcheck=1
gpgkey=https://packages.cloud.google.com/yum/doc/yum-key.gpg https://packages.cloud.google.com/yum/doc/rpm-package-key.gpg
EOF

4.2 安装kubeadm 、kubelet、kubectl(1.20.1)

1
2
3
bash复制代码yum install -y kubelet kubeadm kubectl --disableexcludes=kubernetes
#启动 kubelet
systemctl enable kubelet && systemctl start kubelet

5 准备k8s镜像

5.1 在线拉取镜像

生成默认kubeadm.conf文件

1
bash复制代码kubeadm config print init-defaults > kubeadm.conf

编辑kubeadm.conf,将Kubernetes版本修改为v1.20.1

下载镜像

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
bash复制代码kubeadm config images pull --config kubeadm.conf
[config/images] Pulled k8s.gcr.io/kube-apiserver:v1.20.1
[config/images] Pulled k8s.gcr.io/kube-controller-manager:v1.20.1
[config/images] Pulled k8s.gcr.io/kube-scheduler:v1.20.1
[config/images] Pulled k8s.gcr.io/kube-proxy:v1.20.1
[config/images] Pulled k8s.gcr.io/pause:3.2
[config/images] Pulled k8s.gcr.io/etcd:3.4.13-0
[config/images] Pulled k8s.gcr.io/coredns:1.7.0

docker images
k8s.gcr.io/kube-proxy v1.20.1 e3f6fcd87756 11 days ago 118MB
k8s.gcr.io/kube-apiserver v1.20.1 75c7f7112080 11 days ago 122MB
k8s.gcr.io/kube-controller-manager v1.20.1 2893d78e47dc 11 days ago 116MB
k8s.gcr.io/kube-scheduler v1.20.1 4aa0b4397bbb 11 days ago 46.4MB
k8s.gcr.io/etcd 3.4.13-0 0369cf4303ff 4 months ago 253MB
k8s.gcr.io/coredns 1.7.0 bfe3a36ebd25 6 months ago 45.2MB
k8s.gcr.io/pause 3.2 80d28bedfe5d 10 months ago 683kB

保存镜像

1
2
3
4
5
6
7
8
9
10
11
12
bash复制代码mkdir kubeadm-basic.images
cd kubeadm-basic.images
docker save k8s.gcr.io/kube-apiserver:v1.20.1 > apiserver.tar
docker save k8s.gcr.io/coredns:1.7.0 > coredns.tar
docker save k8s.gcr.io/etcd:3.4.13-0 > etcd.tar
docker save k8s.gcr.io/kube-controller-manager:v1.20.1 > kubec-con-man.tar
docker save k8s.gcr.io/pause:3.2 > pause.tar
docker save k8s.gcr.io/kube-proxy:v1.20.1 > proxy.tar
docker save k8s.gcr.io/kube-scheduler:v1.20.1 > scheduler.tar

cd ..
tar zcvf kubeadm-basic.images.tar.gz kubeadm-basic.images

5.2 离线镜像

链接:pan.baidu.com/s/1UAF_-_sG…

提取码:548z

上传镜像压缩包kubeadm-basic.images.tar.gz,把压缩包中的镜像导入到本地镜像仓库

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
bash复制代码[root@kmaster ~]# ll
total 216676
-rw-------. 1 root root 1391 Dec 22 04:42 anaconda-ks.cfg
drwxr-xr-x 2 root root 142 Dec 30 07:55 kubeadm-basic.images
-rw-r--r-- 1 root root 221857746 Dec 30 08:01 kubeadm-basic.images.tar.gz
-rw-r--r-- 1 root root 827 Dec 30 07:34 kubeadm.conf
-rw-r--r-- 1 root root 20 Dec 30 07:00 kube-images.tar.gz
-rw-r--r-- 1 root root 364 Dec 30 03:40 kubernetes.conf
[root@kmaster ~]# ll kubeadm-basic.images
total 692188
-rw-r--r-- 1 root root 122923520 Dec 30 07:54 apiserver.tar
-rw-r--r-- 1 root root 45364736 Dec 30 07:54 coredns.tar
-rw-r--r-- 1 root root 254677504 Dec 30 07:54 etcd.tar
-rw-r--r-- 1 root root 117107200 Dec 30 07:54 kubec-con-man.tar
-rw-r--r-- 1 root root 691712 Dec 30 07:55 pause.tar
-rw-r--r-- 1 root root 120377856 Dec 30 07:55 proxy.tar
-rw-r--r-- 1 root root 47643136 Dec 30 07:55 scheduler.tar

编写脚本问题,导入镜像包到本地docker镜像仓库:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
bash复制代码# kubeadm 初始化 k8s 集群的时候,会从gce Google云中下载响应的镜像,且镜像相对比较大,下载比较慢
#1 导入镜像脚本代码(在任意目录下创建sh脚本文件:image-load.sh)
#! /bin/bash
#注意 镜像解压的目录位置
ls /root/kubeadm-basic.images > /tmp/images-list.txt
cd /root/kubeadm-basic.images
for i in $(cat /tmp/images-list.txt)
do
docker load -i $i
done
rm -rf /tmp/images-list.txt

#2 修改权限,可执行权限
chmod 755 image-load.sh

#3 开始执行,镜像导入
./image-load.sh

#4 传输文件及镜像到其他node节点
#拷贝到knode1节点
scp -r image-load.sh kubeadm-basic.images root@knode1:/root/
#拷贝到knode2
scp -r image-load.sh kubeadm-basic.images root@knode2:/root/

5.3 node节点导入镜像

knode1导入镜像

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
bash复制代码[root@knode1 ~]# ./image-load.sh
Loaded image: k8s.gcr.io/kube-apiserver:v1.20.1
Loaded image: k8s.gcr.io/coredns:1.7.0
Loaded image: k8s.gcr.io/etcd:3.4.13-0
Loaded image: k8s.gcr.io/kube-controller-manager:v1.20.1
Loaded image: k8s.gcr.io/pause:3.2
Loaded image: k8s.gcr.io/kube-proxy:v1.20.1
Loaded image: k8s.gcr.io/kube-scheduler:v1.20.1
[root@knode1 ~]# docker images
REPOSITORY TAG IMAGE ID CREATED SIZE
k8s.gcr.io/kube-proxy v1.20.1 e3f6fcd87756 11 days ago 118MB
k8s.gcr.io/kube-apiserver v1.20.1 75c7f7112080 11 days ago 122MB
k8s.gcr.io/kube-controller-manager v1.20.1 2893d78e47dc 11 days ago 116MB
k8s.gcr.io/kube-scheduler v1.20.1 4aa0b4397bbb 11 days ago 46.4MB
k8s.gcr.io/etcd 3.4.13-0 0369cf4303ff 4 months ago 253MB
k8s.gcr.io/coredns 1.7.0 bfe3a36ebd25 6 months ago 45.2MB
k8s.gcr.io/pause 3.2 80d28bedfe5d 10 months ago 683kB

knode2导入镜像

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
bash复制代码[root@knode2 ~]# ./image-load.sh
Loaded image: k8s.gcr.io/kube-apiserver:v1.20.1
Loaded image: k8s.gcr.io/coredns:1.7.0
Loaded image: k8s.gcr.io/etcd:3.4.13-0
Loaded image: k8s.gcr.io/kube-controller-manager:v1.20.1
Loaded image: k8s.gcr.io/pause:3.2
Loaded image: k8s.gcr.io/kube-proxy:v1.20.1
Loaded image: k8s.gcr.io/kube-scheduler:v1.20.1
[root@knode2 ~]# docker images
REPOSITORY TAG IMAGE ID CREATED SIZE
k8s.gcr.io/kube-proxy v1.20.1 e3f6fcd87756 11 days ago 118MB
k8s.gcr.io/kube-apiserver v1.20.1 75c7f7112080 11 days ago 122MB
k8s.gcr.io/kube-controller-manager v1.20.1 2893d78e47dc 11 days ago 116MB
k8s.gcr.io/kube-scheduler v1.20.1 4aa0b4397bbb 11 days ago 46.4MB
k8s.gcr.io/etcd 3.4.13-0 0369cf4303ff 4 months ago 253MB
k8s.gcr.io/coredns 1.7.0 bfe3a36ebd25 6 months ago 45.2MB
k8s.gcr.io/pause 3.2 80d28bedfe5d 10 months ago 683kB

6 k8s部署

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
bash复制代码#初始化主节点----只需要在主节点执行

#1 拉去yaml资源配置文件
kubeadm config print init-defaults > kubeadm-config.yaml

#2 修改yaml资源文件
localAPIEndpoint:
advertiseAddress: 192.168.8.121 # 注意: 修改配置文件的IP地址
kubernetesVersion: v1.20.1 # 注意:修改版本号,必须和kubectl版本保持一致
networking:
dnsDomain: cluster.local
# 指定flannel模型通信 pod网段地址,此网段和flannel网络一致
podSubnet: "10.244.0.0/16"
serviceSubnet: "10.96.0.0/12"
#指定使用ipvs网络进行通信
---
apiVersion: kubeadm.k8s.io/v1beta2
kind: kubeProxyConfiguration
featureGates:
supportipvsproxymodedm.ymlvim kubeadm.yml: true
mode: ipvs

#3 初始化主节点,开始部署
kubeadm init --config=kubeadm-config.yaml --upload-certs | tee kubeadm-init.log
#注意:执行此命令,CPU核心数量必须大于1核,否则无法执行成功
W1230 09:44:35.116411 1495 strict.go:47] unknown configuration schema.GroupVersionKind{Group:"kubeadm.k8s.io", Version:"v1beta2", Kind:"kubeProxyConfiguration"} for scheme definitions in "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm/scheme/scheme.go:31" and "k8s.io/kubernetes/cmd/kubeadm/app/componentconfigs/scheme.go:28"
[config] WARNING: Ignored YAML document with GroupVersionKind kubeadm.k8s.io/v1beta2, Kind=kubeProxyConfiguration
[init] Using Kubernetes version: v1.20.1
[preflight] Running pre-flight checks
[WARNING SystemVerification]: this Docker version is not on the list of validated versions: 20.10.1. Latest validated version: 19.03
[preflight] Pulling images required for setting up a Kubernetes cluster
[preflight] This might take a minute or two, depending on the speed of your internet connection
[preflight] You can also perform this action in beforehand using 'kubeadm config images pull'
[certs] Using certificateDir folder "/etc/kubernetes/pki"
[certs] Generating "ca" certificate and key
[certs] Generating "apiserver" certificate and key
[certs] apiserver serving cert is signed for DNS names [kmaster kubernetes kubernetes.default kubernetes.default.svc kubernetes.default.svc.cluster.local] and IPs [10.96.0.1 192.168.8.121]
[certs] Generating "apiserver-kubelet-client" certificate and key
[certs] Generating "front-proxy-ca" certificate and key
[certs] Generating "front-proxy-client" certificate and key
[certs] Generating "etcd/ca" certificate and key
[certs] Generating "etcd/server" certificate and key
[certs] etcd/server serving cert is signed for DNS names [kmaster localhost] and IPs [192.168.8.121 127.0.0.1 ::1]
[certs] Generating "etcd/peer" certificate and key
[certs] etcd/peer serving cert is signed for DNS names [kmaster localhost] and IPs [192.168.8.121 127.0.0.1 ::1]
[certs] Generating "etcd/healthcheck-client" certificate and key
[certs] Generating "apiserver-etcd-client" certificate and key
[certs] Generating "sa" key and public key
[kubeconfig] Using kubeconfig folder "/etc/kubernetes"
[kubeconfig] Writing "admin.conf" kubeconfig file
[kubeconfig] Writing "kubelet.conf" kubeconfig file
[kubeconfig] Writing "controller-manager.conf" kubeconfig file
[kubeconfig] Writing "scheduler.conf" kubeconfig file
[kubelet-start] Writing kubelet environment file with flags to file "/var/lib/kubelet/kubeadm-flags.env"
[kubelet-start] Writing kubelet configuration to file "/var/lib/kubelet/config.yaml"
[kubelet-start] Starting the kubelet
[control-plane] Using manifest folder "/etc/kubernetes/manifests"
[control-plane] Creating static Pod manifest for "kube-apiserver"
[control-plane] Creating static Pod manifest for "kube-controller-manager"
[control-plane] Creating static Pod manifest for "kube-scheduler"
[etcd] Creating static Pod manifest for local etcd in "/etc/kubernetes/manifests"
[wait-control-plane] Waiting for the kubelet to boot up the control plane as static Pods from directory "/etc/kubernetes/manifests". This can take up to 4m0s
[apiclient] All control plane components are healthy after 8.503909 seconds
[upload-config] Storing the configuration used in ConfigMap "kubeadm-config" in the "kube-system" Namespace
[kubelet] Creating a ConfigMap "kubelet-config-1.20" in namespace kube-system with the configuration for the kubelets in the cluster
[upload-certs] Storing the certificates in Secret "kubeadm-certs" in the "kube-system" Namespace
[upload-certs] Using certificate key:
7ecfa579dfa66c0ea9c87146aa5130c1692b85a4d16cfc860473064a75c113c5
[mark-control-plane] Marking the node kmaster as control-plane by adding the labels "node-role.kubernetes.io/master=''" and "node-role.kubernetes.io/control-plane='' (deprecated)"
[mark-control-plane] Marking the node kmaster as control-plane by adding the taints [node-role.kubernetes.io/master:NoSchedule]
[bootstrap-token] Using token: abcdef.0123456789abcdef
[bootstrap-token] Configuring bootstrap tokens, cluster-info ConfigMap, RBAC Roles
[bootstrap-token] configured RBAC rules to allow Node Bootstrap tokens to get nodes
[bootstrap-token] configured RBAC rules to allow Node Bootstrap tokens to post CSRs in order for nodes to get long term certificate credentials
[bootstrap-token] configured RBAC rules to allow the csrapprover controller automatically approve CSRs from a Node Bootstrap Token
[bootstrap-token] configured RBAC rules to allow certificate rotation for all node client certificates in the cluster
[bootstrap-token] Creating the "cluster-info" ConfigMap in the "kube-public" namespace
[kubelet-finalize] Updating "/etc/kubernetes/kubelet.conf" to point to a rotatable kubelet client certificate and key
[addons] Applied essential addon: CoreDNS
[addons] Applied essential addon: kube-proxy

Your Kubernetes control-plane has initialized successfully!

To start using your cluster, you need to run the following as a regular user:

mkdir -p $HOME/.kube
sudo cp -i /etc/kubernetes/admin.conf $HOME/.kube/config
sudo chown $(id -u):$(id -g) $HOME/.kube/config

Alternatively, if you are the root user, you can run:

export KUBECONFIG=/etc/kubernetes/admin.conf

You should now deploy a pod network to the cluster.
Run "kubectl apply -f [podnetwork].yaml" with one of the options listed at:
https://kubernetes.io/docs/concepts/cluster-administration/addons/

Then you can join any number of worker nodes by running the following on each as root:

kubeadm join 192.168.8.121:6443 --token abcdef.0123456789abcdef \
--discovery-token-ca-cert-hash sha256:7459fa01464531734d3eee182461b77b043d31eff7df2233635654d7c199c947
[root@kmaster ~]#

参考kubeadm-config.yaml

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
config.yaml复制代码apiVersion: kubeadm.k8s.io/v1beta2
bootstrapTokens:
- groups:
- system:bootstrappers:kubeadm:default-node-token
token: abcdef.0123456789abcdef
ttl: 24h0m0s
usages:
- signing
- authentication
kind: InitConfiguration
localAPIEndpoint:
advertiseAddress: 192.168.8.121
bindPort: 6443
nodeRegistration:
criSocket: /var/run/dockershim.sock
name: kmaster
taints:
- effect: NoSchedule
key: node-role.kubernetes.io/master
---
apiServer:
timeoutForControlPlane: 4m0s
apiVersion: kubeadm.k8s.io/v1beta2
certificatesDir: /etc/kubernetes/pki
clusterName: kubernetes
controllerManager: {}
dns:
type: CoreDNS
etcd:
local:
dataDir: /var/lib/etcd
imageRepository: k8s.gcr.io
kind: ClusterConfiguration
kubernetesVersion: v1.20.1
networking:
dnsDomain: cluster.local
podSubnet: 10.244.0.0/16
serviceSubnet: 10.96.0.0/12
scheduler: {}
---
apiVersion: kubeadm.k8s.io/v1beta2
kind: kubeProxyConfiguration
featureGates:
supportipvsproxymodedm.ymlvim kubeadm.yml: true
mode: ipvs

按照k8s指示,执行以下命令:

1
2
3
4
bash复制代码mkdir -p $HOME/.kube
sudo cp -i /etc/kubernetes/admin.conf $HOME/.kube/config
sudo chown $(id -u):$(id -g) $HOME/.kube/config
export KUBECONFIG=/etc/kubernetes/admin.conf

执行命令前:

1
2
bash复制代码kubectl get node
The connection to the server localhost:8080 was refused - did you specify the right host or port?

执行命令后

1
2
3
bash复制代码kubectl get node
NAME STATUS ROLES AGE VERSION
kmaster NotReady control-plane,master 7m24s v1.20.1

我们发现已经可以成功查询node节点信息了,但是节点的状态却是NotReady,不是Runing的状态。

原因是此时我们使用ipvs+flannel的方式进行网络通信,但是flannel网络插件还没有部署,因此节点状态为NotReady

7 flannel插件

1
2
3
4
5
6
7
bash复制代码#部署flannel网络插件---只需要在主节点执行
#1 下载flannel网络插件
wget https://raw.githubusercontent.com/coreos/flannel/master/Documentation/kube-flannel.yml
#2 部署flannel
kubectl create -f kube-flannel.yml
#也可进行部署网络
kubectl apply -f https://raw.githubusercontent.com/coreos/flannel/master/Documentation/kube-flannel.yml

验证

1
2
3
4
5
6
7
8
9
10
bash复制代码[root@kmaster ~]# kubectl get pod -n kube-system
NAME READY STATUS RESTARTS AGE
coredns-74ff55c5b-5n6zs 1/1 Running 0 15m
coredns-74ff55c5b-r9469 1/1 Running 0 15m
etcd-kmaster 1/1 Running 0 15m
kube-apiserver-kmaster 1/1 Running 0 15m
kube-controller-manager-kmaster 1/1 Running 0 15m
kube-flannel-ds-n4sbp 1/1 Running 0 89s
kube-proxy-t7bvn 1/1 Running 0 15m
kube-scheduler-kmaster 1/1 Running 0 15m

8 追加Node节点

1
2
3
4
5
6
bash复制代码# 假如主节点以及其余工作节点,执行安装日志中的命令即可
# 查看日志文件
cat kubeadm-init.log
# 负责命令到其他几个node节点进行执行即可
kubeadm join 192.168.8.121:6443 --token abcdef.0123456789abcdef \
--discovery-token-ca-cert-hash sha256:7459fa01464531734d3eee182461b77b043d31eff7df2233635654d7c199c947

knode1

1
2
bash复制代码kubeadm join 192.168.8.121:6443 --token abcdef.0123456789abcdef \
--discovery-token-ca-cert-hash sha256:7459fa01464531734d3eee182461b77b043d31eff7df2233635654d7c199c947

knode2

1
2
bash复制代码kubeadm join 192.168.8.121:6443 --token abcdef.0123456789abcdef \
--discovery-token-ca-cert-hash sha256:7459fa01464531734d3eee182461b77b043d31eff7df2233635654d7c199c947

9 验证状态

1
2
3
4
5
bash复制代码[root@kmaster ~]# kubectl get node
NAME STATUS ROLES AGE VERSION
kmaster Ready control-plane,master 26m v1.20.1
knode1 Ready <none> 5m37s v1.20.1
knode2 Ready <none> 5m28s v1.20.1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
bash复制代码[root@kmaster ~]# kubectl get pod -n kube-system -o wide
NAME READY STATUS RESTARTS AGE IP NODE NOMINATED NODE READINESS GATES
coredns-74ff55c5b-5n6zs 1/1 Running 0 27m 10.244.0.2 kmaster <none> <none>
coredns-74ff55c5b-r9469 1/1 Running 0 27m 10.244.0.3 kmaster <none> <none>
etcd-kmaster 1/1 Running 0 27m 192.168.8.121 kmaster <none> <none>
kube-apiserver-kmaster 1/1 Running 0 27m 192.168.8.121 kmaster <none> <none>
kube-controller-manager-kmaster 1/1 Running 0 27m 192.168.8.121 kmaster <none> <none>
kube-flannel-ds-9td5g 1/1 Running 0 7m12s 192.168.8.122 knode1 <none> <none>
kube-flannel-ds-n4sbp 1/1 Running 0 13m 192.168.8.121 kmaster <none> <none>
kube-flannel-ds-rvfbt 1/1 Running 0 7m3s 192.168.8.123 knode2 <none> <none>
kube-proxy-knhtb 1/1 Running 0 7m12s 192.168.8.122 knode1 <none> <none>
kube-proxy-t7bvn 1/1 Running 0 27m 192.168.8.121 kmaster <none> <none>
kube-proxy-vpxqm 1/1 Running 0 7m3s 192.168.8.123 knode2 <none> <none>
kube-scheduler-kmaster 1/1 Running 0 27m 192.168.8.121 kmaster <none> <none>

10 查看docker和k8s使用版本

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
bash复制代码[root@kmaster ~]# docker version
Client: Docker Engine - Community
Version: 20.10.1
API version: 1.41
Go version: go1.13.15
Git commit: 831ebea
Built: Tue Dec 15 04:37:17 2020
OS/Arch: linux/amd64
Context: default
Experimental: true

Server: Docker Engine - Community
Engine:
Version: 20.10.1
API version: 1.41 (minimum version 1.12)
Go version: go1.13.15
Git commit: f001486
Built: Tue Dec 15 04:35:42 2020
OS/Arch: linux/amd64
Experimental: false
containerd:
Version: 1.4.3
GitCommit: 269548fa27e0089a8b8278fc4fc781d7f65a939b
runc:
Version: 1.0.0-rc92
GitCommit: ff819c7e9184c13b7c2607fe6c30ae19403a7aff
docker-init:
Version: 0.19.0
GitCommit: de40ad0

[root@kmaster ~]# kubectl version
Client Version: version.Info{Major:"1", Minor:"20", GitVersion:"v1.20.1", GitCommit:"c4d752765b3bbac2237bf87cf0b1c2e307844666", GitTreeState:"clean", BuildDate:"2020-12-18T12:09:25Z", GoVersion:"go1.15.5", Compiler:"gc", Platform:"linux/amd64"}
Server Version: version.Info{Major:"1", Minor:"20", GitVersion:"v1.20.1", GitCommit:"c4d752765b3bbac2237bf87cf0b1c2e307844666", GitTreeState:"clean", BuildDate:"2020-12-18T12:00:47Z", GoVersion:"go1.15.5", Compiler:"gc", Platform:"linux/amd64"}

注意:

使用docker版本20.10.1、Go版本1.13.15,
使用k8s版本1.20.1、Go版本1.15.5

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

mysql数据库事务,MVCC原理详解

发表于 2021-01-09

大家好,我是java小杰要加油,
今天来分享一个京东面试真题,也是这是我前阵子听我旁边高T(高,实在是高)面试候选人的时候问的一个问题,他问,你能说说 mysql的事务吗? MVCC有了解吗?

  • 话不多说,直接开干

事务定义及四大特性

  • 事务是什么?

就是用户定义的一系列数据库操作,这些操作可以视为一个完成的逻辑处理工作单元,要么全部执行,要么全部不执行,是不可分割的工作单元。

  • 事务的四大特性(简称ACID):
+ **原子性(Atomicity)**:一个事务是一个**不可分割**的工作单位,事务中包括的操作要么都做,要么都不做。
+ **一致性(Consistency)**:事务必须是使数据库**从一个一致性状态变到另一个一致性状态**。一致性与原子性是密切相关的。
+ **隔离性(Isolation)**:一个事务的执行不能被其他事务干扰。即一个事务内部的操作及使用的数据对并发的其他事务是隔离的,并发执行的各个事务之间**不能互相干扰**.
+ **持久性(Durability)**:指一个事务一旦提交,它对数据库中数据的**改变就应该是永久性的**,接下来的其他操作或故障不应该对其有任何影响。

事务中常见问题

  • 脏读(dirty read):就是一个A事务即便没有提交,它对数据的修改也可以被其他事务B事务看到,B事务读到了A事务还未提交的数据,这个数据有可能是错的,有可能A不想提交这个数据,这只是A事务修改数据过程中的一个中间数据,但是被B事务读到了,这种行为被称作脏读,这个数据被称为脏数据
  • 不可重复读(non-repeatable read):在A事务内,多次读取同一个数据,但是读取的过程中,B事务对这个数据进行了修改,导致此数据变化了,那么A事务再次读取的时候,数据就和第一次读取的时候不一样了,这就叫做不可重复读
  • 幻读(phantom read):A事务多次查询数据库,结果发现查询的数据条数不一样,A事务多次查询的间隔中,B事务又写入了一些符合查询条件的多条数据(这里的写入可以是update,insert,delete),A事务再查的话,就像发生了幻觉一样,怎么突然改变了这么多,这种现象这就叫做幻读

隔离级别——产生问题的原因

多个事务互相影响,并没有隔离好,就是我们刚才提到的事务的四大特性中的 隔离性(Isolation) 出现了问题 事务的隔离级别并没有设置好,下面我们来看下事务究竟有哪几种隔离级别

  • 隔离级别
    • 读未提交(read uncommitted RU): 一个事务还没提交时,它做的变更就能被别的事务看到
    • 读提交(read committed RC): 一个事务提交之后,它做的变更才会被其他事务看到。
    • 可重复读(repeatable read RR): 一个事务执行过程中看到的数据,总是跟这个事务在启动时看到的数据是一致的。当然在可重复读隔离级别下,未提交变更对其他事务也是不可见的。
    • 串行化(serializable ): 顾名思义是对于同一行记录,“写”会加“写锁”,“读”会加“读锁”。当出现读写锁冲突的时候,后访问的事务必须等前一个事务执行完成,才能继续执行。

我们来看个例子,更加直观的了解这四种隔离级别和上述问题脏读,不可重复读,幻读的关系

下面我们讨论下当事务处于不同隔离级别情况时,V1,V2,V3分别是什么不同的值吧

  • 读未提交 (RU): A事务可以读取到B事务修改的值,即便B事务没有提交。所以V1就是200
+ V1 : 200
+ V2 : 200
+ V3 : 200
  • 读提交(RC): 当B事务没有提交的时候,A事务不可以看到B事务修改的值,只有提交以后才可以看到
+ V1 : 100
+ V2 : 200
+ V3 : 200
  • 可重复读(RR): A事务多次读取数据,数据总和第一次读取的一样,
+ V1 : 100
+ V2 : 100
+ V3 : 200
  • 串行化(S): 事务A在执行的时候,事务B会被锁住,等事务A执行结束后,事务B才可以继续执行
+ V1 : 100
+ V2 : 100
+ V3 : 200

MVCC原理

MVCC(Multi-Version Concurrency Control)多版本并发控制,是数据库控制并发访问的一种手段。

  • 特别要注意MVCC只在 读已提交(RC) 和 可重复度(RR) 这两种事务隔离级别下才有效
  • 是 数据库引擎(InnoDB) 层面实现的,用来处理读写冲突的手段(不用加锁),提高访问性能

MVCC是怎么实现的呢?它靠的就是版本链和一致性视图

1. 版本链

  • 版本链是一条链表,链接的是每条数据曾经的修改记录

那么这个版本链又是如何形成的呢,每条数据又是靠什么链接起来的呢?

其实是这样的,对于InnoDB存储引擎的表来说,它的聚簇索引记录包含两个隐藏字段

  • trx_id: 存储修改此数据的事务id,只有这个事务操作了某些表的数据后当更改操作发生的时候(update,delete,insert),才会分配唯一的事务id,并且此事务id是递增的
  • roll_pointer: 指针,指向上一次修改的记录
  • row_id(非必须): 当有主键或者有不允许为null的unique键时,不包含此字段

假如说当前数据库有一条这样的数据,假设是事务ID为100的事务插入的这条数据,那么此条数据的结构如下

后来,事务200,事务300,分别来修改此数据

时间T trx_id 200 trx_id 300
T1 开始事务 开始事务
T2 更改名字为A
T3 更改名字为B
T4 提交事务 更改名字为C
T6 提交事务

所以此时的版本链如下

我们每更改一次数据,就会插入一条undo日志,并且记录的roll_pointer指针会指向上一条记录,如图所示

  1. 第一条数据是小杰,事务ID为100
  2. 事务ID为200的事务将名称从小杰改为了A
  3. 事务ID为200的事务将名称从A又改为了B
  4. 事务ID为300的事务将名称从B又改为了C

所以串成的链表就是 C -> B -> A -> 小杰 (从最新的数据到最老的数据)

2. 一致性视图(ReadView)

需要判断版本链中的哪个版本是是当前事务可见的,因此有了一致性视图的概念。其中有四个属性比较重要

  • m_ids: 在生成ReadView时,当前活跃的读写事务的事务id列表
  • min_trx_id: m_ids的最小值
  • max_trx_id: m_ids的最大值+1
  • creator_trx_id: 生成该事务的事务id,单纯开启事务是没有事务id的,默认为0,creator_trx_id是0。

版本链中的当前版本是否可以被当前事务可见的要根据这四个属性按照以下几种情况来判断

  • 当 trx_id = creator_trx_id 时:当前事务可以看见自己所修改的数据, 可见,
  • 当 trx_id < min_trx_id 时 : 生成此数据的事务已经在生成readView前提交了, 可见
  • 当 trx_id >= max_trx_id 时 :表明生成该数据的事务是在生成ReadView后才开启的, 不可见
  • 当 min_trx_id <= trx_id < max_trx_id 时
    • trx_id 在 m_ids 列表里面 :生成ReadView时,活跃事务还未提交,不可见
    • trx_id 不在 m_ids 列表里面 :事务在生成readView前已经提交了,可见

如果某个版本数据对当前事务不可见,那么则要顺着版本链继续向前寻找下个版本,继续这样判断,以此类推。

注:RR和RC生成一致性视图的时机不一样 (这也是两种隔离级别实现的主要区别)

  • 读提交(read committed RC) 是在每一次select的时候生成ReadView的
  • 可重复读(repeatable read RR)是在第一次select的时候生成ReadView的

下面咱们一起来举个例子实战一下。

RR与RC和MVCC的例子实战

假如说,我们有多个事务如下执行,我们通过这个例子来分析当数据库隔离级别为RC和RR的情况下,当时读数据的一致性视图和版本链,也就是MVCC,分别是怎么样的。

  • 假设数据库中有一条初始数据 姓名是java小杰要加油,id是1 (id,姓名,trx_id,roll_point),插入此数据的事务id是1
  • 尤其要指出的是,只有这个事务操作了某些表的数据后当更改操作发生的时候(update,delete,insert),才会分配唯一的事务id,并且此事务id是递增的,单纯开启事务是没有事务id的,默认为0,creator_trx_id是0。
  • 以下例子中的A,B,C的意思是将姓名更改为A,B,C 读也是读取当前时刻的姓名,默认全都开启事务,并且此事务都经历过某些操作产生了事务id
时间 事务100 事务200 事务300 事务400
T1 A
T2 B
T3 C
T4 读
T5 提交
T6 D
T7 读
T8 E
T9 提交
T10 读

读已提交(RC)与MVCC

  • 一个事务提交之后,它做的变更才会被其他事务看到

每次读的时候,ReadView(一致性视图)都会重新生成

  1. 当T1时刻时,事务100修改名字为A
  2. 当T2时刻时,事务100修改名字为B
  3. 当T3时刻时,事务200修改名字为C
  4. 当T4时刻时,事务300开始读取名字
  • 此时这条数据的版本链如下

同颜色代表是同一事务内的操作

  • 来我们静下心来好好分析一下此时T4时刻事务300要读了,究竟会读到什么数据?

当前最近的一条数据是,C,事务200修改的,还记得我们前文说的一致性视图的几个属性吗,和按照什么规则判断这个数据能不能被当前事务读。我们就分析这个例子。

此时 (生成一致性视图ReadView)

  • m_ids 是[100,200]: 当前活跃的读写事务的事务id列表
  • min_trx_id 是 100: m_ids的最小值
  • max_trx_id 是 201: m_ids的最大值+1

当前数据的trx_id(事务id)是 200,符合min_trx_id<=trx_id<max_trx_id 此时需要判断
trx_id 是否在m_ids活跃事务列表里面,一看,活跃事务列表里面是【100,200】,只有两个事务活跃,而此时的trx_id是200,则trx_id在活跃事务列表里面,活跃事务列表代表还未提交的事务,所以该版本数据不可见,就要根据roll_point指针指向上一个版本,继续这样的判断,上一个版本事务id是100,数据是B,发现100也在活跃事务列表里面,所以不可见,继续找到上个版本,事务是100,数据是A,发现是同样的情况,继续找到上个版本,发现事务是1,数据是小杰,1小于100,trx_id<min_trx_id,代表生成这个数据的事务已经在生成ReadView前提交了,此数据可以被读到。所以读取的数据就是小杰

分析完第一个读,我们继续向下分析

  1. 当T5时刻时,事务100提交
  2. 当T6时刻时,事务300将名字改为D
  3. 当T7时刻时,事务400读取当前数据
  • 此时这条数据的版本链如下

此时 (重新生成一致性视图ReadView)

  • m_ids 是[200,300]: 当前活跃的读写事务的事务id列表
  • min_trx_id 是 200: m_ids的最小值
  • max_trx_id 是 301: m_ids的最大值+1

当前数据事务id是300,数据为D,符合min_trx_id<=trx_id<max_trx_id 此时需要判断数据是否在活跃事务列表里,300在这里面,所以就是还未提交的事务就是不可见,所以就去查看上个版本的数据,上个版本事务id是200,数据是C,也在活跃事务列表里面,也不可见,继续向上个版本找,上个版本事务id是100,数据是B,100小于min_trx_id,就代表,代表生成这个数据的事务已经在生成ReadView前提交了,此数据可见,所以读取出来的数据就是B

分析完第二个读,我们继续向下分析

  1. 当T8时刻时,事务200将名字改为E
  2. 当T9时刻时,事务200提交
  3. 当T10时刻时,事务300读取当前数据
  • 此时这条数据的版本链如下

此时 (重新生成一致性视图ReadView)

  • m_ids 是[300]: 当前活跃的读写事务的事务id列表
  • min_trx_id 是 300: m_ids的最小值
  • max_trx_id 是 301: m_ids的最大值+1

当前事务id是200,200<min_trx_id ,代表生成这个数据的事务已经在生成ReadView前提交了,此数据可见,所以读出的数据就是E.

当隔离级别是读已提交RC的情况下,每次读都会重新生成 一致性视图(ReadView)

  • T4时刻 事务300读取到的数据是小杰
  • T7时刻 事务400读取到的数据是B
  • T10时刻 事务300读取到的数据是E

可重复读(RR)与MVCC

  • 一个事务执行过程中看到的数据,总是跟这个事务在启动时看到的数据是一致的

所以对于事务300来讲,它分别在T4和T10的时候,读取数据,但是它的一致性视图,用的永远都是第一次读取时的视图,就是T3时刻产生的一致性视图

RR和RC的版本链是一样的,但是判断当前数据可见与否用到的一致性视图不一样

在此可重复读RR隔离级别下,

  1. T4时刻时事务300第一次读时的分析和结果与RC都一样,可以见上文分析与结果
  2. T7时刻时事务400第一次读时的分析和结果与RC都一样,可以见上文分析与结果
  3. T10时刻时事务300第二次读时的一致性视图和第一次读时的一样,所以此时到底读取到什么数据就要重新分析了

此时 (用的是第一次读时生成的一致性视图ReadView)

  • m_ids 是[100,200]: 当前活跃的读写事务的事务id列表
  • min_trx_id 是 100: m_ids的最小值
  • max_trx_id 是 201: m_ids的最大值+1

此时的版本链是

当前数据的事务id是200,数据是E,在当前事务活跃列表里面,所以数据不可见,根据回滚指针找到上个版本,发现事务id是300,当前事务也是300,可见,所以读取的数据是D

  • 我们可以自己思考下,要是没有事务300这条更改的这条记录,又该怎么继续向下分析呢?

当隔离级别是可重复读RR的情况下,每次读都会用第一次读取数据时生成的一致性视图(ReadView)

  • T4时刻 事务300读取到的数据是小杰
  • T7时刻 事务400读取到的数据是B
  • T10时刻 事务300读取到的数据是D

往期精彩推荐

  • 京东这道面试题你会吗?
  • ?线程池为什么可以复用,我是蒙圈了。。。
  • 学会了volatile,你变心了,我看到了
  • mysql可以靠索引,而我只能靠打工,加油,打工人!
  • 你好,我叫AQS(系列一:加锁)

絮絮叨叨

如果大家觉得这篇文章对自己有一点点帮助的话,欢迎关注此公众号 java小杰要加油,

  • 随手一个点击,我将无法忘记,这是我坚持创作的最大动力!
  • 非常欢迎 各位号主读者一起交流学习,互相开白转发,网络一线牵,珍惜这段缘

若文章有误欢迎指出,靓仔靓女们,我们下篇文章见,扫一扫,开启我们的故事

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

Serverless 在 SaaS 领域的最佳实践 SaaS

发表于 2021-01-08

头图.png

作者 | 计缘
来源|阿里巴巴云原生公众号

随着互联网人口红利逐渐减弱,基于流量的增长已经放缓,互联网行业迫切需要找到一片足以承载自身持续增长的新蓝海,产业互联网正是这一宏大背景下的新趋势。我们看到互联网浪潮正在席卷传统行业,云计算、大数据、人工智能开始大规模融入到金融、制造、物流、零售、文娱、教育、医疗等行业的生产环节中,这种融合称为产业互联网。而在产业互联网中,有一块不可小觑的领域是 SaaS 领域,它是 ToB 赛道的中间力量,比如 CRM、HRM、费控系统、财务系统、协同办公等等。

SaaS 系统面临的挑战

在消费互联网时代,大家是搜索想要的东西,各个厂商在云计算、大数据、人工智能等技术基座之上建立流量最大化的服务与生态,基于海量内容分发与流量共享为逻辑构建系统。而到了产业互联网时代,供给关系发生了变化,大家是定制想要的东西,需要从供给与需求两侧出发进行双向建设,这个时候系统的灵活性和扩展性面临着前所未有的挑战,尤其是 ToB 的 SaaS 领域。

1.png

特别是对于当下的经济环境,SaaS 厂商要明白,不能再通过烧钱的方式,只关注在自己的用户数量上,而更多的要思考如何帮助客户降低成本、增加效率,所以需要将更多的精力放在自己产品的定制化能力上。

2.png

如何应对挑战

SaaS 领域中的佼佼者 Salesforce,将 CRM 的概念扩展到 Marketing、Sales、Service,而这三块领域中只有 Sales 有专门的 SaaS 产品,其他两个领域都是各个 ISV 在不同行业的行业解决方案,靠的是什么?毋庸置疑,是 Salesforce 强大的 aPaaS 平台。ISV、内部实施、客户均可以在各自维度通过 aPaaS 平台构建自己行业、自己领域的 SaaS 系统,建立完整的生态。所以在我看来,现在的 Salesforce 已经由一家 SaaS 公司升华为一家 aPaaS 平台公司了。这种演进的过程也印证了消费互联网和产业互联网的转换逻辑以及后者的核心诉求。

然而不是所有 SaaS 公司都有财力和时间去孵化和打磨自己的 aPaaS 平台,但市场的变化、用户的诉求是实实在在存在的。若要生存,就要求变。这个变的核心就是能够让自己目前的 SaaS 系统变得灵活起来,相对建设困难的 aPaaS 平台,我们其实可以选择轻量且有效的 Serverless 方案来提升现有系统的灵活性和可扩展性,从而实现用户不同的定制需求。

Serverless 工作流

在上一篇文章《资源成本双优化!看 Serverless 颠覆编程教育的创新实践》中,已经对 Serverless 的概念做过阐述了,并且也介绍了 Serverless 函数计算(FC)的概念和实践。这篇文章中介绍一下构建系统灵活性的核心要素服务编排—— Serverless 工作流。

Serverless 工作流是一个用来协调多个分布式任务执行的全托管云服务。在 Serverless工作流中,可以用顺序、分支、并行等方式来编排分布式任务,Serverless 工作流会按照设定好的步骤可靠地协调任务执行,跟踪每个任务的状态转换,并在必要时执行您定义的重试逻辑,以确保工作流顺利完成。Serverless 工作流通过提供日志记录和审计来监视工作流的执行,可以轻松地诊断和调试应用。

下面这张图描述了 Serverless 工作流如何协调分布式任务,这些任务可以是函数、已集成云服务 API、运行在虚拟机或容器上的程序。

3.png

看完 Serverless 工作流的介绍,大家可能已经多少有点思路了吧。系统灵活性和可扩展性的核心是服务可编排,无论是以前的 BPM 还是现在的 aPaaS。所以基于 Serverless 工作流重构 SaaS 系统灵活性方案的核心思路,是将系统内用户最希望定制的功能进行梳理、拆分、抽离,再配合函数计算(FC)提供无状态的能力,通过 Serverless 工作流进行这些功能点的编排,从而实现不同的业务流程。

通过函数计算 FC 和 Serverless 工作流搭建灵活的订餐模块

订餐场景相信大家都不会陌生,在家叫外卖或者在餐馆点餐,都涉及到这个场景。当下也有很多提供点餐系统的 SaaS 服务厂商,有很多不错的 SaaS 点餐系统。随着消费互联网向产业互联网转换,这些 SaaS 点餐系统面临的定制化的需求也越来越多,其中有一个需求是不同的商家在支付时会显示不同的支付方式,比如从 A 商家点餐后付款时显示支付宝、微信支付、银联支付,从 B 商家点餐后付款时显示支付宝、京东支付。突然美团又冒出来了美团支付,此时 B 商家接了美团支付,那么从 B 商家点餐后付款时显示支付宝、京东支付、美团支付。诸如此类的定制化需求越来越多,这些 SaaS 产品如果没有 PaaS 平台,那么就会疲于不断的通过硬代码增加条件判断来实现不同商家的需求,这显然不是一个可持续发展的模式。

那么我们来看看通过函数计算 FC 和 Serverless 工作流如何优雅的解决这个问题。先来看看这个点餐流程:

4.png

  1. 通过 Serverless 工作流创建流程

首选我需要将上面用户侧的流程转变为程序侧的流程,此时就需要使用 Serverless 工作流来担任此任务了。

打开 Serverless 控制台,创建订餐流程,这里 Serverless 工作流使用流程定义语言 FDL 创建工作流,如何使用 FDL 创建工作流请参阅文档。流程图如下图所示:

5.png

FDL 代码为:

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
less复制代码version: v1beta1
type: flow
timeoutSeconds: 3600
steps:
- type: task
name: generateInfo
timeoutSeconds: 300
resourceArn: acs:mns:::/topics/generateInfo-fnf-demo-jiyuan/messages
pattern: waitForCallback
inputMappings:
- target: taskToken
source: $context.task.token
- target: products
source: $input.products
- target: supplier
source: $input.supplier
- target: address
source: $input.address
- target: orderNum
source: $input.orderNum
- target: type
source: $context.step.name
outputMappings:
- target: paymentcombination
source: $local.paymentcombination
- target: orderNum
source: $local.orderNum
serviceParams:
MessageBody: $
Priority: 1
catch:
- errors:
- FnF.TaskTimeout
goto: orderCanceled
-type: task
name: payment
timeoutSeconds: 300
resourceArn: acs:mns:::/topics/payment-fnf-demo-jiyuan/messages
pattern: waitForCallback
inputMappings:
- target: taskToken
source: $context.task.token
- target: orderNum
source: $local.orderNum
- target: paymentcombination
source: $local.paymentcombination
- target: type
source: $context.step.name
outputMappings:
- target: paymentMethod
source: $local.paymentMethod
- target: orderNum
source: $local.orderNum
- target: price
source: $local.price
- target: taskToken
source: $input.taskToken
serviceParams:
MessageBody: $
Priority: 1
catch:
- errors:
- FnF.TaskTimeout
goto: orderCanceled
- type: choice
name: paymentCombination
inputMappings:
- target: orderNum
source: $local.orderNum
- target: paymentMethod
source: $local.paymentMethod
- target: price
source: $local.price
- target: taskToken
source: $local.taskToken
choices:
- condition: $.paymentMethod == "zhifubao"
steps:
- type: task
name: zhifubao
resourceArn: acs:fc:cn-hangzhou:your_account_id:services/FNFDemo-jiyuan/functions/zhifubao-fnf-demo
inputMappings:
- target: price
source: $input.price
- target: orderNum
source: $input.orderNum
- target: paymentMethod
source: $input.paymentMethod
- target: taskToken
source: $input.taskToken
- condition: $.paymentMethod == "weixin"
steps:
- type: task
name: weixin
resourceArn: acs:fc:cn-hangzhou:your_account_id:services/FNFDemo-jiyuan.LATEST/functions/weixin-fnf-demo
inputMappings:
- target: price
source: $input.price
- target: orderNum
source: $input.orderNum
- target: paymentMethod
source: $input.paymentMethod
- target: taskToken
source: $input.taskToken
- condition: $.paymentMethod == "unionpay"
steps:
- type: task
name: unionpay
resourceArn: acs:fc:cn-hangzhou:your_account_id:services/FNFDemo-jiyuan.LATEST/functions/union-fnf-demo
inputMappings:
- target: price
source: $input.price
- target: orderNum
source: $input.orderNum
- target: paymentMethod
source: $input.paymentMethod
- target: taskToken
source: $input.taskToken
default:
goto: orderCanceled
- type: task
name: orderCompleted
resourceArn: acs:fc:cn-hangzhou:your_account_id:services/FNFDemo-jiyuan.LATEST/functions/orderCompleted
end: true
- type: task
name: orderCanceled
resourceArn: acs:fc:cn-hangzhou:your_account_id:services/FNFDemo-jiyuan.LATEST/functions/cancerOrder

在解析整个流程之前,我先要说明的一点是,我们不是完全通过 Serverless 函数计算和 Serverless 工作流来搭建订餐模块,只是用它来解决灵活性的问题,所以这个示例的主体应用是 Java 编写的,然后结合了 Serverless 函数计算和 Serverless 工作流。下面我们来详细解析这个流程。

  1. 启动流程

按常理,开始点餐时流程就应该启动了,所以在这个示例中,我的设计是当我们选择完商品和商家、填完地址后启动流程:

6.png

这里我们通过 Serverless 工作流提供的 OpenAPI 来启动流程。

7.png

  • Java 启动流程

这个示例我使用 Serverless 工作流的 Java SDK,首先在 POM 文件中添加依赖:

1
2
3
4
5
6
7
8
9
10
xml复制代码<dependency>
<groupId>com.aliyun</groupId>
<artifactId>aliyun-java-sdk-core</artifactId>
<version>[4.3.2,5.0.0)</version>
</dependency>
<dependency>
<groupId>com.aliyun</groupId>
<artifactId>aliyun-java-sdk-fnf</artifactId>
<version>[1.0.0,5.0.0)</version>
</dependency>

然后创建初始化 Java SDK 的 Config 类:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
java复制代码@Configuration
public class FNFConfig {

@Bean
public IAcsClient createDefaultAcsClient(){
DefaultProfile profile = DefaultProfile.getProfile(
"cn-xxx", // 地域ID
"ak", // RAM 账号的AccessKey ID
"sk"); // RAM 账号Access Key Secret
IAcsClient client = new DefaultAcsClient(profile);
return client;
}

}

再来看 Controller 中的 startFNF 方法,该方法暴露 GET 方式的接口,传入三个参数:

  • fnfname:要启动的流程名称。
  • execuname:流程启动后的流程实例名称。
  • input:启动输入参数,比如业务参数。
1
2
3
4
5
6
7
8
9
10
less复制代码   @GetMapping("/startFNF/{fnfname}/{execuname}/{input}")
public StartExecutionResponse startFNF(@PathVariable("fnfname") String fnfName,
@PathVariable("execuname") String execuName,
@PathVariable("input") String inputStr) throws ClientException {
JSONObject jsonObject = new JSONObject();
jsonObject.put("fnfname", fnfName);
jsonObject.put("execuname", execuName);
jsonObject.put("input", inputStr);
return fnfService.startFNF(jsonObject);
}

再来看 Service 中的 startFNF 方法,该方法分两部分,第一个部分是启动流程,第二部分是创建订单对象,并模拟入库(示例中是放在 Map 里了):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
vbscript复制代码    @Override
public StartExecutionResponse startFNF(JSONObject jsonObject) throws ClientException {
StartExecutionRequest request = new StartExecutionRequest();
String orderNum = jsonObject.getString("execuname");
request.setFlowName(jsonObject.getString("fnfname"));
request.setExecutionName(orderNum);
request.setInput(jsonObject.getString("input"));

JSONObject inputObj = jsonObject.getJSONObject("input");
Order order = new Order();
order.setOrderNum(orderNum);
order.setAddress(inputObj.getString("address"));
order.setProducts(inputObj.getString("products"));
order.setSupplier(inputObj.getString("supplier"));
orderMap.put(orderNum, order);

return iAcsClient.getAcsResponse(request);
}

启动流程时,流程名称和启动流程实例的名称是需要传入的参数,这里我将每次的订单编号作为启动流程的实例名称。至于 Input,可以根据需求构造 JSON 字符串传入。这里我将商品、商家、地址、订单号构造了 JSON 字符串在流程启动时传入流程中。

另外,创建了此次订单的 Order 实例,并存在 Map 中,模拟入库,后续环节还会查询该订单实例更新订单属性。

  • VUE 选择商品/商家页面

前端我使用 VUE 搭建,当点击选择商品和商家页面中的下一步后,通过 GET 方式调用 HTTP 协议的接口/startFNF/{fnfname}/{execuname}/{input}。和上面的 Java 方法对应。

  • fnfname:要启动的流程名称。
  • execuname:随机生成 uuid,作为订单的编号,也作为启动流程实例的名称。
  • input:将商品、商家、订单号、地址构建为 JSON 字符串传入流程。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
kotlin复制代码            submitOrder(){
const orderNum = uuid.v1()
this.$axios.$get('/startFNF/OrderDemo-Jiyuan/'+orderNum+'/{\n' +
' "products": "'+this.products+'",\n' +
' "supplier": "'+this.supplier+'",\n' +
' "orderNum": "'+orderNum+'",\n' +
' "address": "'+this.address+'"\n' +
'}' ).then((response) => {
console.log(response)
if(response.message == "success"){
this.$router.push('/orderdemo/' + orderNum)
}
})
}
  1. generateInfo 节点

第一个节点 generateInfo,先来看看 FDL 的含义:

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
yaml复制代码  - type: task
name: generateInfo
timeoutSeconds: 300
resourceArn: acs:mns:::/topics/generateInfo-fnf-demo-jiyuan/messages
pattern: waitForCallback
inputMappings:
- target: taskToken
source: $context.task.token
- target: products
source: $input.products
- target: supplier
source: $input.supplier
- target: address
source: $input.address
- target: orderNum
source: $input.orderNum
- target: type
source: $context.step.name
outputMappings:
- target: paymentcombination
source: $local.paymentcombination
- target: orderNum
source: $local.orderNum
serviceParams:
MessageBody: $
Priority: 1
catch:
- errors:
- FnF.TaskTimeout
goto: orderCanceled
  • name:节点名称。
  • timeoutSeconds:超时时间。该节点等待的时长,超过时间后会跳转到 goto 分支指向的 orderCanceled 节点。
  • pattern:设置为 waitForCallback,表示需要等待确认。inputMappings:该节点入参。
+ taskToken:Serverless 工作流自动生成的 Token。
+ products:选择的商品。
+ supplier:选择的商家。
+ address:送餐地址。
+ orderNum:订单号。
  • outputMappings:该节点的出参。
+ paymentcombination:该商家支持的支付方式。
+ orderNum:订单号。
  • catch:捕获异常,跳转到其他分支。

这里 resourceArn 和 serviceParams 需要拿出来单独解释。Serverless 工作流支持与多个云服务集成,即:将其他服务作为任务步骤的执行单元。服务集成方式由 FDL 语言表达,在任务步骤中,可以使用 resourceArn 来定义集成的目标服务,使用 pattern 定义集成模式。所以可以看到在 resourceArn 中配置 acs:mns:::/topics/generateInfo-fnf-demo-jiyuan/messages 信息,即在 generateInfo 节点中集成了 MNS 消息队列服务,当 generateInfo 节点触发后会向 generateInfo-fnf-demo-jiyuanTopic 中发送一条消息。那么消息正文和参数则在 serviceParams 对象中指定。MessageBody 是消息正文,配置 $ 表示通过输入映射 inputMappings 产生消息正文。

看完第一个节点的示例,大家可以看到,在 Serverless 工作流中,节点之间的信息传递可以通过集成 MNS 发送消息来传递,也是使用比较广泛的方式之一。

  1. generateInfo-fnf-demo 函数

向 generateInfo-fnf-demo-jiyuanTopic 中发送的这条消息包含了商品信息、商家信息、地址、订单号,表示一个下订单流程的开始,既然有发消息,那么必然有接受消息进行后续处理。所以打开函数计算控制台,创建服务,在服务下创建名为 generateInfo-fnf-demo 的事件触发器函数,这里选择 Python Runtime:

8.png

创建 MNS 触发器,选择监听 generateInfo-fnf-demo-jiyuanTopic。

9.png

打开消息服务 MNS 控制台,创建 generateInfo-fnf-demo-jiyuanTopic:

10.png

做好函数的准备工作,我们来开始写代码:

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
ini复制代码# -*- coding: utf-8 -*-
import logging
import json
import time
import requests
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkfnf.request.v20190315 import ReportTaskSucceededRequest
from aliyunsdkfnf.request.v20190315 import ReportTaskFailedRequest


def handler(event, context):
# 1. 构建Serverless工作流Client
region = "cn-hangzhou"
account_id = "XXXX"
ak_id = "XXX"
ak_secret = "XXX"
fnf_client = AcsClient(
ak_id,
ak_secret,
region
)
logger = logging.getLogger()
# 2. event内的信息即接受到Topic generateInfo-fnf-demo-jiyuan中的消息内容,将其转换为Json对象
bodyJson = json.loads(event)
logger.info("products:" + bodyJson["products"])
logger.info("supplier:" + bodyJson["supplier"])
logger.info("address:" + bodyJson["address"])
logger.info("taskToken:" + bodyJson["taskToken"])
supplier = bodyJson["supplier"]
taskToken = bodyJson["taskToken"]
orderNum = bodyJson["orderNum"]
# 3. 判断什么商家使用什么样的支付方式组合,这里的示例比较简单粗暴,正常情况下,应该使用元数据配置的方式获取
paymentcombination = ""
if supplier == "haidilao":
paymentcombination = "zhifubao,weixin"
else:
paymentcombination = "zhifubao,weixin,unionpay"

# 4. 调用Java服务暴露的接口,更新订单信息,主要是更新支付方式
url = "http://xx.xx.xx.xx:8080/setPaymentCombination/" + orderNum + "/" + paymentcombination + "/0"
x = requests.get(url)

# 5. 给予generateInfo节点响应,并返回数据,这里返回了订单号和支付方式
output = "{\"orderNum\": \"%s\", \"paymentcombination\":\"%s\" " \
"}" % (orderNum, paymentcombination)
request = ReportTaskSucceededRequest.ReportTaskSucceededRequest()
request.set_Output(output)
request.set_TaskToken(taskToken)
resp = fnf_client.do_action_with_exception(request)
return 'hello world'

因为 generateInfo-fnf-demo 函数配置了 MNS 触发器,所以当 TopicgenerateInfo-fnf-demo-jiyuan 有消息后就会触发执行 generateInfo-fnf-demo 函数。

整个代码分五部分:

  • 构建 Serverless 工作流 Client。
  • event 内的信息即接受到 TopicgenerateInfo-fnf-demo-jiyuan 中的消息内容,将其转换为 Json 对象。
  • 判断什么商家使用什么样的支付方式组合,这里的示例比较简单粗暴,正常情况下,应该使用元数据配置的方式获取。比如在系统内有商家信息的配置功能,通过在界面上配置该商家支持哪些支付方式,形成元数据配置信息,提供查询接口,在这里进行查询。
  • 调用 Java 服务暴露的接口,更新订单信息,主要是更新支付方式。
  • 给予 generateInfo 节点响应,并返回数据,这里返回了订单号和支付方式。因为该节点的 pattern 是 waitForCallback,所以需要等待响应结果。
  1. payment 节点

我们再来看第二个节点 payment,先来看 FDL 代码:

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
yaml复制代码- type: task
name: payment
timeoutSeconds: 300
resourceArn: acs:mns:::/topics/payment-fnf-demo-jiyuan/messages
pattern: waitForCallback
inputMappings:
- target: taskToken
source: $context.task.token
- target: orderNum
source: $local.orderNum
- target: paymentcombination
source: $local.paymentcombination
- target: type
source: $context.step.name
outputMappings:
- target: paymentMethod
source: $local.paymentMethod
- target: orderNum
source: $local.orderNum
- target: price
source: $local.price
- target: taskToken
source: $input.taskToken
serviceParams:
MessageBody: $
Priority: 1
catch:
- errors:
- FnF.TaskTimeout
goto: orderCanceled

当流程流转到 payment 节点后,意味着用户进入了支付页面。

11.png

这时 payment 节点会向 MNS 的 Topicpayment-fnf-demo-jiyuan 发送消息,会触发 payment-fnf-demo 函数。

  1. payment-fnf-demo 函数

payment-fnf-demo 函数的创建方式和 generateInfo-fnf-demo 函数类似,这里不再累赘。我们直接来看代码:

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
ini复制代码# -*- coding: utf-8 -*-
import logging
import json
import os
import time
import logging
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkcore.client import AcsClient
from aliyunsdkfnf.request.v20190315 import ReportTaskSucceededRequest
from aliyunsdkfnf.request.v20190315 import ReportTaskFailedRequest
from mns.account import Account # pip install aliyun-mns
from mns.queue import *


def handler(event, context):
logger = logging.getLogger()
region = "xxx"
account_id = "xxx"
ak_id = "xxx"
ak_secret = "xxx"
mns_endpoint = "http://your_account_id.mns.cn-hangzhou.aliyuncs.com/"
queue_name = "payment-queue-fnf-demo"
my_account = Account(mns_endpoint, ak_id, ak_secret)
my_queue = my_account.get_queue(queue_name)
# my_queue.set_encoding(False)
fnf_client = AcsClient(
ak_id,
ak_secret,
region
)
eventJson = json.loads(event)

isLoop = True
while isLoop:
try:
recv_msg = my_queue.receive_message(30)
isLoop = False
# body = json.loads(recv_msg.message_body)
logger.info("recv_msg.message_body:======================" + recv_msg.message_body)
msgJson = json.loads(recv_msg.message_body)
my_queue.delete_message(recv_msg.receipt_handle)
# orderCode = int(time.time())
task_token = eventJson["taskToken"]
orderNum = eventJson["orderNum"]
output = "{\"orderNum\": \"%s\", \"paymentMethod\": \"%s\", \"price\": \"%s\" " \
"}" % (orderNum, msgJson["paymentMethod"], msgJson["price"])
request = ReportTaskSucceededRequest.ReportTaskSucceededRequest()
request.set_Output(output)
request.set_TaskToken(task_token)
resp = fnf_client.do_action_with_exception(request)
except Exception as e:
logger.info("new loop")
return 'hello world'

该函数的核心思路是等待用户在支付页面选择某个支付方式确认支付。所以这里使用了 MNS 的队列来模拟等待。循环等待接收队列 payment-queue-fnf-demo 中的消息,当收到消息后将订单号和用户选择的具体支付方式以及金额返回给 payment 节点。

  1. VUE 选择支付方式页面

因为经过 generateInfo 节点后,该订单的支付方式信息已经有了,所以对于用户而言,当填完商品、商家、地址后,跳转到的页面就是该确认支付页面,并且包含了该商家支持的支付方式。

12.png

当进入该页面后,会请求 Java 服务暴露的接口,获取订单信息,根据支付方式在页面上显示不同的支付方式。代码片段如下:

13.png

当用户选定某个支付方式点击提交订单按钮后,向 payment-queue-fnf-demo 队列发送消息,即通知 payment-fnf-demo 函数继续后续的逻辑。

这里我使用了一个 HTTP 触发器类型的函数,用于实现向 MNS 发消息的逻辑,paymentMethod-fnf-demo 函数代码如下。

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
ini复制代码# -*- coding: utf-8 -*-

import logging
import urllib.parse
import json
from mns.account import Account # pip install aliyun-mns
from mns.queue import *
HELLO_WORLD = b'Hello world!\n'

def handler(environ, start_response):
logger = logging.getLogger()
context = environ['fc.context']
request_uri = environ['fc.request_uri']
for k, v in environ.items():
if k.startswith('HTTP_'):
# process custom request headers
pass
try:
request_body_size = int(environ.get('CONTENT_LENGTH', 0))
except (ValueError):
request_body_size = 0
request_body = environ['wsgi.input'].read(request_body_size)
paymentMethod = urllib.parse.unquote(request_body.decode("GBK"))
logger.info(paymentMethod)
paymentMethodJson = json.loads(paymentMethod)

region = "cn-xxx"
account_id = "xxx"
ak_id = "xxx"
ak_secret = "xxx"
mns_endpoint = "http://your_account_id.mns.cn-hangzhou.aliyuncs.com/"
queue_name = "payment-queue-fnf-demo"
my_account = Account(mns_endpoint, ak_id, ak_secret)
my_queue = my_account.get_queue(queue_name)
output = "{\"paymentMethod\": \"%s\", \"price\":\"%s\" " \
"}" % (paymentMethodJson["paymentMethod"], paymentMethodJson["price"])
msg = Message(output)
my_queue.send_message(msg)

status = '200 OK'
response_headers = [('Content-type', 'text/plain')]
start_response(status, response_headers)
return [HELLO_WORLD]

该函数的逻辑很简单,就是向 MNS 的队列 payment-queue-fnf-demo 发送用户选择的支付方式和金额。
VUE代码片段如下:

14.png

  1. paymentCombination 节点

paymentCombination 节点是一个路由节点,通过判断某个参数路由到不同的节点,这里自然使用 paymentMethod 作为判断条件。FDL 代码如下:

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
yaml复制代码- type: choice
name: paymentCombination
inputMappings:
- target: orderNum
source: $local.orderNum
- target: paymentMethod
source: $local.paymentMethod
- target: price
source: $local.price
- target: taskToken
source: $local.taskToken
choices:
- condition: $.paymentMethod == "zhifubao"
steps:
- type: task
name: zhifubao
resourceArn: acs:fc:cn-hangzhou:your_account_id:services/FNFDemo-jiyuan/functions/zhifubao-fnf-demo
inputMappings:
- target: price
source: $input.price
- target: orderNum
source: $input.orderNum
- target: paymentMethod
source: $input.paymentMethod
- target: taskToken
source: $input.taskToken
- condition: $.paymentMethod == "weixin"
steps:
- type: task
name: weixin
resourceArn: acs:fc:cn-hangzhou:your_account_id:services/FNFDemo-jiyuan.LATEST/functions/weixin-fnf-demo
inputMappings:
- target: price
source: $input.price
- target: orderNum
source: $input.orderNum
- target: paymentMethod
source: $input.paymentMethod
- target: taskToken
source: $input.taskToken
- condition: $.paymentMethod == "unionpay"
steps:
- type: task
name: unionpay
resourceArn: acs:fc:cn-hangzhou:your_account_id:services/FNFDemo-jiyuan.LATEST/functions/union-fnf-demo
inputMappings:
- target: price
source: $input.price
- target: orderNum
source: $input.orderNum
- target: paymentMethod
source: $input.paymentMethod
- target: taskToken
source: $input.taskToken
default:
goto: orderCanceled

这里的流程是,用户选择支付方式后,通过消息发送给 payment-fnf-demo 函数,然后将支付方式返回,于是流转到 paymentCombination 节点通过判断支付方式流转到具体处理支付逻辑的节点和函数。

  1. zhifubao节点

我们具体来看一个 zhifubao 节点:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
yaml复制代码    choices:
- condition: $.paymentMethod == "zhifubao"
steps:
- type: task
name: zhifubao
resourceArn: acs:fc:cn-hangzhou:your_account_id:services/FNFDemo-jiyuan/functions/zhifubao-fnf-demo
inputMappings:
- target: price
source: $input.price
- target: orderNum
source: $input.orderNum
- target: paymentMethod
source: $input.paymentMethod
- target: taskToken
source: $input.taskToken

这个节点的 resourceArn 和之前两个节点的不同,这里配置的是函数计算中函数的 ARN,也就是说当流程流转到这个节点时会触发 zhifubao-fnf-demo 函数,该函数是一个事件触发函数,但不需要创建任何触发器。流程将订单金额、订单号、支付方式传给 zhifubao-fnf-demo 函数。

  1. zhifubao-fnf-demo 函数

现在我们来看 zhifubao-fnf-demo 函数的代码:

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
ini复制代码# -*- coding: utf-8 -*-
import logging
import json
import requests
import urllib.parse
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkfnf.request.v20190315 import ReportTaskSucceededRequest
from aliyunsdkfnf.request.v20190315 import ReportTaskFailedRequest


def handler(event, context):
region = "cn-xxx"
account_id = "xxx"
ak_id = "xxx"
ak_secret = "xxx"
fnf_client = AcsClient(
ak_id,
ak_secret,
region
)
logger = logging.getLogger()
logger.info(event)
bodyJson = json.loads(event)
price = bodyJson["price"]
taskToken = bodyJson["taskToken"]
orderNum = bodyJson["orderNum"]
paymentMethod = bodyJson["paymentMethod"]
logger.info("price:" + price)
newPrice = int(price) * 0.8
logger.info("newPrice:" + str(newPrice))
url = "http://xx.xx.xx.xx:8080/setPaymentCombination/" + orderNum + "/" + paymentMethod + "/" + str(newPrice)
x = requests.get(url)

return {"Status":"ok"}

示例中的代码逻辑很简单,接收到金额后,将金额打 8 折,然后将价格更新回订单。其他支付方式的节点和函数如法炮制,变更实现逻辑就可以。在这个示例中,微信支付打了 5 折,银联支付打 7 折。

  1. 完整流程

流程中的 orderCompleted 和 orderCanceled 节点没做什么逻辑,大家可以自行发挥,思路和之前的节点一样。所以完整的流程是这样:

15.png

从 Serverless 工作流中看到的节点流转是这样的:

16.png

总结

到此,我们基于 Serverless 工作流和 Serverless 函数计算构建的订单模块示例就算完成了,在示例中,有两个点需要大家注意:

  • 配置商家和支付方式的元数据规则。
  • 确认支付页面的元数据规则。

因为在实际生产中,我们需要将可定制的部分都抽象为元数据描述,需要有配置界面制定商家的支付方式即更新元数据规则,然后前端页面基于元数据信息展示相应的内容。

所以如果之后需要接入其他的支付方式,只需在 paymentCombination 路由节点中确定好路由规则,然后增加对应的支付方式函数即可。通过增加元数据配置项,就可以在页面显示新加的支付方式,并且路由到处理新支付方式的函数中。

以上内容作为抛砖引玉之石,探索 Serverless 的应用场景,来解决 SaaS 厂商灵活性和扩展性的痛点。大家如果有任何疑问也可以加入钉钉群:35712134 来寻找答案,我们不见不散!

课程推荐

为了更多开发者能够享受到 Serverless 带来的红利,这一次,我们集结了 10+ 位阿里巴巴 Serverless 领域技术专家,打造出最适合开发者入门的 Serverless 公开课,让你即学即用,轻松拥抱云计算的新范式——Serverless。

点击即可免费观看课程:developer.aliyun.com/learning/ro…

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

最全 SpringAOP 切面表达式 导言 使用指南 其他

发表于 2021-01-08

往期相关文章:

AOP核心概念和SpringAOP切面

10分钟入门SpringAOP

导言

什么是PCD

PCD(pointcut designators )就是SpringAOP的切点表达式。SpringAOP的PCD是完全兼容AspectJ的,一共有10种。

PCD一览图

image-20210108140219721
使用指南
====

SpringAOP是基于动态代理实现的,以下以目标对象表示被代理bean,代理对象表示AOP构建出来的bean。目标方法表示被代理的方法。

execution

execution是最常用的PCD。它的匹配式模板如下展示:

1
2
3
java复制代码execution(modifiers-pattern? ret-type-pattern declaring-type-pattern? name-pattern(param-pattern)
throws-pattern?)
execution(修饰符匹配式? 返回类型匹配式 类名匹配式? 方法名匹配式(参数匹配式) 异常匹配式?)

代码块中带?符号的匹配式都是可选的,对于execution PCD必不可少的只有三个:

  1. 返回值匹配值
  2. 方法名匹配式
  3. 参数匹配式

举例分析: execution(public * ServiceDemo.*(..)) 匹配public修饰符,返回值是*,即任意返回值类型都行,ServiceDemo是类名匹配式不一定要全路径,只要全局依可见性唯一就行,.*是方法名匹配式,匹配所有方法,..是参数匹配式,匹配任意数量、任意类型参数的方法。

再举一些其他例子:

  • execution(* com.xyz.service..*.*(..)): 匹配com.xyz.service及其子包下的任意方法。
  • execution(* joke(Object+))):匹配任意名字为joke的方法,且其动态入参是是Object类型或该类的子类。
  • execution(* joke(String,..)):匹配任意名字为joke的方法,该方法 一个入参为String(不可以为子类),后面可以有任意个入参且入参类型不限
  • execution(* com..*.*Dao.find*(..)): 匹配指定包下find开头的方法
  • execution(* com.baobaotao.Waiter+.*(..)) : 匹配com.baobaotao包下Waiter及其子类的所有方法。

within

筛选出某包下的所有类,注意要带有*。

  • within(com.xyz.service.*)com.xyz.service包下的类,不包括子包
  • within(com.xyz.service..*)com.xyz.service包下及其子包下的类

this

常用于命名绑定模式。对由代理对象的类型进行过滤筛选。

如果目标类是基于接口实现的,则this()中可以填该接口的全路径名,否则非接口实现由于是基于CGLIB实现的,this中可以填写目标类的全路径名。

this(com.xyz.service.AccountService): 代理类是com.xyz.service.AccountService或其子类。

使用@EnableAspectJAutoProxy(proxyTargetClass = true)可以强制使用CGLIB。否则默认首先使用jdk动态代理,jdk代理不了才会用CGLIB。

target

this作用于代理对象,target作用于目标对象。

target(com.xyz.service.AccountService): 被代理类(目标对象)是com.xyz.service.AccountService或其子类

args

常用于对目标方法的参数匹配。一般不单独使用,而是配合其他PCD来使用。args可以使用命名绑定模式,如下举例:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
java复制代码@Aspect // 切面声明
@Component // 注入IOC
@Slf4j
class AspectDemo {
@Around("within(per.aop.*) && args(str)") // 在per.aop包下,且被代理方法的只有一个参数,参数类型是String或者其子类
@SneakyThrows
public Object logAspect(ProceedingJoinPoint pjp, String str) {
String signature = pjp.getSignature().toString();
log.info("{} start,param={}", signature, pjp.getArgs());
Object res = pjp.proceed();
log.info("{} end", signature);
return res;
}
}
  1. 如果args中是参数名,则配合切面(advice)方法的使用来确定要匹配的方法参数类型。
  2. 如果args中是类型,例如@Around("within(per.aop.*) && args(String)”),则可以不必使用切面方法来确定类型,但此时也不能使用参数绑定了见下文了。

虽然args()支持+符号,但本省args()就支持子类通配。

和带参数匹配execution区别

举个例子: args(com.xgj.Waiter)等价于 execution(* *(com.xgj.Waiter+))。而且execution不能支持带参数的advice。

@target

使用场景举例: 当一个Service有多个子类时, 某些子类需要打日志,某些子类不需要打日志时可以如下处理(配合java多态):

筛选出具有给定注解的被代理对象是对象不是类,@target是动态的。如下自定义一个注解LogAble:

1
2
3
4
5
java复制代码//全限定名: annotation.LogAble
@Target({ElementType.TYPE,ElementType.PARAMETER}) // 支持在方法参数、类上注
@Retention(RetentionPolicy.RUNTIME)
public @interface LogAble {
}

假如需要“注上了这个注解的所有类的的public方法“都打日志的话日志逻辑要自定义,可以如下这么写PCD,当然对应方法的bean要注入到SpringIOC容器中:

1
2
java复制代码@Around("@target(annotation.LogAble) && execution(public * *.*(..))")
// 自定义日志逻辑

@args

对于目标方法参数的运行时类型要有@args指定的注解。是方法参数的类型上有指定注解,不是方法参数上带注解。

使用场景: 假如参数类型有多个子类,只有某个子类才可以匹配该PCD。

  • @args(com.ms.aop.jargs.demo1.Anno1): 匹配1个参数,且第1个参数运行时需要有Anno1注解
  • @args(com.ms.aop.jargs.demo1.Anno1,..)匹配一个或多个参数,第一个参数运行时需要带上Anno1注解。
  • @args(com.ms.aop.jargs.demo1.Anno1,com.ms.aop.jargs.demo1.Anno2): 一参匹配Anno1,二参匹配Annno2 。

@within

非运行时类型的的@target。@target关注的是被调用的对象,@within关注的是调用的方法所在的类。

@target 和 @within 的不同点:

@target(注解A):判断被调用的目标对象中是否声明了注解A,如果有,会被拦截

@within(注解A): 判断被调用的方法所属的类中是否声明了注解A,如果有,会被拦截

@annotation

匹配有指定注解的方法(注解作用在方法上面)

bean

根据beanNam来匹配。支持*通配符。

1
java复制代码bean(*Service) // 匹配所有Service结尾的Service

其他

组合使用

PCD之间支持,&& || !三种运算符。上文示例中就使用了&& 运算符。||表示或(不是短路或)。!表示非。

命名绑定模式

上文中的 @Around("within(per.aop.*) && args(str)")示例就是使用了命名绑定模式,在PCD中写上变量名,在方法上对变量名的类型进行限定。

1
2
java复制代码@Around("within(per.aop.*) && args(str)")
public Object logAspect(ProceedingJoinPoint pjp, String str) { ...}

如上举例,str要是String类型或其子类,且方法入参只能有一个。

name binding only allowed in target, this, and args pcds

命名绑定模式只支持target、this、args三种PCD。

argNames

观察源码可以发现,所有的Advice注解都带有argNames字段,例如@Around:

1
2
3
4
5
6
java复制代码@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
public @interface Around {
String value();
String argNames() default "";
}

什么情况下会使用这个属性呢,如下举例解释:

1
2
3
4
5
6
java复制代码@Around(value = "execution(* TestBean.paramArgs(..))  && args(decimal,str,..)&& target(bean)", argNames = "pjp,str,decimal,bean")
@SneakyThrows // proceed会抛受检异常
Object aroundArgs(ProceedingJoinPoint pjp,/*使用命名绑定模式*/ String str, BigDecimal decimal, Object bean) {
// 在方法执行前做一些操作
return pjp.proceed();
}

argnames 必须要和args、target、this标签一起使用。虽然实际操作中可以不带,但官方建议所有带参数的都带,原因如下:

因此如果‘ argernames’属性没有指定,那么 Spring AOP 将查看类的调试信息,并尝试从局部变量表中确定参数名。只要使用调试信息(至少是‘-g: vars’)编译了类,就会出现此信息。使用这个标志编译的结果是:

(1)你的代码将会更容易被反向工程)

(2)类文件大小将会非常大(通常是无关紧要的)

(3)删除未使用的局部变量的优化将不会被编译器应用。

此外,如果编译的代码没有必要的调试信息,那么 Spring AOP 将尝试推断绑定变量与参数的配对。如果变量的绑定在可用信息下是不明确的,那么一个 AmbiguousBindingException 就会被抛出。如果上面的策略都失败了,那么就会抛出一个 IllegalArgumentException。

建议所有的advice注解里都带argNames,反正idea也会提醒。

参考文章

  1. Spring文档
  2. SpringAOP Point表达式

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

SpringBoot集成WebSocket,实现后台向前端推

发表于 2021-01-08

前言

在一次项目开发中,使用到了Netty网络应用框架,以及MQTT进行消息数据的收发,这其中需要后台来将获取到的消息主动推送给前端,于是就使用到了MQTT,特此记录一下。

一、什么是websocket?

为什么不使用HTTP 协议呢?这是因为HTTP是单工通信,通信只能由客户端发起,客户端请求一下,服务器处理一下,这就太麻烦了。于是websocket应运而生。

下面我们就直接开始使用Springboot开始整合。以下案例都在我自己的电脑上测试成功,你可以根据自己的功能进行修改即可。

我的项目结构如下:

二、使用步骤

1.添加依赖

1
2
3
4
xml复制代码<dependency>  
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

2.启用Springboot对WebSocket的支持

启用WebSocket的支持也是很简单,几句代码搞定:

1
2
3
4
5
6
7
8
9
10
11
12
13
java复制代码import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
/**
* @ Description: 开启WebSocket支持
*/
@Configuration
public class WebSocketConfig {
@Bean
public ServerEndpointExporter serverEndpointExporter() {
return new ServerEndpointExporter();
}
}

3.核心配置:WebSocketServer

因为WebSocket是类似客户端服务端的形式(采用ws协议),那么这里的WebSocketServer其实就相当于一个ws协议的Controller

  • @ ServerEndpoint 注解是一个类层次的注解,它的功能主要是将目前的类定义成一个websocket服务器端, 注解的值将被用于监听用户连接的终端访问URL地址,客户端可以通过这个URL来连接到WebSocket服务器端
  • 新建一个ConcurrentHashMap webSocketMap 用于接收当前userId的WebSocket,方便传递之间对userId进行推送消息。
    下面是具体业务代码:
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
java复制代码@Component
@Slf4j
@Service
@ServerEndpoint("/api/websocket/{sid}")
public class WebSocketServer {
//静态变量,用来记录当前在线连接数。应该把它设计成线程安全的。
private static int onlineCount = 0;
//concurrent包的线程安全Set,用来存放每个客户端对应的MyWebSocket对象。
private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<WebSocketServer>();

//与某个客户端的连接会话,需要通过它来给客户端发送数据
private Session session;

//接收sid
private String sid = "";

/**
* 连接建立成功调用的方法
*/
@OnOpen
public void onOpen(Session session, @PathParam("sid") String sid) {
this.session = session;
webSocketSet.add(this); //加入set中
this.sid = sid;
addOnlineCount(); //在线数加1
try {
sendMessage("conn_success");
log.info("有新窗口开始监听:" + sid + ",当前在线人数为:" + getOnlineCount());
} catch (IOException e) {
log.error("websocket IO Exception");
}
}

/**
* 连接关闭调用的方法
*/
@OnClose
public void onClose() {
webSocketSet.remove(this); //从set中删除
subOnlineCount(); //在线数减1
//断开连接情况下,更新主板占用情况为释放
log.info("释放的sid为:"+sid);
//这里写你 释放的时候,要处理的业务
log.info("有一连接关闭!当前在线人数为" + getOnlineCount());

}

/**
* 收到客户端消息后调用的方法
* @ Param message 客户端发送过来的消息
*/
@OnMessage
public void onMessage(String message, Session session) {
log.info("收到来自窗口" + sid + "的信息:" + message);
//群发消息
for (WebSocketServer item : webSocketSet) {
try {
item.sendMessage(message);
} catch (IOException e) {
e.printStackTrace();
}
}
}

/**
* @ Param session
* @ Param error
*/
@OnError
public void onError(Session session, Throwable error) {
log.error("发生错误");
error.printStackTrace();
}

/**
* 实现服务器主动推送
*/
public void sendMessage(String message) throws IOException {
this.session.getBasicRemote().sendText(message);
}

/**
* 群发自定义消息
*/
public static void sendInfo(String message, @PathParam("sid") String sid) throws IOException {
log.info("推送消息到窗口" + sid + ",推送内容:" + message);

for (WebSocketServer item : webSocketSet) {
try {
//这里可以设定只推送给这个sid的,为null则全部推送
if (sid == null) {
// item.sendMessage(message);
} else if (item.sid.equals(sid)) {
item.sendMessage(message);
}
} catch (IOException e) {
continue;
}
}
}

public static synchronized int getOnlineCount() {
return onlineCount;
}

public static synchronized void addOnlineCount() {
WebSocketServer.onlineCount++;
}

public static synchronized void subOnlineCount() {
WebSocketServer.onlineCount--;
}

public static CopyOnWriteArraySet<WebSocketServer> getWebSocketSet() {
return webSocketSet;
}
}

4.测试Controller

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
java复制代码@Controller("web_Scoket_system")
@RequestMapping("/api/socket")
public class SystemController {
//页面请求
@GetMapping("/index/{userId}")
public ModelAndView socket(@PathVariable String userId) {
ModelAndView mav = new ModelAndView("/socket1");
mav.addObject("userId", userId);
return mav;
}

//推送数据接口
@ResponseBody
@RequestMapping("/socket/push/{cid}")
public Map pushToWeb(@PathVariable String cid, String message) {
Map<String,Object> result = new HashMap<>();
try {
WebSocketServer.sendInfo(message, cid);
result.put("code", cid);
result.put("msg", message);
} catch (IOException e) {
e.printStackTrace();
}
return result;
}
}

5.测试页面index.html

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
html复制代码<!DOCTYPE html>
<html>

<head>
<meta charset="utf-8">
<title>Java后端WebSocket的Tomcat实现</title>
<script type="text/javascript" src="js/jquery.min.js"></script>
</head>

<body>
<div id="main" style="width: 1200px;height:800px;"></div>
Welcome<br/><input id="text" type="text" />
<button onclick="send()">发送消息</button>
<hr/>
<button onclick="closeWebSocket()">关闭WebSocket连接</button>
<hr/>
<div id="message"></div>
</body>
<script type="text/javascript">
var websocket = null;
//判断当前浏览器是否支持WebSocket
if('WebSocket' in window) {
//改成你的地址
websocket = new WebSocket("ws://192.168.100.196:8082/api/websocket/100");
} else {
alert('当前浏览器 Not support websocket')
}

//连接发生错误的回调方法
websocket.onerror = function() {
setMessageInnerHTML("WebSocket连接发生错误");
};

//连接成功建立的回调方法
websocket.onopen = function() {
setMessageInnerHTML("WebSocket连接成功");
}
var U01data, Uidata, Usdata
//接收到消息的回调方法
websocket.onmessage = function(event) {
console.log(event);
setMessageInnerHTML(event);
setechart()
}

//连接关闭的回调方法
websocket.onclose = function() {
setMessageInnerHTML("WebSocket连接关闭");
}

//监听窗口关闭事件,当窗口关闭时,主动去关闭websocket连接,防止连接还没断开就关闭窗口,server端会抛异常。
window.onbeforeunload = function() {
closeWebSocket();
}

//将消息显示在网页上
function setMessageInnerHTML(innerHTML) {
document.getElementById('message').innerHTML += innerHTML + '<br/>';
}

//关闭WebSocket连接
function closeWebSocket() {
websocket.close();
}

//发送消息
function send() {
var message = document.getElementById('text').value;
websocket.send('{"msg":"' + message + '"}');
setMessageInnerHTML(message + "&#13;");
}
</script>
</html>

6.结果展示

后台,如果有连接请求

前台显示

总结

这中间我遇到一个问题,就是说WebSocket启动的时候优先于spring容器,从而导致在WebSocketServer中调用业务Service会报空指针异常,所以需要在WebSocketServer中将所需要用到的service给静态初始化一下,如图所示:

还需要做如下配置:

本文:blog.csdn.net/MacWx/artic…

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

Spring Boot 2x基础教程:多个文件的上传

发表于 2021-01-08

昨天,我们介绍了如何在Spring Boot中实现文件的上传。有读者问:那么如果有多个文件要同时上传呢?这就马上奉上,当碰到多个文件要同时上传的处理方法。

动手试试

本文的动手环节将基于Spring Boot中实现文件的上传一文的例子之上,所以读者可以拿上一篇的例子作为基础来进行改造,以体会这之间的区别,下面也主要讲解核心区别的地方。

第一步:修改文件上传页面的上传表单

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
html复制代码<!DOCTYPE html>
<html>
<head lang="en">
<meta charset="UTF-8" />
<title>文件上传页面 - didispace.com</title>
</head>
<body>
<h1>文件上传页面</h1>
<form method="post" action="/upload" enctype="multipart/form-data">
文件1:<input type="file" name="files"><br>
文件2:<input type="file" name="files"><br>
<hr>
<input type="submit" value="提交">
</form>
</body>
</html>

可以看到这里多增加一个input文件输入框,同时文件输入框的名称修改为了files,因为是多个文件,所以用了复数。注意:这几个输入框的name是一样的,这样才能在后端处理文件的时候组织到一个数组中。

第二步:修改后端处理接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
java复制代码@PostMapping("/upload")
@ResponseBody
public String create(@RequestPart MultipartFile[] files) throws IOException {
StringBuffer message = new StringBuffer();

for (MultipartFile file : files) {
String fileName = file.getOriginalFilename();
String filePath = path + fileName;

File dest = new File(filePath);
Files.copy(file.getInputStream(), dest.toPath());
message.append("Upload file success : " + dest.getAbsolutePath()).append("<br>");
}
return message.toString();
}

几个重要改动:

  1. MultipartFile使用数组,参数名称files对应html页面中input的name,一定要对应。
  2. 后续处理文件的主体(for循环内)跟之前的一样,就是对MultipartFile数组通过循环遍历的方式对每个文件进行存储,然后拼接结果返回信息。

更多本系列免费教程连载「点击进入汇总目录」

测试验证

第一步:启动Spring Boot应用,访问http://localhost:8080,可以看到如下的文件上传页面。

第二步:选择2个不大于2MB的文件,点击“提交”按钮,完成上传。

如果上传成功,将显示类似下面的页面:

你可以根据打印的文件路径去查看文件是否真的上传了。

代码示例

本文的相关例子可以查看下面仓库中的chapter4-4目录:

  • Github:github.com/dyc87112/Sp…
  • Gitee:gitee.com/didispace/S…

如果您觉得本文不错,欢迎Star支持,您的关注是我坚持的动力!

欢迎关注我的公众号:程序猿DD,获得独家整理的学习资源、日常干货及福利赠送。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

Dubbo服务是怎么暴露的?看源码就知道了

发表于 2021-01-07

​ 先放一张官网的服务暴露时序图,对我们梳理源码有很大的帮助。注:不论是暴露还是导出或者是其他翻译,都是描述export的,只是翻译不同。

0.配置解析

​ 在Spring的配置文件中,Dubbo指明了DubboNamespaceHandler类作为标签解析。

​ 与服务相关的显然就是service,找到对应的ServiceBean类,进入这个类,开始服务暴露的源码分析。这个类位于Dubbo源码config模块-spring模块下的根目录。

1.开始export

​ export也是上面时序图中最开始的一个方法,从这个方法名也知道,这就是服务暴露或者叫出口最关键的方法。进入ServiceBean类,在这个类中一共有两处调用了此方法。即onApplicationEvent和afterPropertiesSet,了解过Spring Bean生命周期的朋友看到这两个方法肯定眼熟,果然,这个类实现了相关的接口:

​ 看一下onApplicationEvent方法:

​ 从它的if判断条件调用的几个方法名可以看出,如果是延迟暴露、还未暴露过且支持暴露就可以执行export方法了。这里说一下,这个isDelay方法有点迷惑,字面意思应该为是否延迟,返回ture代表延迟。但是实际意思却为返回true代表不延迟,因为这个判断条件是delay==null || delay==-1,代表没有设置延迟。所以这个方法中的export才是第一个触发的。

​ 接着进入到export方法。这个方法会跳转到ServiceConfig类,是ServiceBean的父类,也正好符合时序图。

​ 这几个if的作用就是判断是否需要暴露和延迟暴露。如果不需要暴露就返回,否则都会执行doExport方法的。进入这个方法,这个方法代码很多,前面一堆if都是检测配置信息的,关注的重点在doExportUrls方法。

​ Dubbo是支持多注册中心和多协议的,在这里就表现出来了。获取到的注册中心URL放到一个list里面。其中loadRegistries方法就是根据配置组装成相关的URL并返回,如加载注册中心地址、检查地址是否合法、添加配置信息等。咱们先关注重点,这个方法就不跟下去了,不然没完没了。至于组装后的URL可以debug自己看看,大概样子如下:

2.组装URL

​ 进入到doExportUrlsFor1Protocol方法,这个比较重要。从它的名字可以看出,它的作用是组装暴露URL。

​ 这个方法很长,主要就是创建一个map然后添加各种值,包括配置信息、提供的服务等等。由于这个方法分支非常多,官网给了各个分支含义的解释,配合源码能很好理解其意思:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
go复制代码// 获取 ArgumentConfig 列表
for (遍历 ArgumentConfig 列表) {
if (type 不为 null,也不为空串) { // 分支1
1. 通过反射获取 interfaceClass 的方法列表
for (遍历方法列表) {
1. 比对方法名,查找目标方法
2. 通过反射获取目标方法的参数类型数组 argtypes
if (index != -1) { // 分支2
1. 从 argtypes 数组中获取下标 index 处的元素 argType
2. 检测 argType 的名称与 ArgumentConfig 中的 type 属性是否一致
3. 添加 ArgumentConfig 字段信息到 map 中,或抛出异常
} else { // 分支3
1. 遍历参数类型数组 argtypes,查找 argument.type 类型的参数
2. 添加 ArgumentConfig 字段信息到 map 中
}
}
} else if (index != -1) { // 分支4
1. 添加 ArgumentConfig 字段信息到 map 中
}
}

​ 当然,如果你没有配置相关的信息,如dubbo:method,在debug源码时,压根就不会进入到这些分支里面。现在我们看一下URL长啥样:

​ 可以看到协议已经变成了dubbo,具体的服务接口也显示了出来。而map的值就存在parameters当中。

3.服务暴露

​ 依旧在doExportUrlsFor1Protocol方法里,具体的服务URL已经组装好了,接下来就是服务暴露了。先看这么一段代码:

​ 这段代码有两个关键点,已经在图中标注。**第一处是先进行本地暴露。第二处判断如果有注册中心,就会进行远程暴露。**注册中心的URL在doExportUrls中已经获取了。

​ 先看本地暴露,进入到exportLocal方法:

​ exportLocal方法比较简单,根据协议头判断是否需要暴露服务,如果需要,就创建一个新的URL

​ 我们看一下这个URL长啥样:

​ 协议变成了injvm,从这个协议名称就可以猜测到,这个在一个jvm内的协议。IP地址也从远程注册中心的IP地址变成了本机地址。

​ 本地URL组装好后,会创建一个exporter对象。这个对象是由protocol的export方法生成,我们点进这个抽象方法,会发现它有一个@Adaptive注解。这个注解修饰方法时会生成一个代理类。主要配合SPI机制使用,SPI的作用简单的说就是提供一个标准化的接口,可能有不同的实现,而这个实现类的路径我们就放在一个固定的位置,让框架去读取。同样的用法也在proxyFactory.getInvoker()中。关于SPI的解析放在最后。这个export的具体实现方法如下图:

​ 所在类为InjvmProtocol。这个实现方法就不说了,主要就是根据传入的参数进行封装,我们直接看最终的exporter:

​ 可以看到,已经找到了服务接口的实现类了。最后就是将exporter添加到exporters中,这个exporters是本地的一个集合,专门缓存exporter。

​ 接着就是远程暴露了,其实和本地暴露的目的一样,都要封装成invoker——>exporter,最后添加到exporters中,还多了一步注册。首先依旧是通过getInvoker封装成invoker。(这里说句题外话,可以根据参数的协议类型找到这些抽象方法的实现类。Dubbo命名很严谨,比如参数中,URL的协议为registry,那么其实现类就是RegistryProtocol。至于为什么要封装成invoker我们最后再分析,现在只需理解这么做是为了屏蔽细节,统一暴露)。

​ 封装成invoker后又弄了一层wrapperInvoker,点进这个类,可以发现其实就给invoker额外封装一层,可以提供更多信息以及一些工具方法,比如ServiceConfig、检测是否有效。

​ 接着主要区别在export方法当中,其实现方法在RegistryProtocol类中(因为参数wrapperInvoker的url协议为registry)。实现方法部分截图如下:

​ 这个方法主要做了如下工作:

​ 1.调用doLocalExport导出服务

​ 2.向注册中心注册

​ 3.向注册中心订阅override数据

​ 4.创建并返回DestroyableExporter

​ 首先进入到doLocalExport方法,这个方法主要就是会调用DubboProtocol的export方法,为了避免过多的代码截图把自己弄昏了,就不贴这个方法了。这个方法开头同样的,根据invoker获取URL,关键在于它调用了一个openServer。看到这个方法名应该知道是啥意思了,即打开服务。好家伙,终于要结束了么。

​ 这个方法很清晰,获取注册中心的IP和端口号、检查缓存、创建server。接着跟进源码,bind过程,主要关注Transports的bind方法。这里Dubbo也是用Adaptive注解和SPI机制,实现了拓展功能。它会根据传入的参数选择不同类型的Transport,默认是NettyTransporter。接下来就是Netty服务启动的相关过程了,以前写过相关博客,就不跟进了。

​ 接着,我们看上上张截图,有一个if会判断是否需要注册,如果需要注册就会向注册中心注册。我们接着跟踪源码,一直到如下方法:

​ 看到了Zookeeper客户端,到这里就明白了,是向Zookeeper添加信息。我们最后看一下Zookeeper里面的内容。我们打开Zookeeper客户端,查看一下服务:

​ 可以发现,已经有我们注册的服务了。最好下个可视化的Zookeeper客户端,可以进入到这些目录,可以找到Provider的IP地址。

疑问解析

  • 为什么要本地暴露?
    • 调用本地服务时,避免网络通信。
  • 为什么要封装成invoker和export?
    • 前面的源码分析中,本地和远程都经过了封装invoker和export两个步骤。export是服务暴露的最终形态,其包含invoker以及其他更多信息,比如注册中心、服务接口、实现类等等信息。下面是官网的一张截图:

    • 官网是这么说的:由于 Invoker 是 Dubbo 领域模型中非常重要的一个概念,很多设计思路都是向它靠拢、或转换为它。这个所谓的靠拢就如图中显示的那样,不管在消费者方还是服务提供方,均会出现Invoker,它代表一个可执行体,并屏蔽了内部细节。既然它这么重要,我们就看一下它是如果创建的。
      • 其是由proxyFactory.getInvoker创建而来,通过debug找到它的实现类:

    • 上面的方法在JavassistProxyFactory类中,其重写了doInvoke方法,比较简单,只是转发了invokeMethod。其中AbstractProxyInvoker是一个抽象类,实现了Invoker接口。而这个Wrapper的作用是包裹目标类,仅可通过getWrapper(Classs)创建子类。子类可以对入参Class进行解析,拿到类方法、成员变量等信息。在这里,目标类就是暴露服务的实现类。
      • 关于Wrapper的分析内容非常多,这里记录一下官网的解析:dubbo.apache.org/zh/docs/v2.…。
  • SPI是什么?
    • SPI(Service Provider Interface),其作用前面也说了,就是定义一个标准接口,这个接口的实现由用户决定。这样做的好处就是提高了框架的拓展性。但是这个接口的实现放在哪,得让框架知道。在Java SPI中,规定在META-INF/services/ 目录下,创建一个以接口全路径名命名的文件,文件中写出接口实现类的全路径名。然后Java就会去遍历加载这些实现类并创建实例。
      • 前面说了Java SPI,但是Dubbo并没有用Java规定的方法,而是自己实现了SPI机制。可以从ServiceLoader.load()方法跟踪源码看一下,Java SPI机制是遍历了所有的实现类,而不是按需加载,造成了不必要的浪费。说到Dubbo SPI,那么它的规定目录在哪?在META-INF/dubbo/internal目录下。我们从源码的该路径下找个文件看看。

​ 可以看到Dubbo SPI的配置文件内容是键值对的形式,这样就可以实现按需加载。根据key值,获取全路径名,然后加载。 如果需要自己自定义,就直接在MEATA-INF/dubbo/目录下创建配置文件即可。同样的,类似Java SPI中的ServiceLoader,Dubbo中叫ExtensionLoader。这个类的几个方法,作用很明确,也不复杂,这里就不跟踪了。其中getExtensionLoader方法,入参是需要加载的接口,这个方法会检查是否有对应类型的ExtensionLoader对象,如果没有就新建一个。createExtension方法就是根据名字获取对应的实现类,这样就实现了按需加载。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

每秒30W次的点赞业务,怎么优化?

发表于 2021-01-07

继续答星球水友提问,30WQPS 的点赞计数业务,如何设计?

可以看到,这个业务的特点是:

(1)吞吐量超高;

(2)能够接受一定数据不一致;

_画外音:_计数有微小不准确,不是大问题。

先用最朴素的思想,只考虑点赞计数,可以怎么做?

有几点是最容易想到的:

(1)肯定不能用数据库抗实时读写流量;

(2)redis 天然支持固化,可以用高可用 redis 集群来做固化存储;

(3)也可以用 MySQL 来做固化存储,redis 做缓存,读写操作都落缓存,异步线程定期刷 DB;

(4)架一层计数服务,将计数与业务逻辑解耦;

此时 MySQL 核心数据结构是:

t_count(msg_id, praise_count)

此时 redis 的 KV 设计也不难:

_key:_msg_id

_value:_praise_count

似乎很容易就搞定了:

(1)服务可以水平扩展;

(2)数据量增加时,数据库可以水平扩展;

(3)读写量增加时,缓存也可以水平扩展;

计数系统的难点,还在于业务扩展性问题,以及效率问题。

以微博为例:

(1)用户微博首页,有多条消息 list<msg_id>,这是一种扩展;

(2)同一条消息 msg_id,不止有点赞计数,还有阅读计数,转发计数,评论计数,这也是一种扩展;

假如用最朴素的方式实现,多条消息多个计数的获取伪代码如下:

// (1) 获取首页所有消息 msg_id

list<msg_id> = getHomePageMsg(uid);

// (2) 对于首页的所有消息要__拉取多个计数

for(msg_id in list<msg_id>){

//(3.1) 获取__阅读计数

getReadCount(msg_id);

//(3.2) 获取__转发计数

getForwordCount(msg_id);

//(3.3) 获取__评论计数

getCommentCount(msg_id);

//(3.4) 获取__赞计数

getPraiseCount(msg_id);

}

由于同一个 msg_id 多了几种业务计数,redis 的 key 需要带上业务 flag,升级为:

msg_id:read

msg_id:forword

msg_id:comment

msg_id:praise

用来区分共一个 msg_id 的四种不同业务计数,redis 不能支持 key 的模糊操作,必须访问四次 reids。

假设首页有 100 条消息,这个方案总结为:

(1)for 循环每一条消息,100 条消息 100 次;

(2)每条消息 4 次 RPC 获取计数接口调用;

(3)每次调用服务要访问 reids,拼装 key 获取 count;

画外音:这种方案的扩展性和效率是非常低的。

那如何进行优化呢?

首先看下数据库层面元数据扩展,常见的扩展方式是,增加列,记录更多的业务计数。

如上图所示,由一列点赞计数,扩充为四列阅读、转发、评论、点赞计数。

增加列这种业务计数扩展方式的缺点是:每次要扩充业务计数时,总是需要修改表结构,增加列,很烦。

有没有不需要变更表结构的扩展方式呢?

行扩展是一种扩展性更好的方式。

表结构固化为:

t_count(msg_id, count_key, count_value)

当要扩充业务计数时,增加一行就行,不需要修改表结构。

_画外音:_很多配置业务,会使用这种方案,方便增加配置。

增加行这种业务计数扩展方式的缺点是:表数据行数会增加,但这不是主要矛盾,数据库水平扩展能很轻松解决数据量大的问题。

接下来看下 redis 批量获取计数的优化方案。

原始方案,通过拼装 key 来区分同一个 msg_id 的不同业务计数。

可以升级为,同一个 value 来存储多个计数。

如上图所示,同一个 msg_id 的四个计数,存储在一个 value 里,从而避免多次 redis 访问。

_画外音:_通过 value 来扩展,是不是很巧妙?

总结

计数业务,在数据量大,并发量大的时候,要考虑的一些技术点:

(1)用缓存抗读写;

(2)服务化,计数系统与业务系统解耦;

(3)水平切分扩展吞吐量、数据量、读写量;

(4)要考虑扩展性,数据库层面常见的优化有:列扩展,行扩展两种方式;

(5)要考虑批量操作,缓存层面常见的优化有:一个 value 存储多个业务计数;

计数系统优化先聊到这里,希望大家有收获。

欢迎大家继续提问,有问必答。

调研:

大伙是用 redis 搞计数么?

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

1…740741742…956

开发者博客

9558 日志
1953 标签
RSS
© 2025 开发者博客
本站总访问量次
由 Hexo 强力驱动
|
主题 — NexT.Muse v5.1.4
0%