构建基于 Flink 与 Micronaut 的实时特征存储服务及其容器化与监控面板实现

数据工程

文章字数: 4.3k

阅读时长: 20 分

为机器学习模型提供实时特征是一项常见的工程挑战，其核心诉求是“极致的低延迟”。当业务要求在数百毫秒内结合用户当前行为与历史模式进行决策时，传统的批处理特征工程链路便显得力不从心。我们需要一个能够捕获流式数据、实时计算特征、并能通过高性能API对外提供服务的完整解决方案。

问题的复杂性并不仅仅在于数据处理本身，它是一个横跨数据管道、服务API、部署运维和可观测性的系统工程。一个常见的方案是采用 Kafka + Flink/Spark Streaming + Redis + Spring Boot 的技术栈。这套组合功能强大，生态成熟，但在追求极致性能和资源效率的场景下，会暴露出一些值得权衡的短板。例如，Spring Boot 虽然功能全面，但其基于反射和运行时代理的机制导致了较慢的启动速度和更高的内存占用，这在需要快速弹性伸缩的云原生环境中会增加成本和延迟。同时，标准的 Dockerfile 构建流程虽然普及，但在追求最小化镜像和更高构建安全性时，其分层机制和守护进程依赖也并非最优解。

因此，我们决定探索一套更为激进的技术栈，目标是构建一个从数据处理到服务交付全链路优化的实时特征平台。具体的技术选型决策如下：

数据处理层 (Data Processing Layer): 维持 Apache Flink 不变。其强大的状态管理、精确一次（Exactly-Once）语义以及高吞吐低延迟的特性，在实时计算领域依然是无可争议的王者。
特征服务层 (Feature Serving Layer): 采用 Micronaut 替代 Spring Boot。Micronaut 通过在编译期进行依赖注入和AOP代理，生成无反射的字节码，从而实现毫秒级启动和极低的内存占用。这对于要求快速扩缩容的特征API服务至关重要。
容器化方案 (Containerization): 使用 Buildah 替代 Dockerfile。Buildah 是一个无守护进程的容器镜像构建工具，它提供了更细粒度的控制，允许我们通过脚本构建出更小、更安全的“From Scratch”镜像，减少攻击面并提升部署效率。
监控与观测层 (Monitoring & Observability): 自研一个轻量级的React监控面板，UI部分采用 Styled-components。相比于配置复杂的通用监控系统（如Grafana），一个专用的前端面板能更直观地展示我们关心的核心指标，例如特定特征的最新值、更新延迟、数据流QPS等，并能与后端API进行更深度的交互。

这个决策的核心是基于对生产环境稳定性和资源效率的务实考量。我们牺牲了一部分生态的广度，换取了在关键性能指标上的显著提升。

flowchart TD
    subgraph "数据源"
        A[Kafka Topic: user_interactions]
    end

    subgraph "实时计算 (Apache Flink)"
        B[Flink Job: RealtimeFeatureAggregator]
        B -- stateful aggregation --> C{Flink Managed State}
        A --> B
    end

    subgraph "特征存储 (Low-Latency Sink)"
        D[Redis Hash]
        B -- sink --> D
    end

    subgraph "特征服务 (Micronaut)"
        E[Micronaut REST API]
        E -- GET /features/{userId} --> D
    end

    subgraph "容器化 (Buildah)"
        F[Optimized Container Image]
        E -- packaged into --> F
    end

    subgraph "监控面板 (React)"
        G[React UI with Styled-components]
        G -- polls --> E
    end

    subgraph "最终用户/系统"
        H[ML Model Inference Service]
        H -- HTTP Request --> E
    end

    style F fill:#d4f0d1,stroke:#333,stroke-width:2px
    style E fill:#cde4f9,stroke:#333,stroke-width:2px
    style B fill:#f9e7cd,stroke:#333,stroke-width:2px
    style G fill:#f9d5e5,stroke:#333,stroke-width:2px

接下来的内容将详细阐述这个架构中各个核心组件的具体实现。

Apache Flink: 实时特征聚合

我们的场景是计算用户在最近1分钟、5分钟和15分钟内的点击次数。这是典型的窗口聚合计算。

首先，定义输入事件和输出的特征数据结构。

// src/main/java/com/example/flink/events/InteractionEvent.java
package com.example.flink.events;

import java.io.Serializable;

// 用户交互事件，假设从 Kafka 传入
public class InteractionEvent implements Serializable {
    public long userId;
    public String eventType; // e.g., "CLICK", "VIEW"
    public long timestamp;

    public InteractionEvent() {}

    public InteractionEvent(long userId, String eventType, long timestamp) {
        this.userId = userId;
        this.eventType = eventType;
        this.timestamp = timestamp;
    }
}

// src/main/java/com/example/flink/features/UserClickFeatures.java
package com.example.flink.features;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;

// 输出到 Redis 的特征结构
public class UserClickFeatures implements Serializable {
    public long userId;
    public Map<String, Long> clickCounts;
    public long lastUpdated;

    public UserClickFeatures(long userId) {
        this.userId = userId;
        this.clickCounts = new HashMap<>();
        this.lastUpdated = System.currentTimeMillis();
    }

    public void setFeature(String window, long count) {
        this.clickCounts.put(window, count);
    }
}

接下来是 Flink 作业的核心逻辑。这里我们使用 TumblingEventTimeWindows 来实现固定时间窗口的聚合。为了将结果写入 Redis，我们自定义一个 RedisSink。在真实项目中，使用 Flink 官方或社区成熟的 Redis Connector 会是更稳妥的选择，但这里为了演示，我们自己实现一个简单的版本。

// src/main/java/com/example/flink/job/RealtimeFeatureAggregator.java
package com.example.flink.job;

import com.example.flink.events.InteractionEvent;
import com.example.flink.features.UserClickFeatures;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import java.time.Duration;
import java.util.Properties;

public class RealtimeFeatureAggregator {

    private static final String REDIS_HOST = "localhost";
    private static final int REDIS_PORT = 6379;
    private static final String REDIS_KEY_PREFIX = "user_features:";

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "flink-feature-aggregator");

        FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<>(
            "user_interactions",
            new SimpleStringSchema(),
            props
        );

        ObjectMapper objectMapper = new ObjectMapper();

        DataStream<InteractionEvent> events = env.addSource(kafkaSource)
            .map(value -> objectMapper.readValue(value, InteractionEvent.class))
            .filter(event -> "CLICK".equals(event.getEventType()));

        // 分配时间戳和水印
        DataStream<InteractionEvent> withTimestampsAndWatermarks = events.assignTimestampsAndWatermarks(
            WatermarkStrategy.<InteractionEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner((event, timestamp) -> event.timestamp)
        );

        // 计算1分钟窗口的点击数
        DataStream<UserClickFeatures> oneMinFeatures = withTimestampsAndWatermarks
            .keyBy(event -> event.userId)
            .window(TumblingEventTimeWindows.of(Time.minutes(1)))
            .aggregate(new ClickCounter(), (key, window, input, out) -> {
                UserClickFeatures features = new UserClickFeatures(key);
                features.setFeature("1m", input.iterator().next());
                out.collect(features);
            });

        // 写入 Redis
        oneMinFeatures.addSink(new RedisFeatureSink());

        env.execute("Realtime User Click Feature Aggregation");
    }

    // 简单的聚合函数，只计数
    public static class ClickCounter implements AggregateFunction<InteractionEvent, Long, Long> {
        @Override
        public Long createAccumulator() {
            return 0L;
        }
        @Override
        public Long add(InteractionEvent value, Long accumulator) {
            return accumulator + 1;
        }
        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }
        @Override
        public Long merge(Long a, Long b) {
            return a + b;
        }
    }

    // 自定义 Sink 写入 Redis
    public static class RedisFeatureSink extends RichSinkFunction<UserClickFeatures> {
        private transient JedisPool jedisPool;
        private transient ObjectMapper mapper;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            JedisPoolConfig poolConfig = new JedisPoolConfig();
            poolConfig.setMaxTotal(10); // 根据并行度调整
            jedisPool = new JedisPool(poolConfig, REDIS_HOST, REDIS_PORT);
            mapper = new ObjectMapper();
        }

        @Override
        public void invoke(UserClickFeatures value, Context context) throws Exception {
            try (Jedis jedis = jedisPool.getResource()) {
                String key = REDIS_KEY_PREFIX + value.userId;
                // 注意：这里是覆盖写，真实场景可能需要先读取再合并 (HGET, HSET)
                // 这样做是为了简化示例，但可能丢失其他窗口的数据
                String field = "clicks_1m";
                String fieldValue = value.clickCounts.get("1m").toString();
                jedis.hset(key, field, fieldValue);
                jedis.hset(key, "lastUpdated", String.valueOf(System.currentTimeMillis()));
                
                // 设置一个过期时间，防止冷数据无限堆积
                jedis.expire(key, 3600 * 24); 
            } catch (Exception e) {
                // 必须处理异常，否则可能导致 Flink 作业失败
                // log.error("Error writing to Redis", e);
            }
        }

        @Override
        public void close() throws Exception {
            if (jedisPool != null) {
                jedisPool.close();
            }
            super.close();
        }
    }
}

一个常见的错误是在 invoke 方法中每次都创建新的 Redis 连接，这会严重影响性能。正确的做法是在 open 方法中初始化连接池，在 invoke 中获取和归还连接。此外，生产级的 Sink 还需要考虑容错和幂等性写入。

Micronaut: 高性能特征服务API

Micronaut 服务的目标是提供一个端点 /features/{userId}，它能从 Redis 中快速读取特征并返回。

项目结构依赖 (build.gradle.kts):

plugins {
    id("com.github.johnrengelman.shadow") version "7.1.2"
    id("io.micronaut.application") version "3.7.9"
}

// ... micronaut configuration ...

dependencies {
    implementation("io.micronaut:micronaut-http-client")
    implementation("io.micronaut:micronaut-jackson-databind")
    implementation("io.micronaut.redis:micronaut-redis-lettuce") // Redis 客户端
    implementation("io.micronaut.serde:micronaut-serde-jackson")
    // ...
    runtimeOnly("ch.qos.logback:logback-classic")
}

配置 application.yml:

micronaut:
  application:
    name: feature-server
redis:
  uri: redis://localhost:6379

服务实现:

// src/main/java/com/example/featureserver/dto/FeatureResponse.java
package com.example.featureserver.dto;

import io.micronaut.serde.annotation.Serdeable;
import java.util.Map;

@Serdeable // Micronaut 的序列化注解
public record FeatureResponse(long userId, Map<String, String> features, long lastUpdated) {}


// src/main/java/com/example/featureserver/service/FeatureService.java
package com.example.featureserver.service;

import com.example.featureserver.dto.FeatureResponse;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import jakarta.inject.Singleton;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.Map;
import java.util.Optional;

@Singleton
public class FeatureService {

    private static final Logger LOG = LoggerFactory.getLogger(FeatureService.class);
    private static final String REDIS_KEY_PREFIX = "user_features:";

    private final StatefulRedisConnection<String, String> connection;

    // Micronaut 自动注入 Redis 连接
    public FeatureService(StatefulRedisConnection<String, String> connection) {
        this.connection = connection;
    }

    public Optional<FeatureResponse> getFeaturesByUserId(long userId) {
        String key = REDIS_KEY_PREFIX + userId;
        try {
            RedisCommands<String, String> syncCommands = connection.sync();
            Map<String, String> features = syncCommands.hgetall(key);

            if (features == null || features.isEmpty()) {
                return Optional.empty();
            }

            // 从 Map 中安全地提取 lastUpdated 字段
            long lastUpdated = Long.parseLong(features.getOrDefault("lastUpdated", "0"));
            features.remove("lastUpdated"); // 从特征列表中移除元数据

            return Optional.of(new FeatureResponse(userId, features, lastUpdated));
        } catch (Exception e) {
            // 在真实项目中，错误处理至关重要
            // 这里可能需要区分 Redis 连接问题和数据格式问题
            LOG.error("Failed to fetch features for user {} from Redis. Key: {}", userId, key, e);
            // 返回空，让上游处理，避免将错误暴露给客户端
            return Optional.empty();
        }
    }
}


// src/main/java/com/example/featureserver/controller/FeatureController.java
package com.example.featureserver.controller;

import com.example.featureserver.dto.FeatureResponse;
import com.example.featureserver.service.FeatureService;
import io.micronaut.http.HttpResponse;
import io.micronaut.http.annotation.Controller;
import io.micronaut.http.annotation.Get;
import io.micronaut.http.annotation.PathVariable;

@Controller("/features")
public class FeatureController {

    private final FeatureService featureService;

    public FeatureController(FeatureService featureService) {
        this.featureService = featureService;
    }

    @Get("/{userId}")
    public HttpResponse<FeatureResponse> getUserFeatures(@PathVariable long userId) {
        return featureService.getFeaturesByUserId(userId)
            .map(HttpResponse::ok) // 如果找到，返回 200 OK
            .orElse(HttpResponse.notFound()); // 如果未找到，返回 404 Not Found
    }
}

这个实现非常直接。Micronaut 的 @Singleton 和依赖注入机制使其看起来与 Spring 类似，但底层实现完全不同。没有运行时反射，所有代理和注入都在编译时完成，这就是其性能优势的来源。

Buildah: 精简与安全的容器化

现在，我们将上面构建的 Micronaut 应用打包成一个优化的容器镜像。我们将使用一个 shell 脚本来执行 Buildah 命令，这比维护一个 Dockerfile 更具灵活性和可编程性。

#!/bin/bash
set -eo pipefail

# -- 变量定义 --
# 使用 Red Hat 的通用基础镜像，非常小
BASE_IMAGE="registry.access.redhat.com/ubi8/ubi-minimal:latest"
# 我们构建的 Micronaut 应用的 fat jar
APP_JAR="./build/libs/feature-server-0.1-all.jar"
# 最终的镜像名称
IMAGE_NAME="localhost/feature-server:latest"
# 在容器内运行应用的非 root 用户
APP_USER="appuser"
USER_ID=1001

# -- 检查环境 --
if [ ! -f "$APP_JAR" ]; then
    echo "Error: Application JAR not found at $APP_JAR. Run ./gradlew shadowJar first."
    exit 1
fi

echo "--- Starting container image build with Buildah ---"

# -- 构建流程 --
# 1. 从基础镜像创建一个工作容器
# --cidfile: 将容器ID写入文件，方便后续引用
container=$(buildah from --cidfile "buildah-cid" "$BASE_IMAGE")
echo "Created container: $container"
trap 'buildah rm "$container"' EXIT # 确保在脚本退出时清理容器

# 2. 配置容器元数据
echo "Configuring container metadata..."
buildah config --author "TechCrafter" --label name="feature-server" "$container"

# 3. 创建一个非 root 用户来运行应用
# 这是安全最佳实践，避免在容器内使用 root
echo "Creating non-root user '$APP_USER'..."
buildah run "$container" -- useradd -u "$USER_ID" -ms /bin/bash "$APP_USER"

# 4. 在容器内创建应用目录
echo "Creating application directory..."
buildah run "$container" -- mkdir /app
buildah run "$container" -- chown "$APP_USER":"$APP_USER" /app

# 5. 复制应用 JAR 文件到容器内
# 这里的坑在于，直接复制后文件的所有者是 root。
# 我们需要使用 --chown 标志来确保文件属于我们创建的非 root 用户。
echo "Copying application JAR to /app/application.jar..."
buildah copy --chown "$APP_USER":"$APP_USER" "$container" "$APP_JAR" "/app/application.jar"

# 6. 设置容器工作目录和用户
echo "Setting working directory and user..."
buildah config --workingdir /app "$container"
buildah config --user "$APP_USER" "$container"

# 7. 暴露端口
echo "Exposing port 8080..."
buildah config --port 8080 "$container"

# 8. 设置启动命令 (Entrypoint)
# Micronaut 应用已经包含了 Netty，所以直接用 java -jar 启动
echo "Setting entrypoint..."
buildah config --cmd '["java", "-jar", "application.jar"]' "$container"

# 9. 提交工作容器为最终镜像
# --squash: 将所有层压缩为一层，生成更小的镜像
# --rm: 提交后删除工作容器
echo "Committing the container to an image: $IMAGE_NAME"
buildah commit --squash --rm "$container" "$IMAGE_NAME"

# 清理 cid 文件
rm -f "buildah-cid"

echo "--- Build successful! Image '$IMAGE_NAME' created. ---"
echo "You can now run it with: podman run -p 8080:8080 --rm $IMAGE_NAME"

这个脚本的核心优势在于：

无守护进程: 不需要像 Docker 那样有一个后台运行的 daemon，可以直接在 CI/CD pipeline 中作为普通命令执行。
细粒度控制: 每个步骤都是一个独立的命令，易于调试和扩展。
安全性: 轻松实现非 root 用户运行应用，并精确控制文件权限，最小化攻击面。
优化: --squash 选项可以轻松创建单层镜像，减小镜像体积。

React & Styled-components: 定制化监控面板

最后，我们构建一个简单的前端面板来实时查看某个用户的特征。这里使用 styled-components 来实现组件化的 CSS，使得样式与组件逻辑紧密耦合，非常适合构建这种目的单一的设计系统。

// src/components/FeatureDisplay.js
import React, { useState, useEffect } from 'react';
import styled, { keyframes, css } from 'styled-components';

// --- Styled Components Definition ---

const fadeIn = keyframes`
  from { opacity: 0; transform: translateY(10px); }
  to { opacity: 1; transform: translateY(0); }
`;

const Card = styled.div`
  background: #2a2a2a;
  border-radius: 8px;
  padding: 24px;
  color: #e0e0e0;
  font-family: 'Inter', sans-serif;
  width: 400px;
  box-shadow: 0 4px 15px rgba(0, 0, 0, 0.4);
  animation: ${fadeIn} 0.5s ease-out;
`;

const Header = styled.h2`
  margin: 0 0 20px 0;
  font-size: 1.5em;
  color: #ffffff;
  border-bottom: 1px solid #444;
  padding-bottom: 10px;
`;

const FeatureRow = styled.div`
  display: flex;
  justify-content: space-between;
  align-items: center;
  padding: 12px 0;
  border-bottom: 1px solid #3a3a3a;
  &:last-child {
    border-bottom: none;
  }
`;

const FeatureName = styled.span`
  font-weight: 500;
  color: #a0a0a0;
`;

const FeatureValue = styled.span`
  font-family: 'Roboto Mono', monospace;
  font-size: 1.2em;
  font-weight: bold;
  color: #4caf50; /* A vibrant color for the value */
`;

const StatusIndicator = styled.div`
  display: flex;
  align-items: center;
  margin-top: 20px;
  font-size: 0.9em;
  color: #888;
`;

const StatusDot = styled.div`
  width: 10px;
  height: 10px;
  border-radius: 50%;
  margin-right: 8px;
  background-color: ${props => (props.stale ? '#f44336' : '#4caf50')};
  transition: background-color 0.3s ease;
`;

// --- React Component ---

const API_ENDPOINT = 'http://localhost:8080/features/';
const USER_ID = 123; // Hardcoded for this example
const REFRESH_INTERVAL_MS = 2000;
const STALE_THRESHOLD_MS = 10000; // 10 seconds

export const FeatureDisplay = () => {
  const [features, setFeatures] = useState(null);
  const [error, setError] = useState(null);
  const [lastFetchTime, setLastFetchTime] = useState(null);

  useEffect(() => {
    const fetchData = async () => {
      try {
        const response = await fetch(`${API_ENDPOINT}${USER_ID}`);
        if (!response.ok) {
          throw new Error(`HTTP error! status: ${response.status}`);
        }
        const data = await response.json();
        setFeatures(data);
        setError(null);
      } catch (e) {
        setError('Failed to fetch features. Service might be down.');
        setFeatures(null); // Clear old data on error
      } finally {
        setLastFetchTime(Date.now());
      }
    };

    fetchData(); // Initial fetch
    const intervalId = setInterval(fetchData, REFRESH_INTERVAL_MS);

    return () => clearInterval(intervalId); // Cleanup on unmount
  }, []);

  const isStale = features && (Date.now() - features.lastUpdated > STALE_THRESHOLD_MS);

  return (
    <Card>
      <Header>User #{USER_ID} Features</Header>
      {error && <p style={{ color: '#f44336' }}>{error}</p>}
      {features ? (
        <>
          {Object.entries(features.features).map(([key, value]) => (
            <FeatureRow key={key}>
              <FeatureName>{key}</FeatureName>
              <FeatureValue>{value}</FeatureValue>
            </FeatureRow>
          ))}
          <StatusIndicator>
            <StatusDot stale={isStale} />
            <span>
              {isStale ? 'Data is stale' : 'Live'} - Last Updated: {new Date(features.lastUpdated).toLocaleTimeString()}
            </span>
          </StatusIndicator>
        </>
      ) : (
        !error && <p>Loading features...</p>
      )}
    </Card>
  );
};

最酷的是，styled-components 允许我们将动态逻辑（如数据是否“陈旧”）直接绑定到CSS属性上，如 StatusDot 的背景颜色。这种方式比传统的CSS类切换更为声明式和直观。

架构的扩展性与局限性

这套架构并非银弹，它在获得性能和效率的同时，也带来了自身的挑战。

扩展性:

添加新特征: 只需在 Flink 作业中增加新的聚合逻辑，并将结果写入 Redis 的新字段即可。API 和前端无需修改就能动态展示新特征。
服务扩容: Micronaut 服务的无状态设计和快速启动特性使其极易进行水平扩展。在 Kubernetes 中，可以为其配置 HPA（Horizontal Pod Autoscaler），根据CPU或QPS指标自动增减 Pod 数量。
数据流扩容: Flink 和 Kafka 都是为水平扩展而设计的。可以通过增加 Flink 的并行度和 Kafka 的分区来提升整个数据管道的吞吐量。

局限性:

存储瓶颈: Redis 是一个基于内存的存储，当用户量（即 key 的数量）或特征维度（即 hash 中的 field 数量）巨大时，成本会急剧上升。对于海量用户或高基数特征，可能需要考虑使用 RocksDB on Flink 或者其他更适合大规模存储的 KV 数据库（如 TiKV）。
一致性模型: 当前方案是最终一致性。从事件发生到特征可查询之间存在一个由 Flink 窗口、处理和网络共同决定的延迟。对于要求强一致性的场景，此架构不适用。
技术栈复杂度: 这套组合（Java/Flink + Java/Micronaut + Shell/Buildah + JS/React）要求团队具备全栈能力。对于分工明确的大型团队，可能会增加沟通和维护成本。
监控的深度: 自研的监控面板虽然直观，但缺乏系统性。它无法替代 Prometheus + Grafana 提供的对 JVM、系统资源、网络IO等底层指标的全面监控。在生产环境中，两者应当是互补关系。

Micronaut Buildah Apache Flink Styled-components 实时特征平台

构建基于 Qwik 与 tRPC 的跨 Azure AKS 和阿里云 ACK 的地理分布式类型安全 API 架构

2023-10-27 分布式架构

Qwik Azure AKS Alibaba Cloud tRPC Kubernetes

基于 Raft 协议从零构建一个支持日志加密的 Docker 化分布式密钥存储节点

2023-10-27 分布式系统

Go 密钥管理分布式一致性 Docker Raft