第03篇_常见分布式组件

第01章_Nginx

第一节 Nginx简介

1. 什么是Nginx？

Nginx (engine x) 是一个高性能的HTTP和反向代理服务器，支持高达 50,000 个并发连接数，此外，也提供了IMAP/POP3/SMTP等服务。

2. Nginx应用场景

1) 反向代理

• 正向代理：在客户端配置代理服务器进行指定网站访问，如在中国大陆通过香港的代理服务器访问外网。

  

• 反向代理：在服务端配置代理服务器统一接收客户端请求，然后路由到内部隐藏的真实服务器。

  

2) 负载均衡

负载均衡指将客户端请求按指定策略分配到不同服务器，以达到平衡负载的效果。Nginx默认支持轮询、加权轮询、IpHash等负载均衡策略，也支持自定义扩展策略，如fair-根据服务器响应时间来均衡负载。

3) 动静分离

动静分离指将静态文件(如css、html、jpg、js等等文件)直接通过nginx返回，不再经过后台处理，提高资源响应的速度。

第二节 安装部署

1. Windows

3
1
- 下载：https://nginx.org/en/download.html    # nginx/Windows-1.25.4 
2
- 启动：./nginx.exe
3
- 访问：http://localhost:80                   # 默认端口为80，如果启动成功，访问时将会出现 Welcome to nginx 提示。

2. Linux

​x
1
# 扩展：一步式安装命令
2
yum install -y make cmake gcc gcc-c++ pcre pcre-devel zlib zlib-devel openssl openssl-devel && wget http://nginx.org/download/nginx-1.25.4.tar.gz && tar zxvf nginx-1.25.4.tar.gz && cd nginx-1.25.4 && ./configure --prefix=/usr/local/nginx --with-http_stub_status_module --with-http_ssl_module --with-file-aio --with-http_realip_module && make && make install && ln -s /usr/local/nginx/sbin/nginx /usr/bin/nginx
3

4
# 安装依赖库
5
yum install gcc-c++                   # 源码编译使用
6
yum install -y pcre pcre-devel        # Perl库，主要用于解析正则表达式  wget http://downloads.sourceforge.net/project/pcre/pcre/8.37/pcre-8.37.tar.gz
7
yum install -y zlib zlib-devel        # Nginx使用 zlib 对 http 包的内容进行 gzip
8
yum install -y openssl openssl-devel  # 用于支持 https
9

10
# 下载&解压安装包
11
https://nginx.org/en/download.html    # nginx-1.25.4
12
tar -zxvf nginx-1.25.4.tar.gz
13
cd nginx-1.25.4
14

15
# 配置&编译&安装
16
./configure
17
make
18
make install                          # 安装后通过 whereis nginx 查看安装位置
19

20
# 开机自启
21
vi /etc/rc.local # 增加一行 /usr/local/nginx/sbin/nginx
22
chmod 755 /etc/rc.local
23

3. 关于防火墙

20
1
# 防火墙服务开启&关闭&重启
2
service firewalld start
3
service firewalld stop
4
service firewalld restart
5

6
# 查看防火墙规则
7
firewall-cmd --list-all
8

9
# 查询端口是否开放
10
firewall-cmd --query-port=8080/tcp                   # firwall-cmd是Linux提供操作firewall的一个工具
11

12
# 开放80端口
13
firewall-cmd --permanent --add-port=80/tcp           # --permanent：表示设置为持久配置
14

15
# 移除端口
16
firewall-cmd --permanent --remove-port=8080/tcp
17

18
#重启防火墙(修改配置后要重启防火墙)
19
firewall-cmd --reload
20

4. 常用命令

20
1
# 启动Nginx
2
./nginx
3

4
# 关闭Nginx
5
nginx -s stop                  # 快速停止nginx
6
nginx -s quit                  # 完整有序的停止nginx
7
taskkill /f /t /im nginx.exe   # 直接杀死进程(仅Windos可用，/f表示强制终止进程，/t表示终止指定的进程和任何由此启动的子进程，/im用于指定的进程名称)
8

9
# 重新加载配置文件
10
./nginx -s reload 
11

12
# 查看nginx进程
13
ps aux|grep nginx              # Linux
14

15
# 查看nginx端口
16
 netstat -ntlp | grep nginx
17

18
# 查看nginx版本号
19
./nginx -v
20

第三节 配置介绍

1. 配置总览

36
1
# 一、全局块：服务器整体运行相关配置
2
worker_processes  1;
3

4
# 二、events块：用户网络连接相关选项
5
events {
6
    worker_connections  1024;
7
}
8

9
# 三、http块：功能模块配置
10
http {
11

12
    # http块全局配置
13
    include       mime.types;                 # 文件扩展名与文件类型映射表
14
    default_type  application/octet-stream;   # 访问到未定义的扩展名的时候，就默认为下载该文件
15
    sendfile        on;
16
    keepalive_timeout  65;
17

18
    # server块(多个)
19
    # 一个http块可以包含多个server块，而一个server块就等于一个虚拟主机
20
    server {
21
        
22
        # server块全局配置
23
        listen       80;                      #  目前监听的端口号
24
        server_name  notes.huangyuanxin.com;  # 主机名称    
25
        charset utf-8;
26

27
        # location块(多个)
28
        location / {
29
            root   /data/notes/;
30
            autoindex on; 
31
        }
32
        
33
    }
34

35
}
36

2. 全局配置

19
1
# 启动用户
2
user  root;
3

4
# 设置 worker 数量
5
# Nginx采用1Master+N*Worker的架构，Worker数一般为CPU数
6
worker_processes 4
7
# Worker绑定Cpu
8
worker_cpu_affinity 0001 0010 0100 1000                  # 4Worker绑定4Cpu
9
worker_cpu_affinity 0000001 00000010 00000100 00001000   # 4Worker绑定8Cpu的其中4个
10

11
# 日志文件
12
error_log  logs/error.log  error; # notice、error
13

14
# 进程文件
15
pid        logs/nginx.pid;
16

17
#  worker 进程的最大文件打开数限制
18
worker_rlimit_nofile 204800;
19

3. Events配置

5
1
# 每个Worker进程所能建立连接的最大值
2
# 一个Nginx服务器的总连接数为 worker_processes*worker_connections
3
# 一个静态资源访问请求占用2个连接，一个动态资源访问请求占4个连接
4
worker_connections  1024;
5

4. http块配置

16
1
# 日志格式
2
log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
3
'$status $body_bytes_sent "$http_referer" '
4
'"$http_user_agent" "$http_x_forwarded_for"';
5

6
# 访问日志
7
access_log  logs/access.log  main;
8

9
# 性能优化选项
10
#  Nginx 直接将文件内容发送给客户端，而不需要将文件内容先读入内存再发送。这样可以减少 CPU 和内存的使用，提高文件传输的效率
11
sendfile        on;
12
# 在 sendfile 打开的状态下才会生效，主要是用来提升网络包的传输效率
13
tcp_nopush      on;
14
# 在keep-alive连接开启的情况下才生效，来提高网络包传输的实时性
15
tcp_nodelay     on;
16

5. server块配置

4
1
# 日志
2
error_log  logs/error.log crit;
3
access_log  logs/access.log  main;
4

6. location块配置

9
1
# 后缀匹配
2
location ^~ /baseapp {
3
    if ($uri ~* \.(html|htm|gif|jpg|jpeg|bmp|png|ico|txt|js|css|tff|woff|swf|svg|xls|xlsx)$) {
4
        rewrite ^\/baseapp\/(.*)$ /$1;
5
        root D:\KingDom\web\web\base;
6
        break;
7
    }
8
    rewrite ^\/baseapp\/(.*)$ /$1;
9
}

第四节 配置示例

1. 反向代理配置

18
1
server {
2
    
3
    # 监听80端口的http请求(Host=106.53.120.230)
4
    listen       80;
5
    server_name  106.53.120.230;
6
    
7
    # 转发url与/order/匹配的请求
8
    location ~ /order/ {
9
        proxy_pass http://localhost:8001;
10
    }
11
    
12
    # 转发url与/pay/匹配的请求
13
    location ~ /pay/ {
14
        proxy_pass http://localhost:8002;
15
    }
16
    
17
}
18

2. 负载均衡配置

31
1
http {
2

3
    # 服务列表，默认为轮询策略
4
    upstream OrderServer_LX{
5
        server 106.53.120.230;
6
        server 106.53.120.231;
7
    }
8
    
9
    # 服务列表，修改为加权轮询策略
10
    upstream OrderServer_JQLX{
11
        server 106.53.120.230 weight=10;
12
        server 106.53.120.231 weight=5;
13
    }
14
    
15
    # 服务列表，修改为 IpHash 策略
16
    upstream OrderServer_IpHash{
17
        ip_hash;
18
        server 106.53.120.230;
19
        server 106.53.120.231;
20
    }
21
    
22
    server{
23
        location / {
24
            # 使用轮询策略转发
25
            proxy_pass http://OrderServer_LX;
26
            proxy_connect_timeout 10;
27
        }
28
    }
29

30
}
31

3. 动静分离配置

27
1
server {
2
    listen       80;
3
    server_name  106.53.120.230;
4
    charset utf-8;
5

6
    # 动态资源
7
    location ~ /order/ {
8
        # 转发到http://localhost:8001(tomcat)
9
        proxy_pass http://localhost:8001;
10
    }
11
    
12
    # 静态资源
13
    location ~ /image/ {
14
        # 转发到本地/data/目录，可访问/data/image/目录中的文件
15
        root   /data/;
16
        autoindex on; 
17
    }
18
    
19
    # 静态资源
20
    location ~ /css/ {
21
        # 转发到本地/data/目录，可访问/data/css/目录中的文件
22
        root   /data/;
23
        autoindex on; 
24
    }
25

26
}
27
        

4. 高可用主备配置

Nginx可基于KeppAlived进行主备配置，当主服务器不可访问时，备服务器将自动接管虚拟IP，处理接入请求。

1) 安装KeppAlived

3
1
# 安装
2
yum install keepalived –y
3

2) KeepAlived配置

44
1
# 全局配置
2
global_defs {
3
    # 告警邮箱配置
4
    notification_email {
5
        acassen@firewall.loc
6
        failover@firewall.loc
7
        sysadmin@firewall.loc
8
    }
9
    notification_email_from Alexandre.Cassen@firewall.loc
10
    smtp_server 192.168.17.129  
11
    smtp_connect_timeout 30     
12
        
13
    # 运行Keepalived的机器的路由标识，主备机应保持一致
14
    router_id LVS_DEVEL         
15
}
16

17
# 状态检查配置
18
vrrp_script chk_http_port {
19

20
    # 检查脚本
21
    script "/usr/local/src/nginx_check.sh"
22
    
23
    # 检查间隔(s)
24
    interval 2
25
        
26
    # 
27
    weight 2
28
}
29

30
# 虚拟服务器配置
31
vrrp_instance VI_1 {
32
    state MASTER         # 主服务器(MASTER)|备服务器(BACKUP)
33
    interface ens33      # 网卡
34
    virtual_router_id 51 # 主、备机的 virtual_router_id 必须相同
35
    priority 90          # 主、备机取不同的优先级，主机值较大，备份机值较小
36
    advert_int 1
37
    authentication {
38
        auth_type PASS
39
        auth_pass 1111
40
    }
41
    virtual_ipaddress {
42
        192.168.17.50    # VRRP H 虚拟地址
43
    }
44
}

3) nginx_check.sh

9
1
#!/bin/bash
2
A=`ps -C nginx –no-header |wc -l`
3
if [ $A -eq 0 ];then
4
    /usr/local/nginx/sbin/nginx
5
    sleep 2
6
    if [ `ps -C nginx --no-header |wc -l` -eq 0 ];then
7
        killall keepalived
8
    fi
9
fi

4)启动测试

10
1
# 启动主服务器和备服务器nginx
2
./nginx
3

4
# 启动主服务器和备服务器keepalived
5
systemctl start keepalived.service
6

7
# 测试
8
1) 浏览器访问虚拟IP 192.168.17.50 ，路由到主服务器
9
2) 关闭主服务器，备服务器的keepalived收不到主服务器信息时，将会自动选主，成为主服务器
10
3) 继续访问虚拟IP 192.168.17.50，路由到新的主服务器

第五节 其它配置

1. 响应压缩

在http块内或者在单个server块里添加如下配置开启响应压缩：

5
1
gzip  on;               # 开启压缩响应
2
gzip_http_version 1.1;  # 只针对http1.1进行压缩
3
gzip_comp_level 9;      # 压缩等级为9
4
gzip_types text/plain;  # 压缩类型为文本
5

然后重载nginx配置文件，在响应报文头中看到 Content-Encoding:gzip 即表示配置成功。

2. 支持HTTPS

首先通过nginx -V确认是否安装SSL模块(–with-http_ssl_module)，然后修改配置文件，添加server配置。

69
1
user  root;
2
worker_processes  1;
3

4
#error_log  logs/error.log;
5
#error_log  logs/error.log  notice;
6
#error_log  logs/error.log  info;
7

8
events {
9
    worker_connections  1024;
10
}
11

12

13
http {
14
    include            mime.types;
15
    default_type       application/octet-stream;
16
    sendfile           on;
17
    keepalive_timeout  65;
18
    #tcp_nopush        on;
19
    #gzip  on;
20

21
    # HTTP server
22
    server {
23
        listen       80;
24
        server_name  huangyuanxin.com www.huangyuanxin.com;
25
        # charset    koi8-r;
26

27
        # 重定向到HTTPS
28
        return 301 https://$server_name$request_uri;
29
    }
30

31

32
    # HTTPS server
33
    server {
34
        listen       443 ssl;
35
        server_name  huangyuanxin.com www.huangyuanxin.com;
36

37
        # SSL证书和私钥路径
38
        ssl_certificate      /usr/local/nginx/conf/cert.pem;
39
        ssl_certificate_key  /usr/local/nginx/conf/cert.key;
40

41
        # SSL相关配置
42
        ssl_session_cache          shared:SSL:1m;
43
        ssl_session_timeout        5m;
44
        ssl_ciphers                HIGH:!aNULL:!MD5;
45
        ssl_prefer_server_ciphers  on;
46

47
        # 静态文件
48
        location / {
49
            root   /root/htmls;
50
            index  index.html index.htm;
51
        }
52
        
53
        # API接口
54
        location ~/api/notes/ {
55
            proxy_pass https://localhost:8001;
56
        }
57
        location ~/api/ai/ {
58
            proxy_pass https://localhost:9001;
59
        }
60

61
        # 错误页面
62
        error_page   500 502 503 504  /50x.html;
63
        location = /50x.html {
64
            root   html;
65
        }
66
    }
67

68
}
69

3. 路径重写

在Nginx中，可以使用rewrite指令进行URL路径的重写：

14
1
server {
2
    listen 80;
3
    server_name example.com;
4
 
5
    location /oldpath/ {
6
        # 将所有以/oldpath/开头的URL重定向到以/newpath/开头的URL
7
        # last 表示完成rewrite步骤，并继续在新的URI上执行location匹配
8
        rewrite ^/oldpath/(.*)$ /newpath/$1 last;
9
    }
10
 
11
    location /newpath/ {
12
        # 这里是处理新路径的配置
13
    }
14
}

如果你想要内部重定向而不改变浏览器的URL，可以使用internal标识符：

12
1
server {
2
    listen 80;
3
    server_name example.com;
4
 
5
    location /oldpath/ {
6
        rewrite ^/oldpath/(.*)$ /newpath/$1 internal;
7
    }
8
 
9
    location /newpath/ {
10
        # 这里是处理新路径的配置
11
    }
12
}

这样，重写操作对用户来说是透明的，浏览器的地址栏不会显示新的URL路径。

第六节 常用插件

Nginx 的插件主要分为以下几类：

1. HTTP 模块插件

HTTP 模块插件扩展了 Nginx 的 HTTP 功能，包括添加新的 HTTP 模块、自定义请求处理流程等。

• ngx_http_rewrite_module

提供 URL 重写功能，可以实现 URL 的重定向、重写等操作。

7
1
server {
2
    ...
3
    location / {
4
        rewrite ^/old-url$ /new-url permanent;
5
    }
6
    ...
7
}

• ngx_http_access_module

提供访问控制功能，可以限制特定 IP 或者网络的访问。

8
1
server {
2
    ...
3
    location / {
4
        deny 192.168.1.1;
5
        allow all;
6
    }
7
    ...
8
}

2. 过滤器插件

过滤器插件对请求和响应进行过滤和处理，如压缩、重写、限速等。

• ngx_http_gzip_module

提供 Gzip 压缩功能，可以减小 HTTP 响应的大小，加快页面加载速度。

6
1
http {
2
    ...
3
    gzip on;
4
    gzip_types text/plain text/css application/json;
5
    ...
6
}

• ngx_http_ssl_module

提供 SSL/TLS 加密功能，可以保护 HTTP 通信的安全性。

7
1
server {
2
    ...
3
    listen 443 ssl;
4
    ssl_certificate /path/to/cert.pem;
5
    ssl_certificate_key /path/to/key.pem;
6
    ...
7
}

3. 负载均衡插件

负载均衡插件实现了负载均衡功能，将请求分发到多个后端服务器，实现高可用性和性能的提升。

• ngx_http_upstream_module

提供负载均衡功能。

15
1
http {
2
    ...
3
    upstream backend {
4
        server 127.0.0.1:8080;
5
        server 127.0.0.1:8081;
6
    }
7
    server {
8
        ...
9
        location / {
10
            proxy_pass http://backend;
11
        }
12
        ...
13
    }
14
    ...
15
}

• ngx_http_upstream_ip_hash_module

提供 IP 地址哈希负载均衡功能，将同一 IP 的请求分发到同一个后端服务器。

9
1
http {
2
    ...
3
    upstream backend {
4
        ip_hash;
5
        server 192.168.1.10;
6
        server 192.168.1.11;
7
    }
8
    ...
9
}

4. 安全插件

安全插件增强了 Nginx 的安全性，包括防火墙、反爬虫、反盗链等功能。

• ngx_http_limit_req_module

提供请求速率限制功能，可以防止恶意请求对服务器造成过载。

12
1
http {
2
    ...
3
    limit_req_zone $binary_remote_addr zone=one:10m rate=1r/s;
4
    server {
5
        ...
6
        location / {
7
            limit_req zone=one burst=5;
8
        }
9
        ...
10
    }
11
    ...
12
}

• ngx_http_limit_conn_module

提供连接数限制功能，可以限制每个 IP 的并发连接数。

12
1
http {
2
    ...
3
    limit_conn_zone $binary_remote_addr zone=addr:10m;
4
    server {
5
        ...
6
        location / {
7
            limit_conn addr 5;
8
        }
9
        ...
10
    }
11
    ...
12
}

5. 插件的安装和配置

下面以stub_status模块为例，简要说明Nginx插件的安装和配置：

15
1
# 1. 查看插件是否已安装
2
./nginx -v 2>&1|grep stub_status 
3

4
# 2. 配置编译参数(在原来的参数上加 --with-http_stub_status_module)
5
./configure --prefix=/usr/local/nginx --pid-path=/var/local/nginx/nginx.pid --lock-path=/var/local/nginx/nginx.lock --error-log-path=/var/local/nginx/error.log --http-log-path=/var/local/nginx/access.log --with-http_gzip_static_module --with-http_stub_status_module --http-client-body-temp-path=/var/local/nginx/client --http-proxy-temp-path=/var/local/nginx/proxy --http-fastcgi-temp-path=/var/local/nginx/fastcgi --http-uwsgi-temp-path=/var/local/nginx/uwsgi --http-scgi-temp-path=/var/local/nginx/scgi
6

7
# 3. 备份原nginx命令
8
cp /usr/local/nginx/sbin/nginx /usr/local/nginx/sbin/nginx.bak
9

10
# 4. 编译
11
make
12

13
# 5. 替换掉原来的二进制nginx文件
14
cp /root/wr/nginx-1.20.2/objs/nginx /usr/local/nginx/sbin/
15

stub_status模块主要用于实现监控功能，简单配置如下：

5
1
# server块
2
location /nginx_status {
3
    stub_status;
4
    allow all;
5
}

最后重载配置文件，访问 http://主机地址/nginx_status 即可测试。

1
1
./nginx -s reload

第02章_ZooKeeper

第一节 ZooKeeper简介

1. 什么是ZooKeeper？

ZooKeeper 是一个基于观察者模式设计的开源分布式协调服务，它负责存储和管理大家都关心的数据，一旦这些数据发生变化，便会通知相应的观察者做出响应。

它数据模型结构如下，和Unix文件系统非常相似：

其中每个节点称为ZNode，可以通过其路径唯一标识，并存储不超过 1MB 的数据。

2. 基本特性

• 1）Zookeeper：一个领导者（Leader），多个跟随者（Follower）组成的集群。

• 2）集群中只要有 半数以上 节点存活，Zookeeper集群就能正常服务。所以Zookeeper适合安装奇数台服务器。

• 3）全局数据一致：每个 Server 保存一份相同的数据副本，Client无论连接到哪个Server，数据都是一致的。

• 4）更新请求顺序执行，来自同一个 Client 的更新请求按其发送顺序依次执行。

• 5）数据更新原子性，一次数据更新要么成功，要么失败。

• 6）实时性，在一定时间范围内，Client能读到最新数据。

3. 选举机制

1) 第一次启动时

总结：投票过半数时，服务器 id 大的胜出

2) 非第一次启动时

总结： ①EPOCH大的直接胜出 ②EPOCH相同，事务id大的胜出 ③事务id相同，服务器id大的胜出

4. 应用场景

Zookeeper 提供的服务包括：统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线(注册中心)、软负载均衡等。

1) 统一命名服务

2) 统一配置管理

3) 统一集群管理

4) 注册中心

5) 负载均衡

第二节 安装部署

1. 集群部署

ZooKeeper一般部署为集群模式，且节点数为奇数，如3或5台等，当整个集群中有半数以上的节点存活时，那么整个集群环境才可用。

服务器台数多：好处，提高可靠性；坏处：提高通信延时。

为什么ZooKeeper集群一般为奇数个节点呢？

答：因为ZooKeeper集群选举leader时，必须要求半数以上的节点通过(不包含半数)，因此3台和2台服务器都是要求至少2台可用，整个集群才可用，以防止“脑裂”现象，同理，5台和4台也是类似情况。

1) 前置准备

18
1
# 1. 安装JDK
2
https://www.oracle.com/java/technologies/javase/jdk11-archive-downloads.html
3

4
# 2. 下载zookpper
5
https://zookeeper.apache.org/releases.html
6
http://mirror.bit.edu.cn/apache/zookeeper/
7

8
# 3. 上传和解压安装包
9
mkdir /usr/local/zookpper
10
tar -zxvf zookeeper-3.4.12.tar.gz
11

12
# 4. 创建data和logs目录
13
mkdir /usr/local/zookeeper/zookeeper-3.4.12/data
14
mkdir /usr/local/zookeeper/zookeeper-3.4.12/logs
15

16
# 5. 拷贝配置模板
17
cp ./conf/zoo_sample.cfg ./conf/zoo.cfg 
18

2) 配置zoo.cfg

22
1
# 通信心跳时间（毫秒）
2
tickTime=2000
3
# Leader和Follower初始连接时能容忍的最多心跳数
4
initLimit=10
5
# Leader和Follower数据同步时能容忍的最多心跳数（如果超过，则认为Follwer死掉，将其剔除）
6
syncLimit=5
7

8
# 数据存储目录及日志保存目录（如果没有指明dataLogDir，则日志也保存到这个文件夹中）
9
dataDir=/usr/local/zookeeper/zookeeper-3.4.12/data
10
dataLogDir=/usr/local/zookeeper/zookeeper-3.4.12/logs
11

12
# 管理服务端端口
13
admin.serverPort=8888
14
# 客户端连接的端口
15
clientPort=2181
16

17
# 集群配置
18
# server.服务器编号=服务器IP:Zookeeper服务器之间的通信端口:Leader选举的端口
19
server.0=10.0.0.116:2888:3888
20
server.1=10.0.0.117:2888:3888
21
server.2=10.0.0.118:2888:3888
22

3) 创建myid文件

2
1
cd /usr/local/zookeeper/zookeeper-3.4.12/data      # 必须配置在 dataDir 目录
2
echo 0 > myid                                      # 服务器编号，在集群中唯一，与服务器IP相对应

4) 设置环境变量

修改用户启动配置文件：vim ~/.bash_profile，然后重新加载：source ~/.bash_profile。

2
1
export ZK_HOME=/usr/local/software/zookeeper-3.4.12
2
export PATH=$PATH:$ZK_HOME/bin

5) 设置开机启动

在cd /etc/rc.d/init.d目录创建zookeeper服务：vim zookeeper，并授权：chmod +x /etc/rc.d/init.d/zookeeper。

7
1
#!/bin/bash
2
#chkconfig: 2345 10 90
3
#description: service zookeeper
4
export   JAVA_HOME=/usr/local/jdk1.8.0_191
5
export   ZOO_LOG_DIR=/opt/zookeeper/log
6
ZOOKEEPER_HOME=/usr/local/zookeeper-cluster/zookeeper-3.4.12/
7
su    root    ${ZOOKEEPER_HOME}/bin/zkServer.sh      "$1"

然后添加到开机启动项：chkconfig --add zookeeper，可通过chkconfig --list查看是否添加成功。

最后，可以通过reboot命令重启机器，查看zookeeper是否开机启动：service zookeeper status。

注意：若出现“服务 zookeeper 不支持 chkconfig”错误，请检查/etc/init.d/zookeeper文件的内容。

2. ZooKeeper启停

15
1
# 启动
2
zkServer.sh start
3

4
# 停止
5
zkServer.sh stop
6

7
# 重启服务
8
zkServer.sh restart
9

10
# 查看节点状态
11
[root@loaclhost ~]# zkServer.sh status
12
ZooKeeper JMX enabled by default
13
Using config: /usr/local/zookeeper-cluster/zookeeper-3.4.12/bin/../conf/zoo.cfg
14
Mode: leader
15

注意：如果查看ZK状态时发现ZK不在运行中，先使用`./zkServer.sh start-foreground查看启动报错信息。

3. 连接ZooKeeper

14
1
# 连接到ZK
2
zkCli.sh -server 127.0.0.1:2181
3

4
# 测试如下
5
--------------------------------------------
6
[zk: 127.0.0.1:2181(CONNECTED) 4] create /test
7
Created /test
8
[zk: 127.0.0.1:2181(CONNECTED) 5] ls /
9
[test, zookeeper]
10
--------------------------------------------
11

12
# 退出
13
quit
14

第三节 客户端命令

1. 查询命令

13
1
# 查看节点
2
ls /
3
ls /狗狗/边牧    # 查询节点
4
ls -R /          # 递归查询节点
5
ls -s /猩猩      # 查询节点的详细信息
6
stat /猩猩       # 查看节点状态
7

8
# 查看节点值
9
get /骆驼
10
get -s /猩猩     # 查询节点的详细信息
11
get -w /猩猩     # 监听节点的详细信息（注意：注册一次只能监听一次，想再次监听，需要再次注册）
12
ls -w /猩猩      # 监听节点的详细信息（注意：注册一次只能监听一次，想再次监听，需要再次注册）
13

2. 修改命令

16
1
# 创建节点
2
create /狗狗
3
create /狗狗/边牧
4
create -s /bb/bb1                  # 序列节点
5
create -e /cc                      # 临时节点 在会话结束后，自动被删除(通过这个特性，zk可以实现服务注册与发现的效果)
6
create -e -s /dd                   # 临时序列节点
7
create \[-s\] \[-e\] path data acl # 创建权限节点
8

9
# 删除节点
10
delete /aa         # 下面没有子节点的，可以直接用 delete
11
deleteall /手机    # 下面有子节点的，用 deleteall
12

13
# 设置节点值
14
set /猩猩 园外区3号
15
create /骆驼 园区2号                # 创建节点并写入值
16

3. 其它命令

7
1
# 查看帮助
2
help
3

4
# 添加认证用户
5
addauth digest maidou        # 不带密码
6
addauth digest beita:123456  # 带密码
7

第四节 客户端API

1. 基础案例

43
1
public class ZkApiTest {
2
    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
3

4
        // 建立连接
5
        // 参数1：连接字符串（支持以逗号分隔配置多个，随机进行连接）
6
        // 参数2：会话超时时间
7
        // 参数3：监听器
8
        // 其它参数：是否只读、密码等
9
        ZooKeeper zooKeeper = new ZooKeeper("10.210.5.252:2181", 1000 * 60, null);
10

11
        // 创建节点
12
        // 参数1：节点路径
13
        // 参数2：节点数据
14
        // 参数3：权限控制，一般是无权限控制，即ZooDefs.Ids.OPEN_ACL_UNSAFE
15
        // 参数4：节点类型，可选 PERSISTENT、PERSISTENT_SEQUENTIAL、EPHEMERAL、EPHEMERAL_SEQUENTIAL 等
16
        zooKeeper.create("/test", "test ok!".getBytes(StandardCharsets.UTF_8), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
17

18
        // 查询子节点
19
        List<String> children = zooKeeper.getChildren("/", null);
20
        System.out.println("子节点：" + children);
21

22
        // 判断节点是否存在
23
        Stat existStat = zooKeeper.exists("/test", null);
24
        System.out.println(existStat != null ? "节点存在" : "节点不存在");
25

26
        // 查询节点数据
27
        Stat stat = new Stat();
28
        byte[] data = zooKeeper.getData("/test", false, stat);
29
        System.out.println("当前事务ID：" + stat.getCzxid());
30
        System.out.println("当前节点数据：" + new String(data));
31

32
        // 修改节点数据
33
        Stat updatestat = zooKeeper.setData("/test", "new-data".getBytes(), -1);
34
        System.out.println("当前事务ID：" + updatestat.getCzxid());
35

36
        // 删除节点
37
        zooKeeper.delete("/test", -1);
38

39
        // 关闭连接
40
        zooKeeper.close();
41
    }
42
}
43

注意：

1. ZooKeeper集群中，只有 Leader 有权限发起数据修改操作，如果向 Follower 发起修改请求，则会转给 Leader 处理。

2. 实现注册中心

1) 服务注册

23
1
public class DistributeServer {
2
    private static ZooKeeper zk = null;
3
    private static String parentNode = "/test";
4
    private static String serverNode = "/server01";
5

6
    public static void main(String[] args) throws Exception {
7
        // 1. 连接到ZK
8
        zk = new ZooKeeper("10.210.5.252:2181", 1000 * 60, new
9
                Watcher() {
10
                    @Override
11
                    public void process(WatchedEvent event) {
12
                    }
13
                });
14

15
        // 2. 注册服务（临时节点）
16
        zk.create(parentNode + serverNode, serverNode.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
17

18
        // 3. 执行业务代码
19
        System.out.println("server01 is working ...");
20
        Thread.sleep(Long.MAX_VALUE);
21
    }
22
}
23

2) 服务监听

47
1
public class DistributeClient {
2
    private static ZooKeeper zk = null;
3
    private static String parentNode = "/test";
4

5
    public static void main(String[] args) throws Exception {
6
        // 1. 连接到ZK
7
        zk = new ZooKeeper("10.210.5.252:2181", 1000 * 60, new
8
                Watcher() {
9
                    @Override
10
                    public void process(WatchedEvent event) {
11
                        try {
12
                            //  再次启动监听
13
                            watchServerList();
14
                        } catch (Exception e) {
15
                            e.printStackTrace();
16
                        }
17
                    }
18
                });
19

20
        // 2. 监听当前在线服务
21
        watchServerList();
22

23
        // 3. 执行业务代码
24
        System.out.println("client is working ...");
25
        Thread.sleep(Long.MAX_VALUE);
26
    }
27

28

29
    // 监听当前在线服务
30
    public static void watchServerList() throws KeeperException, InterruptedException {
31
        List<String> servers = new ArrayList<>();
32

33
        // 1. 查询 /test 的子节点，并监听 /test 节点
34
        List<String> children = zk.getChildren(parentNode, true);
35

36
        // 2. 查询子节点信息
37
        for (String child : children) {
38
            String childNode = parentNode + "/" + child;
39
            byte[] data = zk.getData(childNode, false, null);
40
            servers.add(childNode + ":" + new String(data));
41
        }
42

43
        // 3. 打印
44
        System.out.println("servers = " + servers);
45
    }
46
    
47
}

3) 测试

10
1
操作步骤：
2
    1> 启动客户端进行“服务监听”
3
    2> 启动服务端进行“服务注册”
4
    3> 关闭服务端模拟”服务关闭“
5

6
客户端控制台打印：
7
servers = []
8
servers = [/test/server010000000011:/server01]
9
servers = []
10

3. 实现分布式锁

1) 原生方式

174
1
// 分布式锁实现
2
public class DistributedLock {
3

4
    // ZK连接
5
    private ZooKeeper zk;
6

7
    // 连接信息
8
    private String connectString = "10.210.5.252:2181";
9
    private int sessionTimeout = 1000 * 60;
10

11
    // 根节点、子节点前缀
12
    private String rootNode = "locks";
13
    private String subNode = "seq-";
14

15
    // 当前节点（等待锁或持有锁）
16
    private String currentNode;
17

18
    // 前置节点（当前节点等待的节点）
19
    private String waitPath;
20

21
    // 线程协作工具
22
    private CountDownLatch waitLatch = new CountDownLatch(1);
23

24
    // 和 zk 服务建立连接，并创建根节点
25
    public DistributedLock() throws IOException, InterruptedException, KeeperException {
26
        CountDownLatch connectLatch = new CountDownLatch(1);
27

28
        // 连接到ZK
29
        zk = new ZooKeeper(connectString, sessionTimeout, new
30
                Watcher() {
31
                    @Override
32
                    public void process(WatchedEvent event) {
33
                        // 连接建立时, 打开 latch, 唤醒 wait 在该 latch 上的线程
34
                        if (event.getState() == Event.KeeperState.SyncConnected) {
35
                            connectLatch.countDown();
36
                        }
37

38
                        // 发生了 waitPath 的删除事件
39
                        if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
40
                            waitLatch.countDown();
41
                        }
42
                    }
43
                });
44

45
        // 等待连接建立
46
        connectLatch.await();
47

48
        //获取根节点状态
49
        Stat stat = zk.exists("/" + rootNode, false);
50

51
        //如果根节点不存在，则创建根节点，根节点类型为永久节点
52
        if (stat == null) {
53
            System.out.println("根节点不存在");
54
            zk.create("/" + rootNode, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
55
        }
56
    }
57

58
    // 加锁方法
59
    public void zkLock() {
60
        try {
61
            // 在根节点下创建临时顺序节点，返回值为创建的节点路径
62
            currentNode = zk.create("/" + rootNode + "/" + subNode, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
63

64
            // wait 一小会, 让结果更清晰一些
65
            Thread.sleep(10);
66

67
            // 查询所有子节点
68
            // 注意：没有必要监听 /locks 的子节点的变化情况
69
            List<String> childrenNodes = zk.getChildren("/" + rootNode, false);
70

71
            // 如果列表中只有一个子节点, 那肯定就是 currentNode , 说明 client 获得锁
72
            if (childrenNodes.size() == 1) {
73
                return;
74
            }
75

76
            // 对子节点进行排序（从小到大）
77
            Collections.sort(childrenNodes);
78

79
            // 获取当前节点的位置
80
            String thisNode = currentNode.substring(("/" + rootNode + "/").length());
81
            int index = childrenNodes.indexOf(thisNode);
82

83
            // 如果未找到当前节点，则为数据异常
84
            if (index == -1) {
85
                throw new RuntimeException("数据异常");
86
            }
87

88
            // 如果thisNode 在列表中最小, 当前 client 获得锁
89
            if (index == 0) {
90
                return;
91
            }
92

93
            // 获得前置节点（当前节点的前1节点）
94
            this.waitPath = "/" + rootNode + "/" + childrenNodes.get(index - 1);
95

96
            // 在前置节点上注册监听器,
97
            // 当前置节点被删除时, zookeeper 会回调监听器的 process 方法
98
            zk.getData(waitPath, true, new Stat());
99

100
            // 等待前置节点被回调
101
            waitLatch.await();
102

103
        } catch (KeeperException | InterruptedException e) {
104
            e.printStackTrace();
105
        }
106
    }
107

108
    // 解锁方法
109
    public void zkUnlock() {
110
        try {
111
            zk.delete(this.currentNode, -1);
112
        } catch (InterruptedException | KeeperException e) {
113
            e.printStackTrace();
114
        }
115
    }
116

117
}
118

119
// 分布式锁测试
120
public class DistributedLockTest {
121

122
    // 测试
123
    public static void main(String[] args) throws InterruptedException, IOException, KeeperException {
124

125
        // 创建2个分布式锁
126
        final DistributedLock lock1 = new DistributedLock();
127
        final DistributedLock lock2 = new DistributedLock();
128

129
        // 线程1
130
        new Thread(new Runnable() {
131
            @Override
132
            public void run() {
133
                try {
134
                    lock1.zkLock();
135
                    System.out.println("线程 1 获取锁");
136

137
                    // 执行业务
138
                    Thread.sleep(5 * 1000);
139

140
                    lock1.zkUnlock();
141
                    System.out.println("线程 1 释放锁");
142
                } catch (Exception e) {
143
                    e.printStackTrace();
144
                }
145
            }
146
        }).start();
147

148
        // 线程2
149
        new Thread(new Runnable() {
150
            @Override
151
            public void run() {
152
                try {
153
                    lock2.zkLock();
154
                    System.out.println("线程 2 获取锁");
155

156
                    // 执行业务
157
                    Thread.sleep(5 * 1000);
158

159
                    lock2.zkUnlock();
160
                    System.out.println("线程 2 释放锁");
161
                } catch (Exception e) {
162
                    e.printStackTrace();
163
                }
164
            }
165
        }).start();
166
    }
167
}
168

169
// 结果打印
170
线程 2 获取锁
171
线程 2 释放锁
172
线程 1 获取锁
173
线程 1 释放锁
174