Trino源码学习-Airlift源码分析

发表于 更新于 分类于 Waline： 阅读次数： 本文字数： 14k 阅读时长 ≈ 23 分钟

Trino 源码学习-Airlift 源码分析

上一篇文章了解了 Trino 的基础 Airlift 框架。本篇将深入 Airlift 源码进行学习，Airlift中的服务都通过guice注入。

从 Bootstrap 开始启动

sequenceDiagram
    Bootstrap->>Bootstrap: 初始化系统,设置应用模块
    Bootstrap->>+Logging: 初始化日志子系统
    Logging-->>-Bootstrap: 初始化日志子系统成功
    Bootstrap->>Bootstrap: 从配置文件和环境变量中加载参数
    Bootstrap->>+ConfigurationFactory: 构建日志子系统配置实例
    ConfigurationFactory->>+ConfigurationMetadata: 获取配置类的元信息
    ConfigurationMetadata-->>-ConfigurationFactory: 返回元信息
    ConfigurationFactory->>ConfigurationFactory: 基于元信息反射创建配置类实例,配置设置对应属性值并验证配置类约束
    ConfigurationFactory-->>-Bootstrap:日志子系统配置实例构建成功
    Bootstrap->>+Logging: 日志子系统加载配置
    Logging-->>-Bootstrap: 加载成功
    Bootstrap->>+ConfigurationFactory: 注册模块中默认配置,默认listener和配置provider。验证注册的配置
    ConfigurationFactory-->>-Bootstrap: 模块配置处理成功
    Bootstrap->>Bootstrap: 打印未使用配置,生效配置
    Bootstrap->>Bootstrap: 合并系统模块，应用模块
    Bootstrap->>+Injector: 创建Guice注入器，注入所有模块
    Injector->>+LifeCycleManager: getInstance
    LifeCycleManager-->>-Bootstrap: 启动应用

从上面的时序图中，可以发现 Bootstrap 的主要工作:

  1. 初始化和配置日志子系统
  2. 应用配置加载和注册模块配置类
  3. 创建 Guice 注入器，注入所有模块
  4. 启动 LifeCycleManager

到此为止，Bootstrap 的工作就做完了。下面来看下 LifeCycleManager

生命周期管理-LifeCycleManager

LifeCycleManager 用于管理所有标记 PostConstruct and PreDestroy 的绑定的生命周期。Manager 本身有下面几种状态:

flowchart LR
    LATENT -->|start| STARTING
    STARTING --> STARTED
    STARTED-->|stop| STOPPING
    STOPPING -->STOPPED
  • 初始化，manager 初始化时会:
    • 被注入所有 PostConstruct and PreDestroy 的绑定
    • 调用所有 PostConstruct 方法
  • start，manager start 时会:
    • 修改自身状态为 STARTING
    • 注册 JVM shutdownhook，当 jvm shutdown 时调用 stop 方法。
    • 修改自身状态为 STARTED
  • stop，manager stop 时会:
    • 修改自身状态为 STOPPING
    • 移除 JVM shutdownhook(否则会递归调用)
    • 调用所有 PreDestroy 方法
    • 修改自身状态为 STOPPED

http

Rest

使用jaxrsBinder()方法绑定的 Resource 都会被绑定到@JaxrsResource上。

然后会被io.airlift.jaxrs.JaxrsModule.JaxRsApplication进一步处理。

Static

使用 httpServerBinder(binder).bindResource() 方法绑定的 Resource 都会被绑定到HttpServerBinder.HttpResourceBinding.class + @TheServlet上。然后会在HttpServerProvider 内用于创建 HttpServer。

HttpServerModule 中的 Servlet 就是io.airlift.jaxrs.JaxrsModule中创建的 ServletContainer。

日志

airlift日志相关的模块有两个:

  • io.airlift.log
  • io.airlift.log-manager

io.airlift.log中只有一个门面类io.airlift.log.Logger, 内部封装了java.util.logging.Logger。所有的日志格式都必须符合String.format()的格式。

io.airlift.log-manager模块用于管理日志后端。通过io.airlift.log.Logging初始化日志子系统。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 系统自带的ROOT Logger
private static final java.util.logging.Logger ROOT = java.util.logging.Logger.getLogger("");
private Logging() {
  ROOT.setLevel(Level.INFO.toJulLevel());
  for (Handler handler: ROOT.getHandlers()) {
    // 去除已有的handler
    ROOT.removeHandler(handler);
  }

  enableConsole();
  log.info("Logging to stderr");
  // 重定向标准输出
  redirectStdStreams();
}

然后通过ROOT.addHandler(handler)的方式增加日志后端handler，分别支持socket和文件。

下面是以Trino为例的日志架构图。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
+------------------------------------------------------------------------------+
|Trino Core                                                                    |
|                                                       +-------------------+  |
|  Trino  +-------> Airlift Log API                     |            Airlift|  |
|  code                    +                            |              Log  |  |
|                          |                            |            Backend|  |
|                          |                            |                   |  |
|                          v                            |                   |  |
|  Third party +--> java.util.logging +------------------>   Handler        |  |
|  code        |                                        |       ^           |  |
|              |                                        |       |           |  |
|              |                                              slf4j-jdk14   |  |
|              |   +jcl-over-slf4j                      |       |           |  |
|              |   |                                            +           |  |
|              +-->+log4j-over-slf4j +------------------+-->  slf4j         |  |
|                  |                                    |                   |  |
|                  +log4j2-to-slf4j                     +-------------------+  |
|                                                                              |
+------------------------------------------------------------------------------+

log-manager 中还包含了*-over-slf4j可以将使用commons-logging，log4j，log4j2日志桥接到slf4j-api，然后通过slf4j-jdk使用jul打印日志(和io.airlift.log.Logger一致)。

服务发现

airlift 的服务发现通过自身的 discovery 模块(服务发现客户端)和https://github.com/airlift/discovery项目(服务发现服务端)一起实现。

服务注册客户端

  • 服务注册地址通过discovery.uri属性或者ServiceInventory配置。
    • discovery.uri配置绑定到 DiscoveryClientConfig 的 discoveryServiceURI 属性上。io.airlift.discovery.client.DiscoveryModule 中的 getDiscoveryUriSupplier 会提供 DiscoveryUri 并标记上 @ForDiscoveryClient
  • HttpDiscoveryAnnouncementClient 构造时会注入 DiscoveryUri。然后会调用 announce 和 unannounce 注册服务。
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
public class HttpDiscoveryAnnouncementClient
        implements DiscoveryAnnouncementClient
{
    @Inject
    public HttpDiscoveryAnnouncementClient(
            @ForDiscoveryClient Supplier<URI> discoveryServiceURI,
            NodeInfo nodeInfo,
            JsonCodec<Announcement> announcementCodec,
            @ForDiscoveryClient HttpClient httpClient)
    {
        requireNonNull(discoveryServiceURI, "discoveryServiceURI is null");
        requireNonNull(nodeInfo, "nodeInfo is null");
        requireNonNull(announcementCodec, "announcementCodec is null");
        requireNonNull(httpClient, "httpClient is null");

        this.nodeInfo = nodeInfo;
        this.discoveryServiceURI = discoveryServiceURI;
        this.announcementCodec = announcementCodec;
        this.httpClient = httpClient;
    }
    // 注册节点
    @Override
    public ListenableFuture<Duration> announce(Set<ServiceAnnouncement> services)
    {
        requireNonNull(services, "services is null");

        URI uri = discoveryServiceURI.get();
        if (uri == null) {
            return immediateFailedFuture(new DiscoveryException("No discovery servers are available"));
        }
        // 注册信息
        Announcement announcement = new Announcement(nodeInfo.getEnvironment(), nodeInfo.getNodeId(), nodeInfo.getPool(), nodeInfo.getLocation(), services);
        Request request = preparePut()
                .setUri(createAnnouncementLocation(uri, nodeInfo.getNodeId()))
                .setHeader("User-Agent", nodeInfo.getNodeId())
                .setHeader("Content-Type", MEDIA_TYPE_JSON.toString())
                .setBodyGenerator(jsonBodyGenerator(announcementCodec, announcement))
                .build();
        return httpClient.executeAsync(request, new DiscoveryResponseHandler<Duration>("Announcement", uri)
        {
            @Override
            public Duration handle(Request request, Response response)
                    throws DiscoveryException
            {
                int statusCode = response.getStatusCode();
                if (!isSuccess(statusCode)) {
                    throw new DiscoveryException(String.format("Announcement failed with status code %s: %s", statusCode, getBodyForError(response)));
                }

                Duration maxAge = extractMaxAge(response); //抽取超时时间，默认是10s
                return maxAge;
            }
        });
    }
    // 注册服务URL
    @VisibleForTesting
    static URI createAnnouncementLocation(URI baseUri, String nodeId)
    {
        return uriBuilderFrom(baseUri)
                .appendPath("/v1/announcement")
                .appendPath(nodeId)
                .build();
    }
    // 取消注册
    @Override
    public ListenableFuture<Void> unannounce()
    {
        URI uri = discoveryServiceURI.get();
        if (uri == null) {
            return immediateFuture(null);
        }

        Request request = prepareDelete()
                .setUri(createAnnouncementLocation(uri, nodeInfo.getNodeId()))
                .setHeader("User-Agent", nodeInfo.getNodeId())
                .build();
        return httpClient.executeAsync(request, new DiscoveryResponseHandler<>("Unannouncement", uri));
    }
    ... ...
}

Client 通过向/v1/announcement/{node_id}发送 PUT/DELETE 请求来向服务器注册/取消注册自己。从上面的代码可以看到每次注册会返回超时时间，如果没有返回，按照约定 maxAge 是 10s。

客户端注册定时调度

客户端使用 Announcer 进行周期性的服务状态更新。

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

// io.airlift.discovery.client.Announcer
// announce 准备开始
public void start()
{
    checkState(!executor.isShutdown(), "Announcer has been destroyed");
    if (started.compareAndSet(false, true)) {
        // announce immediately, if discovery is running
        ListenableFuture<Duration> announce = announce(System.nanoTime(), new Duration(0, SECONDS));
        try {
            announce.get(30, SECONDS); // 30秒的最大等待时间
        }
        catch (Exception ignored) {
        }
    }
}
// announce 异步发送，在回调函数中进行延迟等待，然后准备下一次的announce
private ListenableFuture<Duration> announce(long delayStart, Duration expectedDelay)
{
    // log announcement did not happen within 5 seconds of expected delay
    if (System.nanoTime() - (delayStart + expectedDelay.roundTo(NANOSECONDS)) > SECONDS.toNanos(5)) {
        log.error("Expected service announcement after %s, but announcement was delayed %s", expectedDelay, Duration.nanosSince(delayStart));
    }

    long requestStart = System.nanoTime();
    // 在announcementClient中发起请求
    ListenableFuture<Duration> future = announcementClient.announce(getServiceAnnouncements());

    // 通过future的回调函数异步获得结果
    Futures.addCallback(future, new FutureCallback<Duration>()
    {
        @Override
        public void onSuccess(Duration expectedDelay)
        {
            errorBackOff.success();

            // wait 80% of the suggested delay，防止超出maxAge
            expectedDelay = new Duration(expectedDelay.toMillis() * 0.8, MILLISECONDS);
            scheduleNextAnnouncement(expectedDelay); // 下一次的announce
        }

        @Override
        public void onFailure(Throwable t)
        {
            Duration duration = errorBackOff.failed(t);
            scheduleNextAnnouncement(duration); // 下一次的announce
        }
    }, executor);
    return future;
}
// 下一次的announce
private void scheduleNextAnnouncement(Duration expectedDelay)
{
    // already stopped?  avoids rejection exception
    if (executor.isShutdown()) {
        return;
    }
    long delayStart = System.nanoTime();
    executor.schedule(() -> announce(delayStart, expectedDelay), expectedDelay.toMillis(), MILLISECONDS);
}

服务发现 Server

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// io.airlift.discovery.server.DynamicAnnouncementResource 服务端接收注册的代码
@PUT
@Consumes(MediaType.APPLICATION_JSON)
public Response put(@PathParam("node_id") Id<Node> nodeId, @Context UriInfo uriInfo, DynamicAnnouncement announcement)
{
  if (!nodeInfo.getEnvironment().equals(announcement.getEnvironment())) {
    return Response.status(BAD_REQUEST)
            .entity(format("Environment mismatch. Expected: %s, Provided: %s", nodeInfo.getEnvironment(), announcement.getEnvironment()))
            .build();
    }

  String location = firstNonNull(announcement.getLocation(), "/somewhere/" + nodeId.toString());

  DynamicAnnouncement announcementWithLocation = DynamicAnnouncement.copyOf(announcement)
    .setLocation(location)
    .build();

  dynamicStore.put(nodeId, announcementWithLocation);

  return Response.status(ACCEPTED).build();
}

服务端从注册接口收到注册信息后，会存入到 store 中。

1
2
3
4
5
6
7
8
9
10
11
12
13
@Override
public void put(Id<Node> nodeId, DynamicAnnouncement announcement)
{
  List<Service> services = FluentIterable.from(announcement.getServiceAnnouncements())
  .transform(toServiceWith(nodeId, announcement.getLocation(), announcement.getPool()))
                .toList();

  byte[] key = nodeId.getBytes();
  byte[] value = codec.toJsonBytes(services);

  store.put(key, value, maxAge); // 设置注册的超时时间
  // maxAge 可以通过 discovery.max-age 配置，默认是30S
}

服务发现Server模式

服务发现Server有两种启动模式:

  1. EmbeddedDiscoveryModule: discovery作为某个服务的内嵌模块启动。
  2. DiscoveryServer: discovery独立部署。

服务发现Server支持两种部署模式:

  1. 单点部署
  2. 集群部署

这两种方式均是通过ServiceInventory来实现的,service inventory可以是文件或者一个http链接，但是其内容结构都如下:

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
{
   "environment":"test",
   "services":[
      {
         "id":"C8A9EE64-0476-452C-8638-8E72F3EE3CA6",
         "nodeId":"597A741E-9968-40E2-BB4D-7AF26DE18689",
         "type":"discovery",
         "pool":"general",
         "location":"/172.17.31.245",
         "state":"RUNNING",
         "properties":{
            "http":"http://172.17.31.245:8411"
         }
      },
      {
         "id":"370AF416-5F44-47D3-BFB6-D93A92676D49",
         "nodeId":"0BA42FDB-5DBA-4A2C-BE26-9596B7B4368E",
         "type":"discovery",
         "pool":"general",
         "location":"/172.17.31.246",
         "state":"RUNNING",
         "properties":{
            "http":"http://172.17.31.246:8411"
         }
      }
   ]
}

服务发现Server会查找ServiceInventory中type为discovery的服务，如果是一个就是单点部署，如果是多个就是集群模式。也可以不配置ServiceInventory，也是单点部署。

集群模式支持HA:

  1. 前置LoadBalancer
  2. discoveryServer之间互相同步
flowchart LR
    ClientA -->|register|LB(LoadBalancer)
    ClientB -->|register|LB
    ClientC -->|register|LB
    LB -->|rr|server1[discoveryServer1]
    LB -->|rr|server2[discoveryServer2]
    server1-->|sync|server2
    server2-->|sync|server1

DistributedStore

看代码，dynamicStore 内部的 store 是一个分布式 kv 存储(DistributedStore)。信息在存储在本地的localStore同时，会存储到远程的remoteStore。不同启动模式下localStore实例不同

  1. EmbeddedDiscoveryModule: localStore是InMemoryStore，内部是ConcurrentMap
  2. DiscoveryServer: localStore是PersistentStore，内部是leveldb
1
2
3
4
5
6
7
8
9
10
11
12
//io.airlift.discovery.store.DistributedStore
public void put(byte[] key, byte[] value, Duration maxAge)
{
  ... ...

  long now = timeSupplier.get().getMillis();

  Entry entry = new Entry(key, value, new Version(now), now, maxAge.toMillis());

  localStore.put(entry);
  remoteStore.put(entry);
}

在启动这个分布式 kv 存储时，会启动一个过期检查，定时检查 PUT 过来的注册信息，根据里面的生命周期条件来判断是否进行过期释放。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
//io.airlift.discovery.store.DistributedStore
@PostConstruct
public void start(){
  garbageCollector.scheduleAtFixedRate(new Runnable(){
    @Override
    public void run()
    {
      removeExpiredEntries();
    }
  }, 0, garbageCollectionInterval.toMillis(), TimeUnit.MILLISECONDS);
}

@Managed
public void removeExpiredEntries(){
  for (Entry entry : localStore.getAll()) {
    if (isExpired(entry)) {
        localStore.delete(entry.getKey(), entry.getVersion());
    }
  }

  lastGcTimestamp.set(System.currentTimeMillis());
}

节点间同步

节点间同步分为两种方式:

  1. 定时同步(Replicator):定时同步(Replicator)比较简单就是本地启动一个线程，定时(store.remote.update-interval)从其他节点同步数据。
  2. 主备复制(remoteStore):
  3. HttpRemoteStore内部维护了远程的服务的BatchProcessor
  4. 写入HttpRemoteStore的数据会先写入每个BatchProcessor的queue中
  5. BatchProcessor的queue中的数据会不断写入远程服务

注册信息查询

DiscoveryModule 使用 CachingServiceSelector 获取有关在发现服务器上注册的服务的信息。加载 DiscoveryModule 模块时，会继续加载 CachingServiceSelectorFactory 与 MergingServiceSelectorFactory 来实现缓存刷新。

1
2
3
4
5
6
7
8
9
10
// io.airlift.discovery.client.DiscoveryModule
@Override
public void configure(Binder binder)
{
    ...
    // bind selector factory
    binder.bind(CachingServiceSelectorFactory.class).in(Scopes.SINGLETON);
    binder.bind(ServiceSelectorFactory.class).to(MergingServiceSelectorFactory.class).in(Scopes.SINGLETON);
    ...
}

CachingServiceSelector 定时向发现服务器发送一个请求以更新节点列表缓存。

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
// io.airlift.discovery.client.CachingServiceSelector
@Override
public ListenableFuture<List<ServiceDescriptor>> refresh()
{
    ServiceDescriptors oldDescriptors = this.serviceDescriptors.get();
    ListenableFuture<ServiceDescriptors> future;
    if (oldDescriptors == null) {
        future = lookupClient.getServices(type, pool);
    }
    else {
        future = lookupClient.refreshServices(oldDescriptors);
    }
    future = chainedCallback(future, new FutureCallback<ServiceDescriptors>()
    {
        @Override
        public void onSuccess(ServiceDescriptors newDescriptors)
        {
            serviceDescriptors.set(newDescriptors);
            errorBackOff.success();
            Duration delay = newDescriptors.getMaxAge();
            if (delay == null) {
                delay = DEFAULT_DELAY; // 这个变量的默认是10秒
            }
            scheduleRefresh(delay); // 刷新
        }
        @Override
        public void onFailure(Throwable t) { ... }
    }, executor);
    ...
}

同样我们也可以在日志里看到更新缓存的日志。

参考资料