Elasticsearch 存储

::: info 概述

本文介绍了 Elasticsearch 的逻辑存储、物理存储，以及 Elasticsearch 的倒排索引设计。

:::

Flink 运维

docker 安装 flink

（1）使用 docker 命令拉取镜像

1

docker pull flink

（2）编写 docker-compose.yml，内容如下：

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

version: '2.1'
services:
  jobmanager:
    image: flink
    expose:
      - '6123'
    ports:
      - '8081:8081'
    command: jobmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager

  taskmanager:
    image: flink
    expose:
      - '6121'
      - '6122'
    depends_on:
      - jobmanager
    command: taskmanager
    links:
      - 'jobmanager:jobmanager'
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager

（3）执行 docker-compose，命令如下：

1

docker-compose up -d

（4）打开浏览器，访问 http://127.0.0.1:8081

Flink 配置

基础配置

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

# jobManager 的IP地址
jobmanager.rpc.address: localhost

# JobManager 的端口号
jobmanager.rpc.port: 6123

# JobManager JVM heap 内存大小
jobmanager.heap.size: 1024m

# TaskManager JVM heap 内存大小
taskmanager.heap.size: 1024m

# 每个 TaskManager 提供的任务 slots 数量大小
taskmanager.numberOfTaskSlots: 1

# 程序默认并行计算的个数
parallelism.default: 1
# 文件系统来源
# fs.default-scheme

高可用配置

1
2
3
4
5
6
7
8
9
10
11

# 可以选择 'NONE' 或者 'zookeeper'.
# high-availability: zookeeper

# 文件系统路径，让 Flink 在高可用性设置中持久保存元数据
# high-availability.storageDir: hdfs:///flink/ha/

# zookeeper 集群中仲裁者的机器 ip 和 port 端口号
# high-availability.zookeeper.quorum: localhost:2181

# 默认是 open，如果 zookeeper security 启用了该值会更改成 creator
# high-availability.zookeeper.client.acl: open

容错和 checkpoint 配置

1
2
3
4
5
6
7
8
9
10
11

# 用于存储和检查点状态
# state.backend: filesystem

# 存储检查点的数据文件和元数据的默认目录
# state.checkpoints.dir: hdfs://namenode-host:port/flink-checkpoints

# savepoints 的默认目标目录(可选)
# state.savepoints.dir: hdfs://namenode-host:port/flink-checkpoints

# 用于启用/禁用增量 checkpoints 的标志
# state.backend.incremental: false

Web UI 配置

1
2
3
4
5
6
7

# 基于 Web 的运行时监视器侦听的地址.
#jobmanager.web.address: 0.0.0.0

#  Web 的运行时监视器端口
rest.port: 8081
# 是否从基于 Web 的 jobmanager 启用作业提交
# jobmanager.web.submit.enable: false

高级配置

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

# io.tmp.dirs: /tmp

# 是否应在 TaskManager 启动时预先分配 TaskManager 管理的内存
# taskmanager.memory.preallocate: false

# 类加载解析顺序，是先检查用户代码 jar（“child-first”）还是应用程序类路径（“parent-first”）。 默认设置指示首先从用户代码 jar 加载类
# classloader.resolve-order: child-first

# 用于网络缓冲区的 JVM 内存的分数。 这决定了 TaskManager 可以同时拥有多少流数据交换通道以及通道缓冲的程度。 如果作业被拒绝或者您收到系统没有足够缓冲区的警告，请增加此值或下面的最小/最大值。 另请注意，“taskmanager.network.memory.min”和“taskmanager.network.memory.max”可能会覆盖此分数

# taskmanager.network.memory.fraction: 0.1
# taskmanager.network.memory.min: 67108864
# taskmanager.network.memory.max: 1073741824

Flink 集群安全配置

1
2
3
4
5
6
7
8
9
10
11

# 指示是否从 Kerberos ticket 缓存中读取
# security.kerberos.login.use-ticket-cache: true

# 包含用户凭据的 Kerberos 密钥表文件的绝对路径
# security.kerberos.login.keytab: /path/to/kerberos/keytab

# 与 keytab 关联的 Kerberos 主体名称
# security.kerberos.login.principal: flink-user

# 以逗号分隔的登录上下文列表，用于提供 Kerberos 凭据（例如，`Client，KafkaClient`使用凭证进行 ZooKeeper 身份验证和 Kafka 身份验证）
# security.kerberos.login.contexts: Client,KafkaClient

Zookeeper 安全配置

1
2
3
4
5

# 覆盖以下配置以提供自定义 ZK 服务名称
# zookeeper.sasl.service-name: zookeeper

# 该配置必须匹配 "security.kerberos.login.contexts" 中的列表（含有一个）
# zookeeper.sasl.login-context-name: Client

参考资料

• Flink 官方配置文档
• Apache Flink Docker Github
• 借助 Docker 学习大数据：Flink

ZooKeeper ACL

为了避免存储在 Zookeeper 上的数据被其他程序或者人为误修改，Zookeeper 提供了 ACL(Access Control Lists) 进行权限控制。

ACL 权限可以针对节点设置相关读写等权限，保障数据安全性。

ZooKeeper ACL 提供了以下几种命令行：

• getAcl 命令：获取某个节点的 acl 权限信息。
• setAcl 命令：设置某个节点的 acl 权限信息。
• addauth 命令：输入认证授权信息，注册时输入明文密码，加密形式保存。

ACL 组成

Zookeeper 的 acl 通过 [scheme:id:permissions] 来构成权限列表。

• scheme：代表采用的某种权限机制，包括 world、auth、digest、ip、super 几种。
  • world：默认模式，所有客户端都拥有指定的权限。world 下只有一个 id 选项，就是 anyone，通常组合写法为 world:anyone:[permissons]；
  • auth：只有经过认证的用户才拥有指定的权限。通常组合写法为 auth:user:password:[permissons]，使用这种模式时，你需要先进行登录，之后采用 auth 模式设置权限时，user 和 password 都将使用登录的用户名和密码；
  • digest：只有经过认证的用户才拥有指定的权限。通常组合写法为 auth:user:BASE64(SHA1(password)):[permissons]，这种形式下的密码必须通过 SHA1 和 BASE64 进行双重加密；
  • ip：限制只有特定 IP 的客户端才拥有指定的权限。通常组成写法为 ip:182.168.0.168:[permissions]；
  • super：代表超级管理员，拥有所有的权限，需要修改 Zookeeper 启动脚本进行配置。
• id：代表允许访问的用户。
• permissions：权限组合字符串，由 cdrwa 组成，其中每个字母代表支持不同权限。可选项如下：
  • CREATE：允许创建子节点；
  • READ：允许从节点获取数据并列出其子节点；
  • WRITE：允许为节点设置数据；
  • DELETE：允许删除子节点；
  • ADMIN：允许为节点设置权限。

设置与查看权限

想要给某个节点设置权限 (ACL)，有以下两个可选的命令：

1
2
3
4
5

# 1.给已有节点赋予权限
setAcl path acl

# 2.在创建节点时候指定权限
create [-s] [-e] path data acl

查看指定节点的权限命令如下：

1

getAcl path

添加认证信息

可以使用如下所示的命令为当前 Session 添加用户认证信息，等价于登录操作。

1
2
3
4
5

# 格式
addauth scheme auth

#示例：添加用户名为test,密码为root的用户认证信息
addauth digest test:root

权限设置示例

world 模式

world 是一种默认的模式，即创建时如果不指定权限，则默认的权限就是 world。

1
2
3
4
5
6
7
8
9

[zk: localhost:2181(CONNECTED) 32] create /mytest abc
Created /mytest
[zk: localhost:2181(CONNECTED) 4] getAcl /mytest
'world,'anyone # 默认的权限
: cdrwa
[zk: localhost:2181(CONNECTED) 34] setAcl /mytest world:anyone:cwda # 修改节点，不允许所有客户端读
....
[zk: localhost:2181(CONNECTED) 6] get /mytest
org.apache.zookeeper.KeeperException$NoAuthException: KeeperErrorCode = NoAuth for /mytest # 无权访问

auth 模式

1
2
3
4
5
6
7
8
9
10
11

[zk: localhost:2181(CONNECTED) 36] addauth digest test:root # 登录
[zk: localhost:2181(CONNECTED) 37] setAcl /mytest auth::cdrwa # 设置权限
[zk: localhost:2181(CONNECTED) 38] getAcl /mytest # 查看权限信息
'digest,'heibai:sCxtVJ1gPG8UW/jzFHR0A1ZKY5s= # 用户名和密码 (密码经过加密处理)，注意返回的权限类型是 digest
: cdrwa

# 用户名和密码都是使用登录的用户名和密码，即使你在创建权限时候进行指定也是无效的
[zk: localhost:2181(CONNECTED) 39] setAcl /mytest auth:root:root:cdrwa    #指定用户名和密码为 root
[zk: localhost:2181(CONNECTED) 40] getAcl /mytest
'digest,'heibai:sCxtVJ1gPG8UW/jzFHR0A1ZKY5s=  #无效，使用的用户名和密码依然还是 test
: cdrwa

digest 模式

1
2
3
4

[zk:44] create /spark "spark" digest:heibai:sCxtVJ1gPG8UW/jzFHR0A1ZKY5s=:cdrwa  #指定用户名和加密后的密码
[zk:45] getAcl /spark  #获取权限
'digest,'heibai:sCxtVJ1gPG8UW/jzFHR0A1ZKY5s=   # 返回的权限类型是 digest
: cdrwa

到这里你可以发现使用 auth 模式设置的权限和使用 digest 模式设置的权限，在最终结果上，得到的权限模式都是 digest。某种程度上，你可以把 auth 模式理解成是 digest 模式的一种简便实现。因为在 digest 模式下，每次设置都需要书写用户名和加密后的密码，这是比较繁琐的，采用 auth 模式就可以避免这种麻烦。

ip 模式

限定只有特定的 ip 才能访问。

1
2
3

[zk: localhost:2181(CONNECTED) 46] create /hive "hive" ip:192.168.0.108:cdrwa
[zk: localhost:2181(CONNECTED) 47] get /hive
Authentication is not valid : /hive  # 当前主机已经不能访问

这里可以看到当前主机已经不能访问，想要能够再次访问，可以使用对应 IP 的客户端，或使用下面介绍的 super 模式。

super 模式

需要修改启动脚本 zkServer.sh，并在指定位置添加超级管理员账户和密码信息：

1

"-Dzookeeper.DigestAuthenticationProvider.superDigest=heibai:sCxtVJ1gPG8UW/jzFHR0A1ZKY5s="

修改完成后需要使用 zkServer.sh restart 重启服务，此时再次访问限制 IP 的节点：

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

[zk: localhost:2181(CONNECTED) 0] get /hive  #访问受限
Authentication is not valid : /hive
[zk: localhost:2181(CONNECTED) 1] addauth digest heibai:heibai  # 登录 (添加认证信息)
[zk: localhost:2181(CONNECTED) 2] get /hive  #成功访问
hive
cZxid = 0x158
ctime = Sat May 25 09:11:29 CST 2019
mZxid = 0x158
mtime = Sat May 25 09:11:29 CST 2019
pZxid = 0x158
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 0

参考资料

• Zookeeper 安装
• Zookeeper 单机环境和集群环境搭建
• Zookeeper 客户端基础命令使用

ZooKeeper Java Api

ZooKeeper 是 Apache 的顶级项目。ZooKeeper 为分布式应用提供了高效且可靠的分布式协调服务，提供了诸如统一命名服务、配置管理和分布式锁等分布式的基础服务。在解决分布式数据一致性方面，ZooKeeper 并没有直接采用 Paxos 算法，而是采用了名为 ZAB 的一致性协议。

ZooKeeper 主要用来解决分布式集群中应用系统的一致性问题，它能提供基于类似于文件系统的目录节点树方式的数据存储。但是 ZooKeeper 并不是用来专门存储数据的，它的作用主要是用来维护和监控存储数据的状态变化。通过监控这些数据状态的变化，从而可以达到基于数据的集群管理。

很多大名鼎鼎的框架都基于 ZooKeeper 来实现分布式高可用，如：Dubbo、Kafka 等。

ZooKeeper 官方支持 Java 和 C 的 Client API。ZooKeeper 社区为大多数语言（.NET，python 等）提供非官方 API。

ZooKeeper 官方客户端

ZooKeeper 客户端简介

客户端和服务端交互遵循以下基本步骤：

1. 客户端连接 ZooKeeper 服务端集群任意工作节点，该节点为客户端分配会话 ID。
2. 为了保持通信，客户端需要和服务端保持心跳（实质上就是 ping ）。否则，ZooKeeper 服务会话超时时间内未收到客户端请求，会将会话视为过期。这种情况下，客户端如果要通信，就需要重新连接。
3. 只要会话 ID 处于活动状态，就可以执行读写 znode 操作。
4. 所有任务完成后，客户端断开与 ZooKeeper 服务端集群的连接。如果客户端长时间不活动，则 ZooKeeper 集合将自动断开客户端。

ZooKeeper 官方客户端的核心是 ZooKeeper 类。它在其构造函数中提供了连接 ZooKeeper 服务的配置选项，并提供了访问 ZooKeeper 数据的方法。

其主要操作如下：

• connect - 连接 ZooKeeper 服务
• create - 创建 znode
• exists - 检查 znode 是否存在及其信息
• getACL / setACL- 获取/设置一个 znode 的 ACL
• getData / setData- 获取/设置一个 znode 的数据
• getChildren - 获取特定 znode 中的所有子节点
• delete - 删除特定的 znode 及其所有子项
• close - 关闭连接

ZooKeeper 官方客户端的使用方法是在 maven 项目的 pom.xml 中添加：

1
2
3
4
5

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.7.0</version>
</dependency>

创建连接

ZooKeeper 类通过其构造函数提供连接 ZooKeeper 服务的功能。其构造函数的定义如下：

1

ZooKeeper(String connectionString, int sessionTimeout, Watcher watcher)

参数说明：

• connectionString - ZooKeeper 集群的主机列表。
• sessionTimeout - 会话超时时间（以毫秒为单位）。
• watcher - 实现监视机制的回调。当被监控的 znode 状态发生变化时，ZooKeeper 服务端的 WatcherManager 会主动调用传入的 Watcher ，推送状态变化给客户端。

【示例】连接 ZooKeeper

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

import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.junit.jupiter.api.*;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

/**
 * ZooKeeper 官方客户端测试例
 *
 * @author <a href="mailto:forbreak@163.com">Zhang Peng</a>
 * @since 2022-02-19
 */
@DisplayName("ZooKeeper 官方客户端测试例")
public class ZooKeeperTest {

    /**
     * ZooKeeper 连接实例
     */
    private static ZooKeeper zk;

    /**
     * 创建 ZooKeeper 连接
     */
    @BeforeAll
    public static void init() throws IOException, InterruptedException {
        final String HOST = "localhost:2181";
        CountDownLatch latch = new CountDownLatch(1);
        zk = new ZooKeeper(HOST, 5000, watcher -> {
            if (watcher.getState() == Watcher.Event.KeeperState.SyncConnected) {
                latch.countDown();
            }
        });
        latch.await();
    }

    /**
     * 关闭 ZooKeeper 连接
     */
    @AfterAll
    public static void destroy() throws InterruptedException {
        if (zk != null) {
            zk.close();
        }
    }

    /**
     * 建立连接
     */
    @Test
    public void getState() {
        ZooKeeper.States state = zk.getState();
        Assertions.assertTrue(state.isAlive());
    }

}

说明：

添加一个 connect 方法，用于创建一个 ZooKeeper 对象，用于连接到 ZooKeeper 服务。

这里 CountDownLatch 用于停止（等待）主进程，直到客户端与 ZooKeeper 集合连接。

ZooKeeper 对象通过监听器回调来监听连接状态。一旦客户端与 ZooKeeper 建立连接，监听器回调就会被调用；并且监听器回调函数调用 CountDownLatch 的 countDown 方法来释放锁，在主进程中 await。

节点增删改查

判断节点是否存在

ZooKeeper 类提供了 exists 方法来检查 znode 的存在。如果指定的 znode 存在，则返回一个 znode 的元数据。

exists 方法的签名如下：

1

exists(String path, boolean watcher)

• path- Znode 路径
• watcher - 布尔值，用于指定是否监视指定的 znode

【示例】

1
2

Stat stat = zk.exists("/", true);
Assertions.assertNotNull(stat);

创建节点

ZooKeeper 类的 create 方法用于在 ZooKeeper 中创建一个新节点（znode）。

create 方法的签名如下：

1

create(String path, byte[] data, List<ACL> acl, CreateMode createMode)

• path - Znode 路径。例如，/myapp1，/myapp2，/myapp1/mydata1，myapp2/mydata1/myanothersubdata
• data - 要存储在指定 znode 路径中的数据
• acl - 要创建的节点的访问控制列表。ZooKeeper API 提供了一个静态接口 ZooDefs.Ids 来获取一些基本的 acl 列表。例如，ZooDefs.Ids.OPEN_ACL_UNSAFE 返回打开 znode 的 acl 列表。
• createMode - 节点的类型，即临时，顺序或两者。这是一个枚举。

【示例】

1
2
3
4
5
6

private static final String path = "/mytest";

String text = "My first zookeeper app";
zk.create(path, text.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
Stat stat = zk.exists(path, true);
Assertions.assertNotNull(stat);

删除节点

ZooKeeper 类提供了 delete 方法来删除指定的 znode。

delete 方法的签名如下：

1

delete(String path, int version)

• path - Znode 路径。
• version - znode 的当前版本。

让我们创建一个新的 Java 应用程序来了解 ZooKeeper API 的 delete 功能。创建文件 ZKDelete.java 。在 main 方法中，使用 ZooKeeperConnection 对象创建一个 ZooKeeper 对象 zk 。然后，使用指定的路径和版本号调用 zk 对象的 delete 方法。

删除 znode 的完整程序代码如下：

【示例】

1
2
3

zk.delete(path, zk.exists(path, true).getVersion());
Stat stat = zk.exists(path, true);
Assertions.assertNull(stat);

获取节点数据

ZooKeeper 类提供 getData 方法来获取附加在指定 znode 中的数据及其状态。 getData 方法的签名如下：

1

getData(String path, Watcher watcher, Stat stat)

• path - Znode 路径。
• watcher - 监听器类型的回调函数。当指定的 znode 的数据改变时，ZooKeeper 集合将通过监听器回调进行通知。这是一次性通知。
• stat - 返回 znode 的元数据。

【示例】

1
2
3
4

byte[] data = zk.getData(path, false, null);
String text1 = new String(data);
Assertions.assertEquals(text, text1);
System.out.println(text1);

设置节点数据

ZooKeeper 类提供 setData 方法来修改指定 znode 中附加的数据。 setData 方法的签名如下：

1

setData(String path, byte[] data, int version)

• path- Znode 路径
• data - 要存储在指定 znode 路径中的数据。
• version- znode 的当前版本。每当数据更改时，ZooKeeper 会更新 znode 的版本号。

【示例】

1
2
3
4
5
6
7
8

String text = "含子节点的节点";
zk.create(path, text.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
zk.create(path + "/1", "1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
zk.create(path + "/2", "1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
List<String> actualList = zk.getChildren(path, false);
for (String child : actualList) {
    System.out.println(child);
}

获取子节点

ZooKeeper 类提供 getChildren 方法来获取特定 znode 的所有子节点。 getChildren 方法的签名如下：

1

getChildren(String path, Watcher watcher)

• path - Znode 路径。
• watcher - 监听器类型的回调函数。当指定的 znode 被删除或 znode 下的子节点被创建/删除时，ZooKeeper 集合将进行通知。这是一次性通知。

【示例】

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

@Test
public void getChildren() throws InterruptedException, KeeperException {
    byte[] data = "My first zookeeper app".getBytes();
    zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    zk.create(path + "/1", "1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    zk.create(path + "/2", "1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    List<String> actualList = zk.getChildren(path, false);
    List<String> expectedList = CollectionUtil.newArrayList("1", "2");
    Assertions.assertTrue(CollectionUtil.containsAll(expectedList, actualList));
    for (String child : actualList) {
        System.out.println(child);
    }
}

Curator 客户端

Curator 客户端简介

Curator 客户端的使用方法是在 maven 项目的 pom.xml 中添加：

1
2
3
4
5

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>5.1.0</version>
</dependency>

创建连接

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

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.junit.jupiter.api.*;

import java.nio.charset.StandardCharsets;

public class CuratorTest {

    /**
     * Curator ZooKeeper 连接实例
     */
    private static CuratorFramework client = null;
    private static final String path = "/mytest";

    /**
     * 创建连接
     */
    @BeforeAll
    public static void init() {
        // 重试策略
        RetryPolicy retryPolicy = new RetryNTimes(3, 5000);
        client = CuratorFrameworkFactory.builder()
                                        .connectString("localhost:2181")
                                        .sessionTimeoutMs(10000).retryPolicy(retryPolicy)
                                        .namespace("workspace").build();  //指定命名空间后，client 的所有路径操作都会以 /workspace 开头
        client.start();
    }

    /**
     * 关闭连接
     */
    @AfterAll
    public static void destroy() {
        if (client != null) {
            client.close();
        }
    }

}

节点增删改查

判断节点是否存在

1
2

Stat stat = client.checkExists().forPath(path);
Assertions.assertNull(stat);

判读服务状态

1
2

CuratorFrameworkState state = client.getState();
Assertions.assertEquals(CuratorFrameworkState.STARTED, state);

创建节点

1
2
3
4
5
6

// 创建节点
String text = "Hello World";
client.create().creatingParentsIfNeeded()
      .withMode(CreateMode.PERSISTENT)      //节点类型
      .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
      .forPath(path, text.getBytes(StandardCharsets.UTF_8));

删除节点

1
2
3
4
5

client.delete()
      .guaranteed()                     // 如果删除失败，会继续执行，直到成功
      .deletingChildrenIfNeeded()       // 如果有子节点，则递归删除
      .withVersion(stat.getVersion())   // 传入版本号，如果版本号错误则拒绝删除操作，并抛出 BadVersion 异常
      .forPath(path);

获取节点数据

1
2
3

byte[] data = client.getData().forPath(path);
Assertions.assertEquals(text, new String(data));
System.out.println("修改前的节点数据：" + new String(data));

设置节点数据

1
2
3
4

String text2 = "try again";
client.setData()
      .withVersion(client.checkExists().forPath(path).getVersion())
      .forPath(path, text2.getBytes(StandardCharsets.UTF_8));

获取子节点

1
2
3
4
5
6

List<String> children = client.getChildren().forPath(path);
for (String s : children) {
    System.out.println(s);
}
List<String> expectedList = CollectionUtil.newArrayList("1", "2");
Assertions.assertTrue(CollectionUtil.containsAll(expectedList, children));

监听事件

创建一次性监听

和 Zookeeper 原生监听一样，使用 usingWatcher 注册的监听是一次性的，即监听只会触发一次，触发后就销毁。

【示例】

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

// 设置监听器
client.getData().usingWatcher(new CuratorWatcher() {
    public void process(WatchedEvent event) {
        System.out.println("节点 " + event.getPath() + " 发生了事件：" + event.getType());
    }
}).forPath(path);

// 第一次修改
client.setData()
      .withVersion(client.checkExists().forPath(path).getVersion())
      .forPath(path, "第一次修改".getBytes(StandardCharsets.UTF_8));

// 第二次修改
client.setData()
      .withVersion(client.checkExists().forPath(path).getVersion())
      .forPath(path, "第二次修改".getBytes(StandardCharsets.UTF_8));

输出

1

节点 /mytest 发生了事件：NodeDataChanged

说明

修改两次数据，但是监听器只会监听第一次修改。

创建永久监听

Curator 还提供了创建永久监听的 API，其使用方式如下：

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

// 设置监听器
CuratorCache curatorCache = CuratorCache.builder(client, path).build();
PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
    @Override
    public void childEvent(CuratorFramework framework, PathChildrenCacheEvent event) throws Exception {
        System.out.println("节点 " + event.getData().getPath() + " 发生了事件：" + event.getType());
    }
};
CuratorCacheListener listener = CuratorCacheListener.builder()
                                                    .forPathChildrenCache(path, client,
                                                        pathChildrenCacheListener)
                                                    .build();
curatorCache.listenable().addListener(listener);
curatorCache.start();

// 第一次修改
client.setData()
      .withVersion(client.checkExists().forPath(path).getVersion())
      .forPath(path, "第一次修改".getBytes(StandardCharsets.UTF_8));

// 第二次修改
client.setData()
      .withVersion(client.checkExists().forPath(path).getVersion())
      .forPath(path, "第二次修改".getBytes(StandardCharsets.UTF_8));

监听子节点

这里以监听 /hadoop 下所有子节点为例，实现方式如下：

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

// 创建节点
String text = "Hello World";
client.create().creatingParentsIfNeeded()
      .withMode(CreateMode.PERSISTENT)      //节点类型
      .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
      .forPath(path, text.getBytes(StandardCharsets.UTF_8));
client.create().creatingParentsIfNeeded()
      .withMode(CreateMode.PERSISTENT)      //节点类型
      .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
      .forPath(path + "/1", text.getBytes(StandardCharsets.UTF_8));
client.create().creatingParentsIfNeeded()
      .withMode(CreateMode.PERSISTENT)      //节点类型
      .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
      .forPath(path + "/2", text.getBytes(StandardCharsets.UTF_8));

// 设置监听器
// 第三个参数代表除了节点状态外，是否还缓存节点内容
PathChildrenCache childrenCache = new PathChildrenCache(client, path, true);
/*
 * StartMode 代表初始化方式:
 *    NORMAL: 异步初始化
 *    BUILD_INITIAL_CACHE: 同步初始化
 *    POST_INITIALIZED_EVENT: 异步并通知,初始化之后会触发 INITIALIZED 事件
 */
childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);

List<ChildData> childDataList = childrenCache.getCurrentData();
System.out.println("当前数据节点的子节点列表：");
childDataList.forEach(x -> System.out.println(x.getPath()));

childrenCache.getListenable().addListener(new PathChildrenCacheListener() {
    public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) {
        switch (event.getType()) {
            case INITIALIZED:
                System.out.println("childrenCache 初始化完成");
                break;
            case CHILD_ADDED:
                // 需要注意的是: 即使是之前已经存在的子节点，也会触发该监听，因为会把该子节点加入 childrenCache 缓存中
                System.out.println("增加子节点:" + event.getData().getPath());
                break;
            case CHILD_REMOVED:
                System.out.println("删除子节点:" + event.getData().getPath());
                break;
            case CHILD_UPDATED:
                System.out.println("被修改的子节点的路径:" + event.getData().getPath());
                System.out.println("修改后的数据:" + new String(event.getData().getData()));
                break;
        }
    }
});

// 第一次修改
client.setData()
      .forPath(path + "/1", "第一次修改".getBytes(StandardCharsets.UTF_8));

// 第二次修改
client.setData()
      .forPath(path + "/1", "第二次修改".getBytes(StandardCharsets.UTF_8));

ACL 权限管理

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

public class AclOperation {

    private CuratorFramework client = null;
    private static final String zkServerPath = "192.168.0.226:2181";
    private static final String nodePath = "/mytest/hdfs";

    @Before
    public void prepare() {
        RetryPolicy retryPolicy = new RetryNTimes(3, 5000);
        client = CuratorFrameworkFactory.builder()
                .authorization("digest", "heibai:123456".getBytes()) //等价于 addauth 命令
                .connectString(zkServerPath)
                .sessionTimeoutMs(10000).retryPolicy(retryPolicy)
                .namespace("workspace").build();
        client.start();
    }

    /**
     * 新建节点并赋予权限
     */
    @Test
    public void createNodesWithAcl() throws Exception {
        List<ACL> aclList = new ArrayList<>();
        // 对密码进行加密
        String digest1 = DigestAuthenticationProvider.generateDigest("heibai:123456");
        String digest2 = DigestAuthenticationProvider.generateDigest("ying:123456");
        Id user01 = new Id("digest", digest1);
        Id user02 = new Id("digest", digest2);
        // 指定所有权限
        aclList.add(new ACL(Perms.ALL, user01));
        // 如果想要指定权限的组合，中间需要使用 | ,这里的|代表的是位运算中的 按位或
        aclList.add(new ACL(Perms.DELETE | Perms.CREATE, user02));

        // 创建节点
        byte[] data = "abc".getBytes();
        client.create().creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT)
                .withACL(aclList, true)
                .forPath(nodePath, data);
    }


    /**
     * 给已有节点设置权限,注意这会删除所有原来节点上已有的权限设置
     */
    @Test
    public void SetAcl() throws Exception {
        String digest = DigestAuthenticationProvider.generateDigest("admin:admin");
        Id user = new Id("digest", digest);
        client.setACL()
                .withACL(Collections.singletonList(new ACL(Perms.READ | Perms.DELETE, user)))
                .forPath(nodePath);
    }

    /**
     * 获取权限
     */
    @Test
    public void getAcl() throws Exception {
        List<ACL> aclList = client.getACL().forPath(nodePath);
        ACL acl = aclList.get(0);
        System.out.println(acl.getId().getId()
                           + "是否有删读权限:" + (acl.getPerms() == (Perms.READ | Perms.DELETE)));
    }

    @After
    public void destroy() {
        if (client != null) {
            client.close();
        }
    }
}

参考资料

• 官方
  • ZooKeeper 官网
  • ZooKeeper 官方文档
  • ZooKeeper Github
• 书籍
  • 《Hadoop 权威指南（第四版）》
• 文章
  • 分布式服务框架 ZooKeeper – 管理分布式环境中的数据
  • ZooKeeper 的功能以及工作原理
  • ZooKeeper 简介及核心概念
  • 详解分布式协调服务 ZooKeeper
  • 深入浅出 Zookeeper（一） Zookeeper 架构及 FastLeaderElection 机制

ZooKeeper 命令

ZooKeeper 命令用于在 ZooKeeper 服务上执行操作。

启动服务和启动命令行

1
2
3
4
5

# 启动服务
bin/zkServer.sh start

# 启动命令行，不指定服务地址则默认连接到localhost:2181
bin/zkCli.sh -server hadoop001:2181

查看节点列表

ls 命令

ls 命令用于查看某个路径下目录列表。

【语法】

1

ls path

说明：

• path：代表路径。

【示例】

1
2

[zk: localhost:2181(CONNECTED) 0] ls /
[cluster, controller_epoch, brokers, storm, zookeeper, admin,  ...]

ls2 命令

ls2 命令用于查看某个路径下目录列表，它比 ls 命令列出更多的详细信息。

【语法】

1

ls2 path

说明：

• path：代表路径。

【示例】

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

[zk: localhost:2181(CONNECTED) 1] ls2 /
[cluster, controller_epoch, brokers, storm, zookeeper, admin, ....]
cZxid = 0x0
ctime = Thu Jan 01 08:00:00 CST 1970
mZxid = 0x0
mtime = Thu Jan 01 08:00:00 CST 1970
pZxid = 0x130
cversion = 19
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 11

节点的增删改查

get 命令

get 命令用于获取节点数据和状态信息。

【语法】

1

get path [watch]

说明：

• path：代表路径。
• **[watch]**：对节点进行事件监听。

【示例】

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

[zk: localhost:2181(CONNECTED) 31] get /hadoop
123456   #节点数据
cZxid = 0x14b
ctime = Fri May 24 17:03:06 CST 2019
mZxid = 0x14b
mtime = Fri May 24 17:03:06 CST 2019
pZxid = 0x14b
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0

说明：

节点各个属性如下表。其中一个重要的概念是 Zxid(ZooKeeper Transaction Id)，ZooKeeper 节点的每一次更改都具有唯一的 Zxid，如果 Zxid1 小于 Zxid2，则 Zxid1 的更改发生在 Zxid2 更改之前。

状态属性	说明
cZxid	数据节点创建时的事务 ID
ctime	数据节点创建时的时间
mZxid	数据节点最后一次更新时的事务 ID
mtime	数据节点最后一次更新时的时间
pZxid	数据节点的子节点最后一次被修改时的事务 ID
cversion	子节点的更改次数
dataVersion	节点数据的更改次数
aclVersion	节点的 ACL 的更改次数
ephemeralOwner	如果节点是临时节点，则表示创建该节点的会话的 SessionID；如果节点是持久节点，则该属性值为 0
dataLength	数据内容的长度
numChildren	数据节点当前的子节点个数

stat 命令

stat 命令用于查看节点状态信息。它的返回值和 get 命令类似，但不会返回节点数据。

【语法】

1

stat path [watch]

• path：代表路径。
• **[watch]**：对节点进行事件监听。

【示例】

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

[zk: localhost:2181(CONNECTED) 32] stat /hadoop
cZxid = 0x14b
ctime = Fri May 24 17:03:06 CST 2019
mZxid = 0x14b
mtime = Fri May 24 17:03:06 CST 2019
pZxid = 0x14b
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0

create 命令

create 命令用于创建节点并赋值。

【语法】

1

create [-s] [-e] path data acl

说明：

• **[-s][-e]**：-s 和 -e 都是可选的，-s 代表顺序节点，-e 代表临时节点，注意其中 -s 和 -e 可以同时使用的，并且临时节点不能再创建子节点。
  • 默认情况下，所有 znode 都是持久的。
  • 顺序节点保证 znode 路径将是唯一的。
  • 临时节点会在会话过期或客户端断开连接时被自动删除。
• path：指定要创建节点的路径，比如 /hadoop。
• data：要在此节点存储的数据。
• acl：访问权限相关，默认是 world，相当于全世界都能访问。

【示例】创建持久节点

1
2

[zk: localhost:2181(CONNECTED) 4] create /hadoop 123456
Created /hadoop

【示例】创建有序节点，此时创建的节点名为指定节点名 + 自增序号：

1
2
3
4
5
6

[zk: localhost:2181(CONNECTED) 23] create -s /a  "aaa"
Created /a0000000022
[zk: localhost:2181(CONNECTED) 24] create -s /b  "bbb"
Created /b0000000023
[zk: localhost:2181(CONNECTED) 25] create -s /c  "ccc"
Created /c0000000024

【示例】创建临时节点：

1
2

[zk: localhost:2181(CONNECTED) 26] create -e /tmp  "tmp"
Created /tmp

set 命令

set 命令用于修改节点存储的数据。

【语法】

1

set path data [version]

说明：

• path：节点路径。
• data：需要存储的数据。
• **[version]**：可选项，版本号(可用作乐观锁)。

【示例】

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

[zk: localhost:2181(CONNECTED) 33] set /hadoop 345
cZxid = 0x14b
ctime = Fri May 24 17:03:06 CST 2019
mZxid = 0x14c
mtime = Fri May 24 17:13:05 CST 2019
pZxid = 0x14b
cversion = 0
dataVersion = 1  # 注意更改后此时版本号为 1，默认创建时为 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 3
numChildren = 0

也可以基于版本号进行更改，此时类似于乐观锁机制，当你传入的数据版本号 (dataVersion) 和当前节点的数据版本号不符合时，zookeeper 会拒绝本次修改：

1
2

[zk: localhost:2181(CONNECTED) 34] set /hadoop 678 0
version No is not valid : /hadoop    #无效的版本号

delete 命令

delete 命令用于删除某节点。

【语法】

1

delete path [version]

说明：

• path：节点路径。
• **[version]**：可选项，版本号（同 set 命令）。和更新节点数据一样，也可以传入版本号，当你传入的数据版本号 (dataVersion) 和当前节点的数据版本号不符合时，zookeeper 不会执行删除操作。

【示例】

1
2
3
4

[zk: localhost:2181(CONNECTED) 36] delete /hadoop 0
version No is not valid : /hadoop   #无效的版本号
[zk: localhost:2181(CONNECTED) 37] delete /hadoop 1
[zk: localhost:2181(CONNECTED) 38]

delete 命令不能删除带有子节点的节点。如果想要删除节点及其子节点，可以使用 deleteall path

监听器

针对每个节点的操作，都会有一个监听者（watcher）。

• 当监听的某个对象（znode）发生了变化，则触发监听事件。
• zookeeper 中的监听事件是一次性的，触发后立即销毁。
• 父节点，子节点的增删改都能够触发其监听者（watcher）
• 针对不同类型的操作，触发的 watcher 事件也不同：
  • 父节点 Watcher 事件
    • 创建父节点触发：NodeCreated
    • 修改节点数据触发：NodeDatachanged
    • 删除节点数据触发：NodeDeleted
  • 子节点 Watcher 事件
    • 创建子节点触发：NodeChildrenChanged
    • 删除子节点触发：NodeChildrenChanged
    • 修改子节点不触发事件

get path

使用 get path -w 注册的监听器能够在节点内容发生改变的时候，向客户端发出通知。需要注意的是 zookeeper 的触发器是一次性的 (One-time trigger)，即触发一次后就会立即失效。

1
2
3
4

[zk: localhost:2181(CONNECTED) 4] get /hadoop -w
[zk: localhost:2181(CONNECTED) 5] set /hadoop 45678
WATCHER::
WatchedEvent state:SyncConnected type:NodeDataChanged path:/hadoop  #节点值改变

get path [watch] 在当前版本已废弃

stat path

使用 stat path -w 注册的监听器能够在节点状态发生改变的时候，向客户端发出通知。

1
2
3
4

[zk: localhost:2181(CONNECTED) 7] stat path -w
[zk: localhost:2181(CONNECTED) 8] set /hadoop 112233
WATCHER::
WatchedEvent state:SyncConnected type:NodeDataChanged path:/hadoop  #节点值改变

stat path [watch] 在当前版本已废弃

ls\ls2 path

使用 ls path -w 或 ls2 path -w 注册的监听器能够监听该节点下所有子节点的增加和删除操作。

1
2
3
4
5

[zk: localhost:2181(CONNECTED) 9] ls /hadoop -w
[]
[zk: localhost:2181(CONNECTED) 10] create  /hadoop/yarn "aaa"
WATCHER::
WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/hadoop

ls path [watch] 和 ls2 path [watch] 在当前版本已废弃

辅助命令

使用 help 可以查看所有命令帮助信息。

使用 history 可以查看最近 10 条历史记录。

zookeeper 四字命令

更多四字命令可以参阅官方文档：https://zookeeper.apache.org/doc/current/zookeeperAdmin.html

使用前需要使用 yum install nc 安装 nc 命令，使用示例如下：

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

[root@hadoop001 bin]# echo stat | nc localhost 2181
Zookeeper version: 3.4.13-2d71af4dbe22557fda74f9a9b4309b15a7487f03,
built on 06/29/2018 04:05 GMT
Clients:
 /0:0:0:0:0:0:0:1:50584[1](queued=0,recved=371,sent=371)
 /0:0:0:0:0:0:0:1:50656[0](queued=0,recved=1,sent=0)
Latency min/avg/max: 0/0/19
Received: 372
Sent: 371
Connections: 2
Outstanding: 0
Zxid: 0x150
Mode: standalone
Node count: 167

参考资料

• Zookeeper 客户端基础命令使用

Memcached 快速入门

Memcached 简介

Memcached 是一个自由开源的，高性能，分布式内存对象缓存系统。

Memcached 是一种基于内存的 key-value 存储，用来存储小块的任意数据（字符串、对象）。这些数据可以是数据库调用、API 调用或者是页面渲染的结果。

Memcached 简洁而强大。它的简洁设计便于快速开发，减轻开发难度，解决了大数据量缓存的很多问题。它的 API 兼容大部分流行的开发语言。本质上，它是一个简洁的 key-value 存储系统。

Memcached 特性

memcached 作为高速运行的分布式缓存服务器，具有以下的特点。

• 协议简单
• 基于 libevent 的事件处理
• 内置内存存储方式
• memcached 不互相通信的分布式

Memcached 命令

可以通过 telnet 命令并指定主机 ip 和端口来连接 Memcached 服务。

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

telnet 127.0.0.1 11211

Trying 127.0.0.1...
Connected to 127.0.0.1.
Escape character is '^]'.
set foo 0 0 3                                                   保存命令
bar                                                             数据
STORED                                                          结果
get foo                                                         取得命令
VALUE foo 0 3                                                   数据
bar                                                             数据
END                                                             结束行
quit                                                            退出

Java 连接 Memcached

使用 Java 程序连接 Memcached，需要在你的 classpath 中添加 Memcached jar 包。

本站 jar 包下载地址：spymemcached-2.10.3.jar。

Google Code jar 包下载地址：spymemcached-2.10.3.jar（需要科学上网）。

以下程序假定 Memcached 服务的主机为 127.0.0.1，端口为 11211。

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

import net.spy.memcached.MemcachedClient;
import java.net.*;


public class MemcachedJava {
   public static void main(String[] args) {
      try{
         // 本地连接 Memcached 服务
         MemcachedClient mcc = new MemcachedClient(new InetSocketAddress("127.0.0.1", 11211));
         System.out.println("Connection to server sucessful.");

         // 关闭连接
         mcc.shutdown();

      }catch(Exception ex){
         System.out.println( ex.getMessage() );
      }
   }
}

参考资料

• Memcached 官网
• Memcached Github
• Memcached 教程

Jetty 快速入门

Jetty 简介

jetty 是什么？

jetty 是轻量级的 web 服务器和 servlet 引擎。

它的最大特点是：可以很方便的作为嵌入式服务器。

它是 eclipse 的一个开源项目。不用怀疑，就是你常用的那个 eclipse。

它是使用 Java 开发的，所以天然对 Java 支持良好。

官方网址

github 源码地址

什么是嵌入式服务器？

以 jetty 来说明，就是只要引入 jetty 的 jar 包，可以通过直接调用其 API 的方式来启动 web 服务。

用过 Tomcat、Resin 等服务器的朋友想必不会陌生那一套安装、配置、部署的流程吧，还是挺繁琐的。使用 jetty，就不需要这些过程了。

jetty 非常适用于项目的开发、测试，因为非常快捷。如果想用于生产环境，则需要谨慎考虑，它不一定能像成熟的 Tomcat、Resin 等服务器一样支持企业级 Java EE 的需要。

Jetty 的使用

我觉得嵌入式启动方式的一个好处在于：可以直接运行项目，无需每次部署都得再配置服务器。

jetty 的嵌入式启动使用有两种方式：

API 方式

maven 插件方式

API 方式

添加 maven 依赖

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

<dependency>
  <groupId>org.eclipse.jetty</groupId>
  <artifactId>jetty-webapp</artifactId>
  <version>9.3.2.v20150730</version>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.eclipse.jetty</groupId>
  <artifactId>jetty-annotations</artifactId>
  <version>9.3.2.v20150730</version>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.eclipse.jetty</groupId>
  <artifactId>apache-jsp</artifactId>
  <version>9.3.2.v20150730</version>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.eclipse.jetty</groupId>
  <artifactId>apache-jstl</artifactId>
  <version>9.3.2.v20150730</version>
  <scope>test</scope>
</dependency>

官方的启动代码

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

public class SplitFileServer
{
    public static void main( String[] args ) throws Exception
    {
        // 创建Server对象，并绑定端口
        Server server = new Server();
        ServerConnector connector = new ServerConnector(server);
        connector.setPort(8090);
        server.setConnectors(new Connector[] { connector });

        // 创建上下文句柄，绑定上下文路径。这样启动后的url就会是:http://host:port/context
        ResourceHandler rh0 = new ResourceHandler();
        ContextHandler context0 = new ContextHandler();
        context0.setContextPath("/");

        // 绑定测试资源目录（在本例的配置目录dir0的路径是src/test/resources/dir0）
        File dir0 = MavenTestingUtils.getTestResourceDir("dir0");
        context0.setBaseResource(Resource.newResource(dir0));
        context0.setHandler(rh0);

        // 和上面的例子一样
        ResourceHandler rh1 = new ResourceHandler();
        ContextHandler context1 = new ContextHandler();
        context1.setContextPath("/");
        File dir1 = MavenTestingUtils.getTestResourceDir("dir1");
        context1.setBaseResource(Resource.newResource(dir1));
        context1.setHandler(rh1);

        // 绑定两个资源句柄
        ContextHandlerCollection contexts = new ContextHandlerCollection();
        contexts.setHandlers(new Handler[] { context0, context1 });
        server.setHandler(contexts);

        // 启动
        server.start();

        // 打印dump时的信息
        System.out.println(server.dump());

        // join当前线程
        server.join();
    }
}

直接运行 Main 方法，就可以启动 web 服务。

注：以上代码在 eclipse 中运行没有问题，如果想在 Intellij 中运行还需要为它指定配置文件。

如果想了解在 Eclipse 和 Intellij 都能运行的通用方法可以参考我的 github 代码示例。

我的实现也是参考 springside 的方式。

代码行数有点多，不在这里贴代码了。

完整参考代码

Maven 插件方式

如果你熟悉 maven，那么实在太简单了

注： Maven 版本必须在 3.3 及以上版本。

(1) 添加 maven 插件

1
2
3
4
5

<plugin>
  <groupId>org.eclipse.jetty</groupId>
  <artifactId>jetty-maven-plugin</artifactId>
  <version>9.3.12.v20160915</version>
</plugin>

(2) 执行 maven 命令：

1

mvn jetty:run

讲真，就是这么简单。jetty 默认会为你创建一个 web 服务，地址为 127.0.0.1:8080。

当然，你也可以在插件中配置你的 webapp 环境

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

<plugin>
  <groupId>org.eclipse.jetty</groupId>
  <artifactId>jetty-maven-plugin</artifactId>
  <version>9.3.12.v20160915</version>

  <configuration>
 <webAppSourceDirectory>${project.basedir}/src/staticfiles</webAppSourceDirectory>

    <!-- 配置webapp -->
 <webApp>
   <contextPath>/</contextPath>
   <descriptor>${project.basedir}/src/over/here/web.xml</descriptor>
   <jettyEnvXml>${project.basedir}/src/over/here/jetty-env.xml</jettyEnvXml>
 </webApp>

    <!-- 配置classes -->
 <classesDirectory>${project.basedir}/somewhere/else</classesDirectory>
 <scanClassesPattern>
   <excludes>
  <exclude>**/Foo.class</exclude>
   </excludes>
 </scanClassesPattern>
 <scanTargets>
   <scanTarget>src/mydir</scanTarget>
   <scanTarget>src/myfile.txt</scanTarget>
 </scanTargets>

    <!-- 扫描target目录下的资源文件 -->
 <scanTargetPatterns>
   <scanTargetPattern>
 <directory>src/other-resources</directory>
 <includes>
   <include>**/*.xml</include>
   <include>**/*.properties</include>
 </includes>
 <excludes>
   <exclude>**/myspecial.xml</exclude>
   <exclude>**/myspecial.properties</exclude>
 </excludes>
   </scanTargetPattern>
 </scanTargetPatterns>
  </configuration>
</plugin>

官方给的 jetty-env.xml 范例

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

<?xml version="1.0"?>
<!DOCTYPE Configure PUBLIC "-//Mort Bay Consulting//DTD Configure//EN" "http://jetty.mortbay.org/configure.dtd">

<Configure class="org.eclipse.jetty.webapp.WebAppContext">

  <!-- Add an EnvEntry only valid for this webapp               -->
  <New id="gargle"  class="org.eclipse.jetty.plus.jndi.EnvEntry">
    <Arg>gargle</Arg>
    <Arg type="java.lang.Double">100</Arg>
    <Arg type="boolean">true</Arg>
  </New>

 <!-- Add an override for a global EnvEntry                           -->
  <New id="wiggle"  class="org.eclipse.jetty.plus.jndi.EnvEntry">
    <Arg>wiggle</Arg>
    <Arg type="java.lang.Double">55.0</Arg>
    <Arg type="boolean">true</Arg>
  </New>

  <!-- an XADataSource                                                -->
  <New id="mydatasource99" class="org.eclipse.jetty.plus.jndi.Resource">
    <Arg>jdbc/mydatasource99</Arg>
    <Arg>
      <New class="com.atomikos.jdbc.SimpleDataSourceBean">
        <Set name="xaDataSourceClassName">org.apache.derby.jdbc.EmbeddedXADataSource</Set>
        <Set name="xaDataSourceProperties">databaseName=testdb99;createDatabase=create</Set>
        <Set name="UniqueResourceName">mydatasource99</Set>
      </New>
    </Arg>
  </New>

</Configure>

Jetty 的架构

Jetty 架构简介

Jetty Server 就是由多个 Connector（连接器）、多个 Handler（处理器），以及一个线程池组成。

跟 Tomcat 一样，Jetty 也有 HTTP 服务器和 Servlet 容器的功能，因此 Jetty 中的 Connector 组件和 Handler 组件分别来实现这两个功能，而这两个组件工作时所需要的线程资源都直接从一个全局线程池 ThreadPool 中获取。

Jetty Server 可以有多个 Connector 在不同的端口上监听客户请求，而对于请求处理的 Handler 组件，也可以根据具体场景使用不同的 Handler。这样的设计提高了 Jetty 的灵活性，需要支持 Servlet，则可以使用 ServletHandler；需要支持 Session，则再增加一个 SessionHandler。也就是说我们可以不使用 Servlet 或者 Session，只要不配置这个 Handler 就行了。

为了启动和协调上面的核心组件工作，Jetty 提供了一个 Server 类来做这个事情，它负责创建并初始化 Connector、Handler、ThreadPool 组件，然后调用 start 方法启动它们。

Jetty 和 Tomcat 架构区别

对比一下 Tomcat 的整体架构图，你会发现 Tomcat 在整体上跟 Jetty 很相似，它们的第一个区别是 Jetty 中没有 Service 的概念，Tomcat 中的 Service 包装了多个连接器和一个容器组件，一个 Tomcat 实例可以配置多个 Service，不同的 Service 通过不同的连接器监听不同的端口；而 Jetty 中 Connector 是被所有 Handler 共享的。

第二个区别是，在 Tomcat 中每个连接器都有自己的线程池，而在 Jetty 中所有的 Connector 共享一个全局的线程池。

Connector 组件

跟 Tomcat 一样，Connector 的主要功能是对 I/O 模型和应用层协议的封装。I/O 模型方面，最新的 Jetty 9 版本只支持 NIO，因此 Jetty 的 Connector 设计有明显的 Java NIO 通信模型的痕迹。至于应用层协议方面，跟 Tomcat 的 Processor 一样，Jetty 抽象出了 Connection 组件来封装应用层协议的差异。

服务端在 NIO 通信上主要完成了三件事情：监听连接、I/O 事件查询以及数据读写。因此 Jetty 设计了Acceptor、SelectorManager 和 Connection 来分别做这三件事情

Acceptor

Acceptor 用于接受请求。跟 Tomcat 一样，Jetty 也有独立的 Acceptor 线程组用于处理连接请求。在 Connector 的实现类 ServerConnector 中，有一个 _acceptors 的数组，在 Connector 启动的时候, 会根据 _acceptors 数组的长度创建对应数量的 Acceptor，而 Acceptor 的个数可以配置。

1
2
3
4
5

for (int i = 0; i < _acceptors.length; i++)
{
  Acceptor a = new Acceptor(i);
  getExecutor().execute(a);
}

Acceptor 是 ServerConnector 中的一个内部类，同时也是一个 Runnable，Acceptor 线程是通过 getExecutor() 得到的线程池来执行的，前面提到这是一个全局的线程池。

Acceptor 通过阻塞的方式来接受连接，这一点跟 Tomcat 也是一样的。

1
2
3
4
5
6
7
8
9
10
11

public void accept(int acceptorID) throws IOException
{
  ServerSocketChannel serverChannel = _acceptChannel;
  if (serverChannel != null && serverChannel.isOpen())
  {
    // 这里是阻塞的
    SocketChannel channel = serverChannel.accept();
    // 执行到这里时说明有请求进来了
    accepted(channel);
  }
}

接受连接成功后会调用 accepted() 函数，accepted() 函数中会将 SocketChannel 设置为非阻塞模式，然后交给 Selector 去处理，因此这也就到了 Selector 的地界了。

1
2
3
4
5
6
7
8

private void accepted(SocketChannel channel) throws IOException
{
    channel.configureBlocking(false);
    Socket socket = channel.socket();
    configure(socket);
    // _manager 是 SelectorManager 实例，里面管理了所有的 Selector 实例
    _manager.accept(channel);
}

SelectorManager

Jetty 的 Selector 由 SelectorManager 类管理，而被管理的 Selector 叫作 ManagedSelector。SelectorManager 内部有一个 ManagedSelector 数组，真正干活的是 ManagedSelector。咱们接着上面分析，看看在 SelectorManager 在 accept 方法里做了什么。

1
2
3
4
5
6
7

public void accept(SelectableChannel channel, Object attachment)
{
  // 选择一个 ManagedSelector 来处理 Channel
  final ManagedSelector selector = chooseSelector();
  // 提交一个任务 Accept 给 ManagedSelector
  selector.submit(selector.new Accept(channel, attachment));
}

SelectorManager 从本身的 Selector 数组中选择一个 Selector 来处理这个 Channel，并创建一个任务 Accept 交给 ManagedSelector，ManagedSelector 在处理这个任务主要做了两步：

第一步，调用 Selector 的 register 方法把 Channel 注册到 Selector 上，拿到一个 SelectionKey。

1

_key = _channel.register(selector, SelectionKey.OP_ACCEPT, this);

第二步，创建一个 EndPoint 和 Connection，并跟这个 SelectionKey（Channel）绑在一起：

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

private void createEndPoint(SelectableChannel channel, SelectionKey selectionKey) throws IOException
{
    //1. 创建 Endpoint
    EndPoint endPoint = _selectorManager.newEndPoint(channel, this, selectionKey);

    //2. 创建 Connection
    Connection connection = _selectorManager.newConnection(channel, endPoint, selectionKey.attachment());

    //3. 把 Endpoint、Connection 和 SelectionKey 绑在一起
    endPoint.setConnection(connection);
    selectionKey.attach(endPoint);

}

这里需要你特别注意的是，ManagedSelector 并没有直接调用 EndPoint 的方法去处理数据，而是通过调用 EndPoint 的方法返回一个 Runnable，然后把这个 Runnable 扔给线程池执行，所以你能猜到，这个 Runnable 才会去真正读数据和处理请求。

Connection

这个 Runnable 是 EndPoint 的一个内部类，它会调用 Connection 的回调方法来处理请求。Jetty 的 Connection 组件类比就是 Tomcat 的 Processor，负责具体协议的解析，得到 Request 对象，并调用 Handler 容器进行处理。下面我简单介绍一下它的具体实现类 HttpConnection 对请求和响应的处理过程。

请求处理：HttpConnection 并不会主动向 EndPoint 读取数据，而是向在 EndPoint 中注册一堆回调方法：

1

getEndPoint().fillInterested(_readCallback);

这段代码就是告诉 EndPoint，数据到了你就调我这些回调方法 _readCallback 吧，有点异步 I/O 的感觉，也就是说 Jetty 在应用层面模拟了异步 I/O 模型。

而在回调方法 _readCallback 里，会调用 EndPoint 的接口去读数据，读完后让 HTTP 解析器去解析字节流，HTTP 解析器会将解析后的数据，包括请求行、请求头相关信息存到 Request 对象里。

响应处理：Connection 调用 Handler 进行业务处理，Handler 会通过 Response 对象来操作响应流，向流里面写入数据，HttpConnection 再通过 EndPoint 把数据写到 Channel，这样一次响应就完成了。

到此你应该了解了 Connector 的工作原理，下面我画张图再来回顾一下 Connector 的工作流程。

1. Acceptor 监听连接请求，当有连接请求到达时就接受连接，一个连接对应一个 Channel，Acceptor 将 Channel 交给 ManagedSelector 来处理。

2. ManagedSelector 把 Channel 注册到 Selector 上，并创建一个 EndPoint 和 Connection 跟这个 Channel 绑定，接着就不断地检测 I/O 事件。

3. I/O 事件到了就调用 EndPoint 的方法拿到一个 Runnable，并扔给线程池执行。

4. 线程池中调度某个线程执行 Runnable。

5. Runnable 执行时，调用回调函数，这个回调函数是 Connection 注册到 EndPoint 中的。

6. 回调函数内部实现，其实就是调用 EndPoint 的接口方法来读数据。

7. Connection 解析读到的数据，生成请求对象并交给 Handler 组件去处理。

Handler 组件

Jetty 的 Handler 设计是它的一大特色，Jetty 本质就是一个 Handler 管理器，Jetty 本身就提供了一些默认 Handler 来实现 Servlet 容器的功能，你也可以定义自己的 Handler 来添加到 Jetty 中，这体现了“微内核 + 插件”的设计思想。

Handler 就是一个接口，它有一堆实现类，Jetty 的 Connector 组件调用这些接口来处理 Servlet 请求。

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

public interface Handler extends LifeCycle, Destroyable
{
    // 处理请求的方法
    public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response)
        throws IOException, ServletException;

    // 每个 Handler 都关联一个 Server 组件，被 Server 管理
    public void setServer(Server server);
    public Server getServer();

    // 销毁方法相关的资源
    public void destroy();
}

方法说明：

• Handler 的 handle 方法跟 Tomcat 容器组件的 service 方法一样，它有 ServletRequest 和 ServeletResponse 两个参数。
• 因为任何一个 Handler 都需要关联一个 Server 组件，Handler 需要被 Server 组件来管理。Handler 通过 setServer 和 getServer 方法绑定 Server。
• Handler 会加载一些资源到内存，因此通过设置 destroy 方法来销毁。

Handler 继承关系

Handler 只是一个接口，完成具体功能的还是它的子类。那么 Handler 有哪些子类呢？它们的继承关系又是怎样的？这些子类是如何实现 Servlet 容器功能的呢？

在 AbstractHandler 之下有 AbstractHandlerContainer，为什么需要这个类呢？这其实是个过渡，为了实现链式调用，一个 Handler 内部必然要有其他 Handler 的引用，所以这个类的名字里才有 Container，意思就是这样的 Handler 里包含了其他 Handler 的引用。

HandlerWrapper 和 HandlerCollection 都是 Handler，但是这些 Handler 里还包括其他 Handler 的引用。不同的是，HandlerWrapper 只包含一个其他 Handler 的引用，而 HandlerCollection 中有一个 Handler 数组的引用。

HandlerWrapper 有两个子类：Server 和 ScopedHandler。

• Server 比较好理解，它本身是 Handler 模块的入口，必然要将请求传递给其他 Handler 来处理，为了触发其他 Handler 的调用，所以它是一个 HandlerWrapper。
• ScopedHandler 也是一个比较重要的 Handler，实现了“具有上下文信息”的责任链调用。为什么我要强调“具有上下文信息”呢？那是因为 Servlet 规范规定 Servlet 在执行过程中是有上下文的。那么这些 Handler 在执行过程中如何访问这个上下文呢？这个上下文又存在什么地方呢？答案就是通过 ScopedHandler 来实现的。

HandlerCollection 其实维护了一个 Handler 数组。这是为了同时支持多个 Web 应用，如果每个 Web 应用有一个 Handler 入口，那么多个 Web 应用的 Handler 就成了一个数组，比如 Server 中就有一个 HandlerCollection，Server 会根据用户请求的 URL 从数组中选取相应的 Handler 来处理，就是选择特定的 Web 应用来处理请求。

Handler 可以分成三种类型：

• 第一种是协调 Handler，这种 Handler 负责将请求路由到一组 Handler 中去，比如 HandlerCollection，它内部持有一个 Handler 数组，当请求到来时，它负责将请求转发到数组中的某一个 Handler。
• 第二种是过滤器 Handler，这种 Handler 自己会处理请求，处理完了后再把请求转发到下一个 Handler，比如图上的 HandlerWrapper，它内部持有下一个 Handler 的引用。需要注意的是，所有继承了 HandlerWrapper 的 Handler 都具有了过滤器 Handler 的特征，比如 ContextHandler、SessionHandler 和 WebAppContext 等。
• 第三种是内容 Handler，说白了就是这些 Handler 会真正调用 Servlet 来处理请求，生成响应的内容，比如 ServletHandler。如果浏览器请求的是一个静态资源，也有相应的 ResourceHandler 来处理这个请求，返回静态页面。

实现 Servlet 规范

ServletHandler、ContextHandler 以及 WebAppContext 等，它们实现了 Servlet 规范。

Servlet 规范中有 Context、Servlet、Filter、Listener 和 Session 等，Jetty 要支持 Servlet 规范，就需要有相应的 Handler 来分别实现这些功能。因此，Jetty 设计了 3 个组件：ContextHandler、ServletHandler 和 SessionHandler 来实现 Servle 规范中规定的功能，而WebAppContext 本身就是一个 ContextHandler，另外它还负责管理 ServletHandler 和 SessionHandler。

ContextHandler 会创建并初始化 Servlet 规范里的 ServletContext 对象，同时 ContextHandler 还包含了一组能够让你的 Web 应用运行起来的 Handler，可以这样理解，Context 本身也是一种 Handler，它里面包含了其他的 Handler，这些 Handler 能处理某个特定 URL 下的请求。比如，ContextHandler 包含了一个或者多个 ServletHandler。

ServletHandler 实现了 Servlet 规范中的 Servlet、Filter 和 Listener 的功能。ServletHandler 依赖 FilterHolder、ServletHolder、ServletMapping、FilterMapping 这四大组件。FilterHolder 和 ServletHolder 分别是 Filter 和 Servlet 的包装类，每一个 Servlet 与路径的映射会被封装成 ServletMapping，而 Filter 与拦截 URL 的映射会被封装成 FilterMapping。

SessionHandler 用来管理 Session。除此之外 WebAppContext 还有一些通用功能的 Handler，比如 SecurityHandler 和 GzipHandler，同样从名字可以知道这些 Handler 的功能分别是安全控制和压缩 / 解压缩。

WebAppContext 会将这些 Handler 构建成一个执行链，通过这个链会最终调用到我们的业务 Servlet。

Jetty 的线程策略

传统 Selector 编程模型

常规的 NIO 编程思路是，将 I/O 事件的侦测和请求的处理分别用不同的线程处理。具体过程是：

启动一个线程，在一个死循环里不断地调用 select 方法，检测 Channel 的 I/O 状态，一旦 I/O 事件达到，比如数据就绪，就把该 I/O 事件以及一些数据包装成一个 Runnable，将 Runnable 放到新线程中去处理。

在这个过程中按照职责划分，有两个线程在干活，一个是 I/O 事件检测线程，另一个是 I/O 事件处理线程。这样的好处是它们互不干扰和阻塞对方。

Jetty 的 Selector 编程模型

将 I/O 事件检测和业务处理这两种工作分开的思路也有缺点：当 Selector 检测读就绪事件时，数据已经被拷贝到内核中的缓存了，同时 CPU 的缓存中也有这些数据了，我们知道 CPU 本身的缓存比内存快多了，这时当应用程序去读取这些数据时，如果用另一个线程去读，很有可能这个读线程使用另一个 CPU 核，而不是之前那个检测数据就绪的 CPU 核，这样 CPU 缓存中的数据就用不上了，并且线程切换也需要开销。

因此 Jetty 的 Connector 做了一个大胆尝试，那就是把 I/O 事件的生产和消费放到同一个线程来处理，如果这两个任务由同一个线程来执行，如果执行过程中线程不阻塞，操作系统会用同一个 CPU 核来执行这两个任务，这样就能利用 CPU 缓存了。

ManagedSelector

ManagedSelector 的本质就是一个 Selector，负责 I/O 事件的检测和分发。为了方便使用，Jetty 在 Java 原生的 Selector 上做了一些扩展，就变成了 ManagedSelector，我们先来看看它有哪些成员变量：

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

public class ManagedSelector extends ContainerLifeCycle implements Dumpable
{
    // 原子变量，表明当前的 ManagedSelector 是否已经启动
    private final AtomicBoolean _started = new AtomicBoolean(false);

    // 表明是否阻塞在 select 调用上
    private boolean _selecting = false;

    // 管理器的引用，SelectorManager 管理若干 ManagedSelector 的生命周期
    private final SelectorManager _selectorManager;

    //ManagedSelector 不止一个，为它们每人分配一个 id
    private final int _id;

    // 关键的执行策略，生产者和消费者是否在同一个线程处理由它决定
    private final ExecutionStrategy _strategy;

    //Java 原生的 Selector
    private Selector _selector;

    //"Selector 更新任务 " 队列
    private Deque<SelectorUpdate> _updates = new ArrayDeque<>();
    private Deque<SelectorUpdate> _updateable = new ArrayDeque<>();

    ...
}

这些成员变量中其他的都好理解，就是“Selector 更新任务”队列_updates和执行策略_strategy可能不是很直观。

SelectorUpdate 接口

为什么需要一个“Selector 更新任务”队列呢，对于 Selector 的用户来说，我们对 Selector 的操作无非是将 Channel 注册到 Selector 或者告诉 Selector 我对什么 I/O 事件感兴趣，那么这些操作其实就是对 Selector 状态的更新，Jetty 把这些操作抽象成 SelectorUpdate 接口。

1
2
3
4
5
6
7

/**
 * A selector update to be done when the selector has been woken.
 */
public interface SelectorUpdate
{
    void update(Selector selector);
}

这意味着如果你不能直接操作 ManageSelector 中的 Selector，而是需要向 ManagedSelector 提交一个任务类，这个类需要实现 SelectorUpdate 接口 update 方法，在 update 方法里定义你想要对 ManagedSelector 做的操作。

比如 Connector 中 Endpoint 组件对读就绪事件感兴趣，它就向 ManagedSelector 提交了一个内部任务类 ManagedSelector.SelectorUpdate：

1

_selector.submit(_updateKeyAction);

这个_updateKeyAction就是一个 SelectorUpdate 实例，它的 update 方法实现如下：

1
2
3
4
5
6
7
8
9

private final ManagedSelector.SelectorUpdate _updateKeyAction = new ManagedSelector.SelectorUpdate()
{
    @Override
    public void update(Selector selector)
    {
        // 这里的 updateKey 其实就是调用了 SelectionKey.interestOps(OP_READ);
        updateKey();
    }
};

我们看到在 update 方法里，调用了 SelectionKey 类的 interestOps 方法，传入的参数是OP_READ，意思是现在我对这个 Channel 上的读就绪事件感兴趣了。

那谁来负责执行这些 update 方法呢，答案是 ManagedSelector 自己，它在一个死循环里拉取这些 SelectorUpdate 任务类逐个执行。

Selectable 接口

那 I/O 事件到达时，ManagedSelector 怎么知道应该调哪个函数来处理呢？其实也是通过一个任务类接口，这个接口就是 Selectable，它返回一个 Runnable，这个 Runnable 其实就是 I/O 事件就绪时相应的处理逻辑。

1
2
3
4
5
6
7
8

public interface Selectable
{
    // 当某一个 Channel 的 I/O 事件就绪后，ManagedSelector 会调用的回调函数
    Runnable onSelected();

    // 当所有事件处理完了之后 ManagedSelector 会调的回调函数，我们先忽略。
    void updateKey();
}

ManagedSelector 在检测到某个 Channel 上的 I/O 事件就绪时，也就是说这个 Channel 被选中了，ManagedSelector 调用这个 Channel 所绑定的附件类的 onSelected 方法来拿到一个 Runnable。

这句话有点绕，其实就是 ManagedSelector 的使用者，比如 Endpoint 组件在向 ManagedSelector 注册读就绪事件时，同时也要告诉 ManagedSelector 在事件就绪时执行什么任务，具体来说就是传入一个附件类，这个附件类需要实现 Selectable 接口。ManagedSelector 通过调用这个 onSelected 拿到一个 Runnable，然后把 Runnable 扔给线程池去执行。

那 Endpoint 的 onSelected 是如何实现的呢？

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

@Override
public Runnable onSelected()
{
    int readyOps = _key.readyOps();

    boolean fillable = (readyOps & SelectionKey.OP_READ) != 0;
    boolean flushable = (readyOps & SelectionKey.OP_WRITE) != 0;

    // return task to complete the job
    Runnable task= fillable
            ? (flushable
                    ? _runCompleteWriteFillable
                    : _runFillable)
            : (flushable
                    ? _runCompleteWrite
                    : null);

    return task;
}

上面的代码逻辑很简单，就是读事件到了就读，写事件到了就写。

ExecutionStrategy

铺垫了这么多，终于要上主菜了。前面我主要介绍了 ManagedSelector 的使用者如何跟 ManagedSelector 交互，也就是如何注册 Channel 以及 I/O 事件，提供什么样的处理类来处理 I/O 事件，接下来我们来看看 ManagedSelector 是如何统一管理和维护用户注册的 Channel 集合。再回到今天开始的讨论，ManagedSelector 将 I/O 事件的生产和消费看作是生产者消费者模式，为了充分利用 CPU 缓存，生产和消费尽量放到同一个线程处理，那这是如何实现的呢？Jetty 定义了 ExecutionStrategy 接口：

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

public interface ExecutionStrategy
{
    // 只在 HTTP2 中用到，简单起见，我们先忽略这个方法。
    public void dispatch();

    // 实现具体执行策略，任务生产出来后可能由当前线程执行，也可能由新线程来执行
    public void produce();

    // 任务的生产委托给 Producer 内部接口，
    public interface Producer
    {
        // 生产一个 Runnable(任务)
        Runnable produce();
    }
}

我们看到 ExecutionStrategy 接口比较简单，它将具体任务的生产委托内部接口 Producer，而在自己的 produce 方法里来实现具体执行逻辑，也就是生产出来的任务要么由当前线程执行，要么放到新线程中执行。Jetty 提供了一些具体策略实现类：ProduceConsume、ProduceExecuteConsume、ExecuteProduceConsume 和 EatWhatYouKill。它们的区别是：

• ProduceConsume：任务生产者自己依次生产和执行任务，对应到 NIO 通信模型就是用一个线程来侦测和处理一个 ManagedSelector 上所有的 I/O 事件，后面的 I/O 事件要等待前面的 I/O 事件处理完，效率明显不高。通过图来理解，图中绿色表示生产一个任务，蓝色表示执行这个任务。

• ProduceExecuteConsume：任务生产者开启新线程来运行任务，这是典型的 I/O 事件侦测和处理用不同的线程来处理，缺点是不能利用 CPU 缓存，并且线程切换成本高。同样我们通过一张图来理解，图中的棕色表示线程切换。

• ExecuteProduceConsume：任务生产者自己运行任务，但是该策略可能会新建一个新线程以继续生产和执行任务。这种策略也被称为“吃掉你杀的猎物”，它来自狩猎伦理，认为一个人不应该杀死他不吃掉的东西，对应线程来说，不应该生成自己不打算运行的任务。它的优点是能利用 CPU 缓存，但是潜在的问题是如果处理 I/O 事件的业务代码执行时间过长，会导致线程大量阻塞和线程饥饿。

• EatWhatYouKill：这是 Jetty 对 ExecuteProduceConsume 策略的改良，在线程池线程充足的情况下等同于 ExecuteProduceConsume；当系统比较忙线程不够时，切换成 ProduceExecuteConsume 策略。为什么要这么做呢，原因是 ExecuteProduceConsume 是在同一线程执行 I/O 事件的生产和消费，它使用的线程来自 Jetty 全局的线程池，这些线程有可能被业务代码阻塞，如果阻塞得多了，全局线程池中的线程自然就不够用了，最坏的情况是连 I/O 事件的侦测都没有线程可用了，会导致 Connector 拒绝浏览器请求。于是 Jetty 做了一个优化，在低线程情况下，就执行 ProduceExecuteConsume 策略，I/O 侦测用专门的线程处理，I/O 事件的处理扔给线程池处理，其实就是放到线程池的队列里慢慢处理。

分析了这几种线程策略，我们再来看看 Jetty 是如何实现 ExecutionStrategy 接口的。答案其实就是实现 produce 接口生产任务，一旦任务生产出来，ExecutionStrategy 会负责执行这个任务。

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

private class SelectorProducer implements ExecutionStrategy.Producer
{
    private Set<SelectionKey> _keys = Collections.emptySet();
    private Iterator<SelectionKey> _cursor = Collections.emptyIterator();

    @Override
    public Runnable produce()
    {
        while (true)
        {
            // 如何 Channel 集合中有 I/O 事件就绪，调用前面提到的 Selectable 接口获取 Runnable, 直接返回给 ExecutionStrategy 去处理
            Runnable task = processSelected();
            if (task != null)
                return task;

           // 如果没有 I/O 事件就绪，就干点杂活，看看有没有客户提交了更新 Selector 的任务，就是上面提到的 SelectorUpdate 任务类。
            processUpdates();
            updateKeys();

           // 继续执行 select 方法，侦测 I/O 就绪事件
            if (!select())
                return null;
        }
    }
 }

SelectorProducer 是 ManagedSelector 的内部类，SelectorProducer 实现了 ExecutionStrategy 中的 Producer 接口中的 produce 方法，需要向 ExecutionStrategy 返回一个 Runnable。在这个方法里 SelectorProducer 主要干了三件事情

1. 如果 Channel 集合中有 I/O 事件就绪，调用前面提到的 Selectable 接口获取 Runnable，直接返回给 ExecutionStrategy 去处理。
2. 如果没有 I/O 事件就绪，就干点杂活，看看有没有客户提交了更新 Selector 上事件注册的任务，也就是上面提到的 SelectorUpdate 任务类。
3. 干完杂活继续执行 select 方法，侦测 I/O 就绪事件。

参考资料

• Jetty 官方网址
• Jetty Github
• Jetty wiki

Java 和 JSON 序列化

JSON（JavaScript Object Notation）是一种基于文本的数据交换格式。几乎所有的编程语言都有很好的库或第三方工具来提供基于 JSON 的 API 支持，因此你可以非常方便地使用任何自己喜欢的编程语言来处理 JSON 数据。

本文主要从 Java 语言的角度来讲解 JSON 的应用。

JSON 简介

JSON 是什么

JSON 起源于 1999 年的 JS 语言规范 ECMA262 的一个子集（即 15.12 章节描述了格式与解析），后来 2003 年作为一个数据格式ECMA404（很囧的序号有不有？）发布。
2006 年，作为 rfc4627 发布，这时规范增加到 18 页，去掉没用的部分，十页不到。

JSON 的应用很广泛，这里有超过 100 种语言下的 JSON 库：json.org。

更多的可以参考这里，关于 json 的一切。

JSON 标准

这估计是最简单标准规范之一：

• 只有两种结构：对象内的键值对集合结构和数组，对象用 {} 表示、内部是 "key":"value"，数组用 [] 表示，不同值用逗号分开
• 基本数值有 7 个： false / null / true / object / array / number / string
• 再加上结构可以嵌套，进而可以用来表达复杂的数据
• 一个简单实例：

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

{
  "Image": {
    "Width": 800,
    "Height": 600,
    "Title": "View from 15th Floor",
    "Thumbnail": {
      "Url": "http://www.example.com/image/481989943",
      "Height": 125,
      "Width": "100"
    },
    "IDs": [116, 943, 234, 38793]
  }
}

扩展阅读：

• http://www.json.org/json-zh.html - 图文并茂介绍 json 数据形式

• json 的 RFC 文档

JSON 优缺点

优点：

• 基于纯文本，所以对于人类阅读是很友好的。
• 规范简单，所以容易处理，开箱即用，特别是 JS 类的 ECMA 脚本里是内建支持的，可以直接作为对象使用。
• 平台无关性，因为类型和结构都是平台无关的，而且好处理，容易实现不同语言的处理类库，可以作为多个不同异构系统之间的数据传输格式协议，特别是在 HTTP/REST 下的数据格式。

缺点：

• 性能一般，文本表示的数据一般来说比二进制大得多，在数据传输上和解析处理上都要更影响性能。
• 缺乏 schema，跟同是文本数据格式的 XML 比，在类型的严格性和丰富性上要差很多。XML 可以借由 XSD 或 DTD 来定义复杂的格式，并由此来验证 XML 文档是否符合格式要求，甚至进一步的，可以基于 XSD 来生成具体语言的操作代码，例如 apache xmlbeans。并且这些工具组合到一起，形成一套庞大的生态，例如基于 XML 可以实现 SOAP 和 WSDL，一系列的 ws-*规范。但是我们也可以看到 JSON 在缺乏规范的情况下，实际上有更大一些的灵活性，特别是近年来 REST 的快速发展，已经有一些 schema 相关的发展(例如理解 JSON Schema，使用 JSON Schema， 在线 schema 测试)，也有类似于 WSDL 的WADL出现。

JSON 工具

• 使用 JSON 实现 RPC（类似 XML-RPC）：JSON-RPC

• 使用 JSON 实现 path 查询操作（类似 XML-PATH）：JsonPATH

• 在线查询工具：JsonPATH

• 格式化工具：jsbeautifier

• chrome 插件：5 个 Json View 插件

• 在线 Mock: 在线 mock

• 其他 Mock：SoapUI可以支持，SwaggerUI 也可以，RestMock也可以。

Java JSON 库

Java 中比较流行的 JSON 库有：

• Fastjson - 阿里巴巴开发的 JSON 库，性能十分优秀。
• Jackson - 社区十分活跃且更新速度很快。Spring 框架默认 JSON 库。
• Gson - 谷歌开发的 JSON 库，目前功能最全的 JSON 库 。

从性能上来看，一般情况下：Fastjson > Jackson > Gson

JSON 编码指南

遵循好的设计与编码风格，能提前解决 80%的问题，个人推荐 Google JSON 风格指南。

• 英文版Google JSON Style Guide：https://google.github.io/styleguide/jsoncstyleguide.xml
• 中文版Google JSON 风格指南：https://github.com/darcyliu/google-styleguide/blob/master/JSONStyleGuide.md

简单摘录如下：

• 属性名和值都是用双引号，不要把注释写到对象里面，对象数据要简洁
• 不要随意结构化分组对象，推荐是用扁平化方式，层次不要太复杂
• 命名方式要有意义，比如单复数表示
• 驼峰式命名，遵循 Bean 规范
• 使用版本来控制变更冲突
• 对于一些关键字，不要拿来做 key
• 如果一个属性是可选的或者包含空值或 null 值，考虑从 JSON 中去掉该属性，除非它的存在有很强的语义原因
• 序列化枚举类型时，使用 name 而不是 value
• 日期要用标准格式处理
• 设计好通用的分页参数
• 设计好异常处理

JSON API与 Google JSON 风格指南有很多可以相互参照之处。

JSON API是数据交互规范，用以定义客户端如何获取与修改资源，以及服务器如何响应对应请求。

JSON API 设计用来最小化请求的数量，以及客户端与服务器间传输的数据量。在高效实现的同时，无需牺牲可读性、灵活性和可发现性。

Fastjson 应用

添加 maven 依赖

1
2
3
4
5

<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>x.x.x</version>
</dependency>

Fastjson API

定义 Bean

Group.java

1
2
3
4
5
6

public class Group {

    private Long       id;
    private String     name;
    private List<User> users = new ArrayList<User>();
}

User.java

1
2
3
4
5

public class User {

    private Long   id;
    private String name;
}

初始化 Bean

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

Group group = new Group();
group.setId(0L);
group.setName("admin");

User guestUser = new User();
guestUser.setId(2L);
guestUser.setName("guest");

User rootUser = new User();
rootUser.setId(3L);
rootUser.setName("root");

group.addUser(guestUser);
group.addUser(rootUser);

序列化

1
2

String jsonString = JSON.toJSONString(group);
System.out.println(jsonString);

反序列化

1

Group bean = JSON.parseObject(jsonString, Group.class);

Fastjson 注解

@JSONField

扩展阅读：更多 API 使用细节可以参考：JSONField 用法，这里介绍基本用法。

可以配置在属性（setter、getter）和字段（必须是 public field）上。

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

@JSONField(name="ID")
public int getId() {return id;}

// 配置date序列化和反序列使用yyyyMMdd日期格式
@JSONField(format="yyyyMMdd")
public Date date1;

// 不序列化
@JSONField(serialize=false)
public Date date2;

// 不反序列化
@JSONField(deserialize=false)
public Date date3;

// 按ordinal排序
@JSONField(ordinal = 2)
private int f1;

@JSONField(ordinal = 1)
private int f2;

@JSONType

• 自定义序列化：ObjectSerializer
• 子类型处理：SeeAlso

JSONType.alphabetic 属性: fastjson 缺省时会使用字母序序列化，如果你是希望按照 java fields/getters 的自然顺序序列化，可以配置 JSONType.alphabetic，使用方法如下：

1
2
3
4
5
6

@JSONType(alphabetic = false)
public static class B {
    public int f2;
    public int f1;
    public int f0;
}

Jackson 应用

扩展阅读：更多 API 使用细节可以参考 jackson-databind 官方说明

添加 maven 依赖

1
2
3
4
5

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.9.8</version>
</dependency>