14-系统设计

用户 → CDN(静态页) → 网关(Nginx/LB) → 秒杀服务集群
                                            ↓
                                    Redis 集群(库存)
                                    ↓ Lua 原子操作
                                    消息队列(RocketMQ)
                                            ↓
                                    订单服务 → MySQL

-- 秒杀活动表
CREATE TABLE seckill_activity (
    id BIGINT PRIMARY KEY,
    product_id BIGINT,
    total_stock INT,
    start_time DATETIME,
    end_time DATETIME
);

-- 秒杀订单表
CREATE TABLE seckill_order (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    user_id BIGINT,
    activity_id BIGINT,
    product_id BIGINT,
    create_time DATETIME,
    UNIQUE KEY uk_user_activity(user_id, activity_id)
);

-- Redis 数据结构
-- seckill:stock:{activityId} → 库存数
-- seckill:users:{activityId} → Set(已秒杀用户)

// 秒杀接口
public SeckillResult seckill(Long userId, Long activityId) {
    // 1. 检查活动状态
    if (!activity.isActive()) return fail("活动未开始");

    // 2. 检查用户是否已参与（一人一单）
    if (redis.sismember("seckill:users:" + activityId, userId))
        return fail("已参与");

    // 3. Redis Lua 原子扣减库存
    String luaScript = """
        local stock = redis.call('GET', KEYS[1])
        if stock <= 0 then return -1 end
        redis.call('DECR', KEYS[1])
        redis.call('SADD', KEYS[2], ARGV[1])
        return 1
    """;
    Long result = redis.eval(luaScript,
        "seckill:stock:" + activityId,
        "seckill:users:" + activityId,
        userId);

    if (result == -1) return fail("已售罄");

    // 4. 发送 MQ 异步下单
    mq.send("seckill_order", buildOrder(userId, activityId));

    return success("秒杀成功，请等待结果");
}

用户 → APP/Web → CDN → 网关 → 抢票服务
                                    ↓
                            Redis(座位状态+库存)
                            ↓ 分布式锁/Lua
                            订单服务 → MySQL
                                    ↓
                              支付服务 → 第三方支付

-- 演出/车次表
CREATE TABLE show_info (
    id BIGINT PRIMARY KEY,
    name VARCHAR(100),
    total_seats INT,
    start_time DATETIME
);

-- 票档表
CREATE TABLE ticket_level (
    id BIGINT PRIMARY KEY,
    show_id BIGINT,
    level_name VARCHAR(50),
    price DECIMAL(10,2),
    total INT,
    available INT
);

-- 订单表
CREATE TABLE ticket_order (
    id BIGINT PRIMARY KEY,
    user_id BIGINT,
    show_id BIGINT,
    seat_no VARCHAR(20),
    level_id BIGINT,
    status TINYINT, -- 0锁定/1已售/2已退
    lock_expire_time DATETIME,
    create_time DATETIME
);

-- Redis Key 设计
-- ticket:{showId}:{levelId} → 可用座位 Set
-- ticket:{showId}:locked → 锁定用户 Map

发红包：用户 → 红包服务 → 账户服务(扣款) → 消息服务(通知)
拆红包：用户 → 红包服务 → Redis(原子拆) → 账户服务(入账) → DB(记录)

-- 红包表
CREATE TABLE red_packet (
    id BIGINT PRIMARY KEY,
    send_id BIGINT,       -- 发送者
    receive_id BIGINT,    -- 接收者（0为群红包）
    total_amount DECIMAL(10,2),
    total_count INT,
    type TINYINT,         -- 1拼手气/2普通
    status TINYINT,       -- 0待领取/1已领完/2已过期
    create_time DATETIME,
    expire_time DATETIME
);

-- 红包领取记录表
CREATE TABLE red_packet_record (
    id BIGINT PRIMARY KEY,
    packet_id BIGINT,
    user_id BIGINT,
    amount DECIMAL(10,2),
    create_time DATETIME,
    UNIQUE KEY uk_packet_user(packet_id, user_id)
);

-- Redis Key 设计
-- redpacket:{id}:remain → 剩余金额和剩余人数(Hash)
-- redpacket:{id}:received → 已领取用户 Set

// Lua 脚本实现原子拆红包
String lua = """
    local remainAmount = tonumber(redis.call('HGET', KEYS[1], 'amount'))
    local remainCount = tonumber(redis.call('HGET', KEYS[1], 'count'))

    if remainCount <= 0 or remainAmount <= 0 then
        return -1  -- 已领完
    end

    -- 二倍均值法：保证每人至少 0.01 分
    local maxAmount = remainAmount / remainCount * 2
    local amount = math.random(1, math.floor(maxAmount * 100)) / 100
    if remainCount == 1 then
        amount = remainAmount
    end

    -- 更新剩余金额和人数
    redis.call('HSET', KEYS[1], 'amount', remainAmount - amount)
    redis.call('HSET', KEYS[1], 'count', remainCount - 1)
    redis.call('SADD', KEYS[2], ARGV[1])

    return amount
""";

// 幂等检查
if (redis.sismember("redpacket:" + packetId + ":received", userId))
    return fail("已领取");

Double amount = redis.eval(lua,
    "redpacket:" + packetId + ":remain",
    "redpacket:" + packetId + ":received",
    userId);

if (amount < 0) return fail("已领完");

// 异步：入账 + 记录领取
mq.send("redpacket_receive", packetId, userId, amount);

生产者 → 延迟队列(TTL) → 死信交换机 → 消费队列 → 消费者
                ↓
         消息过期(TTL)自动进入死信交换机

生产者 → ZADD(delay_queue, task, target_time)
              ↓
    定时任务(每秒扫描)：ZRANGEBYSCORE(delay_queue, 0, now)
              ↓
         取出到期任务 → 执行 → ZREM

环形数组 + 指针
每个槽表示一个时间单位(如1秒)
任务存入对应槽
指针每 tick 前进一格

// Redis ZSet 实现
// Key: delay:queue:1 (topic)
// Member: taskId
// Score: 目标执行时间戳

public class DelayQueue {
    private StringRedisTemplate redis;

    // 添加延迟任务
    public void addTask(String taskId, long delayMs) {
        long executeTime = System.currentTimeMillis() + delayMs;
        redis.opsForZSet().add("delay:queue", taskId, executeTime);
    }

    // 扫描到期任务（定时调用）
    public List<String> pollExpiredTasks() {
        long now = System.currentTimeMillis();
        Set<ZSetOperations.TypedTuple<String>> tasks =
            redis.opsForZSet().rangeByScore("delay:queue", 0, now);

        List<String> result = new ArrayList<>();
        for (ZSetOperations.TypedTuple<String> task : tasks) {
            redis.opsForZSet().remove("delay:queue", task.getValue());
            result.add(task.getValue());
        }
        return result;
    }
}

// 简化的时间轮实现
public class TimeWheel {
    private int tickDuration;  // 每个 tick 时长(ms)
    private int wheelSize;     // 槽数量
    private List<List<TimerTask>> buckets;
    private int currentIndex;

    public void addTask(TimerTask task, long delayMs) {
        int ticks = (int)(delayMs / tickDuration);
        int bucketIndex = (currentIndex + ticks) % wheelSize;
        buckets.get(bucketIndex).add(task);
    }

    // 每 tick 调用
    public void tick() {
        currentIndex = (currentIndex + 1) % wheelSize;
        List<TimerTask> tasks = buckets.get(currentIndex);
        // 执行所有到期任务
        for (TimerTask task : tasks) task.execute();
        tasks.clear();
    }
}

ID 服务集群
  ↓
雪花算法生成器(每实例独立)
  ↓
实例 ID 分配(Zookeeper/Redis)

| 符号位(1) | 时间戳(41) | 机器ID(10) | 序列号(12) |
|  0 固定   | 毫秒级时间  | 实例标识   | 同毫秒递增 |

public class SnowflakeIdGenerator {
    private long workerId;       // 机器ID (0-1023)
    private long sequence = 0L;  // 序列号
    private long lastTimestamp = -1L;

    private static final long WORKER_ID_BITS = 10L;
    private static final long MAX_WORKER_ID = ~(-1L << WORKER_ID_BITS); // 1023
    private static final long SEQUENCE_BITS = 12L;
    private static final long WORKER_ID_SHIFT = SEQUENCE_BITS;
    private static final long TIMESTAMP_LEFT_SHIFT =
        SEQUENCE_BITS + WORKER_ID_BITS;
    private static final long SEQUENCE_MASK = ~(-1L << SEQUENCE_BITS);
    private static final long EPOCH = 1489111610226L; // 起始时间戳

    public SnowflakeIdGenerator(long workerId) {
        if (workerId > MAX_WORKER_ID || workerId < 0)
            throw new IllegalArgumentException();
        this.workerId = workerId;
    }

    public synchronized long nextId() {
        long timestamp = System.currentTimeMillis();

        if (timestamp < lastTimestamp) {
            // 时钟回拨处理
            long offset = lastTimestamp - timestamp;
            if (offset <= 5) {
                try { Thread.sleep(offset); }
                catch (InterruptedException e) { e.printStackTrace(); }
                timestamp = System.currentTimeMillis();
            } else {
                throw new RuntimeException("时钟回拨过大");
            }
        }

        if (lastTimestamp == timestamp) {
            sequence = (sequence + 1) & SEQUENCE_MASK;
            if (sequence == 0) {
                timestamp = tilNextMillis(lastTimestamp);
            }
        } else {
            sequence = 0L;
        }

        lastTimestamp = timestamp;

        return ((timestamp - EPOCH) << TIMESTAMP_LEFT_SHIFT)
            | (workerId << WORKER_ID_SHIFT)
            | sequence;
    }

    private long tilNextMillis(long lastTimestamp) {
        long timestamp = System.currentTimeMillis();
        while (timestamp <= lastTimestamp) timestamp = System.currentTimeMillis();
        return timestamp;
    }
}

客户端 → SET NX PX 加锁 → Redis
客户端 → Lua 脚本解锁 → Redis

客户端 → 创建临时节点 → Zookeeper
客户端 → 监听前驱节点 → 等待

// Redis 分布式锁实现
public class RedisDistributedLock {
    private StringRedisTemplate redis;

    public boolean tryLock(String lockKey, String requestId, long expireMs) {
        Boolean result = redis.opsForValue()
            .setIfAbsent(lockKey, requestId, expireMs, TimeUnit.MILLISECONDS);
        return Boolean.TRUE.equals(result);
    }

    public boolean unlock(String lockKey, String requestId) {
        // Lua 脚本保证原子解锁
        String luaScript = """
            if redis.call('GET', KEYS[1]) == ARGV[1] then
                return redis.call('DEL', KEYS[1])
            end
            return 0
        """;
        Long result = redis.execute(
            new DefaultRedisScript<>(luaScript, Long.class),
            Collections.singletonList(lockKey), requestId);
        return result == 1L;
    }
}

RLock lock = redissonClient.getLock("myLock");
lock.lock(10, TimeUnit.SECONDS);
try {
    // 业务逻辑
} finally {
    lock.unlock();
}

请求 → 限流层(Sentinel/自研) → 业务服务
         ↓
   规则管理(控制台) → 推送到各节点

桶容量: burst
令牌生成速率: rate tokens/sec
每来一个请求消耗 1 个令牌
桶空时拒绝请求

// 令牌桶限流器
public class TokenBucketRateLimiter {
    private double tokens;       // 当前令牌数
    private double rate;        // 令牌生成速率
    private double burst;       // 桶容量
    private long lastTime;      // 上次时间

    public synchronized boolean tryAcquire() {
        long now = System.currentTimeMillis();
        double elapsed = (now - lastTime) / 1000.0;
        tokens = Math.min(burst, tokens + elapsed * rate);
        lastTime = now;

        if (tokens >= 1.0) {
            tokens -= 1.0;
            return true;
        }
        return false;
    }
}

// 基于 Redis 的分布式限流（Lua 滑动窗口）
public class DistributedRateLimiter {
    public boolean tryAcquire(String key, int limit, int windowSec) {
        String luaScript = """
            local key = KEYS[1]
            local limit = tonumber(ARGV[1])
            local window = tonumber(ARGV[2])
            local now = tonumber(ARGV[3])

            redis.call('ZREMRANGEBYSCORE', key, 0, now - window * 1000)
            local count = redis.call('ZCARD', key)

            if count < limit then
                redis.call('ZADD', key, now, now)
                redis.call('EXPIRE', key, window)
                return 1
            end
            return 0
        """;
        long now = System.currentTimeMillis();
        Long result = redis.execute(
            new DefaultRedisScript<>(luaScript, Long.class),
            Collections.singletonList(key),
            String.valueOf(limit),
            String.valueOf(windowSec),
            String.valueOf(now));
        return result == 1L;
    }
}

用户 → 网关 → 评论服务 → MySQL(存储)
                          ↕
                     Redis(缓存热点)
                          ↓
                     搜索引擎(ES)全文检索

-- 评论表（邻接表）
CREATE TABLE comment (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    article_id BIGINT NOT NULL,
    user_id BIGINT NOT NULL,
    content VARCHAR(2000) NOT NULL,
    parent_id BIGINT DEFAULT 0,  -- 0为顶层评论
    reply_to_user_id BIGINT DEFAULT 0,
    like_count INT DEFAULT 0,
    status TINYINT DEFAULT 1,
    create_time DATETIME,
    INDEX idx_article_parent(article_id, parent_id),
    INDEX idx_article_time(article_id, create_time)
);

-- 点赞表
CREATE TABLE comment_like (
    id BIGINT PRIMARY KEY,
    comment_id BIGINT,
    user_id BIGINT,
    create_time DATETIME,
    UNIQUE KEY uk_comment_user(comment_id, user_id)
);

-- Redis 缓存
-- comment:hot:{articleId} → ZSet(评论ID, 热度分)
-- comment:detail:{id} → Hash(评论详情)
-- comment:count:{articleId} → 评论总数

发帖 → 推送到粉丝收件箱(Redis ZSet) → 通知
                                          ↓
                                    粉丝拉取 Feed → 分页返回

-- 用户关系表
CREATE TABLE user_relation (
    user_id BIGINT,
    follow_id BIGINT,
    create_time DATETIME,
    PRIMARY KEY(user_id, follow_id)
);

-- 帖子表
CREATE TABLE post (
    id BIGINT PRIMARY KEY,
    user_id BIGINT,
    content TEXT,
    create_time DATETIME
);

-- Redis Key 设计
-- feed:{userId} → ZSet(postId, 时间戳)  收件箱
-- fans:{userId} → Set(粉丝ID)           粉丝列表
-- follow:{userId} → Set(关注ID)         关注列表
-- post:detail:{id} → Hash(帖子详情)      帖子缓存

public void publishPost(Long userId, Post post) {
    // 1. 保存帖子
    postService.save(post);

    // 2. 获取粉丝列表（Redis 缓存）
    Set<Long> fans = redis.smembers("fans:" + userId);

    // 3. 推送到粉丝收件箱
    for (Long fanId : fans) {
        redis.zadd("feed:" + fanId, post.getId(), post.getCreateTime());
        // 截断收件箱到最大 500 条
        redis.zremrangebyrank("feed:" + fanId, 0, -501);
    }

    // 4. 大 V 优化：异步推送 + 分批
    // ...
}

客户端 → 分片 → 并发上传 → CDN/OSS → 服务端合并
                                          ↓
                                    上传状态存储(Redis/DB)

-- 上传任务表
CREATE TABLE upload_task (
    id BIGINT PRIMARY KEY,
    user_id BIGINT,
    file_name VARCHAR(200),
    file_size BIGINT,
    file_md5 VARCHAR(64),
    total_chunks INT,
    uploaded_chunks JSON,  -- 已上传分片列表
    status TINYINT,        -- 0上传中/1已完成/2失败
    create_time DATETIME
);

-- Redis 缓存
-- upload:{taskId} → Hash(已上传分片)
-- file:md5:{md5} → fileId(秒传)

// 分片上传核心逻辑
public class ChunkUploader {
    private static final int CHUNK_SIZE = 5 * 1024 * 1024; // 5MB
    private static final int MAX_CONCURRENT = 5;

    public UploadResult upload(File file) {
        // 1. 计算 MD5
        String md5 = calculateMD5(file);

        // 2. 秒传判断
        if (isFileExists(md5)) {
            return UploadResult.quickSuccess(getFileId(md5));
        }

        // 3. 创建上传任务
        UploadTask task = createTask(file, md5);

        // 4. 分片上传（并发）
        List<Chunk> chunks = splitFile(file, CHUNK_SIZE);
        Set<Integer> uploadedChunks = getUploadedChunks(task.getId());

        // 过滤已上传分片
        List<Chunk> remaining = chunks.stream()
            .filter(c -> !uploadedChunks.contains(c.index))
            .collect(Collectors.toList());

        // 并发上传
        ExecutorService executor = Executors.newFixedThreadPool(MAX_CONCURRENT);
        List<Future<?>> futures = new ArrayList<>();

        for (Chunk chunk : remaining) {
            futures.add(executor.submit(() -> {
                uploadChunk(task.getId(), chunk);
                markChunkUploaded(task.getId(), chunk.index);
            }));
        }

        // 等待所有分片完成
        for (Future<?> f : futures) f.get();

        // 5. 合并文件
        return mergeChunks(task);
    }
}

// 固定大小线程池
ExecutorService executor = Executors.newFixedThreadPool(5);
List<Future<Result>> futures = new ArrayList<>();

for (int i = 0; i < 100; i++) {
    futures.add(executor.submit(() -> {
        return processTask(i);
    }));
}

// 收集结果
for (Future<Result> f : futures) {
    Result r = f.get(); // 阻塞等待
    // 处理结果
}

executor.shutdown();

Semaphore semaphore = new Semaphore(5);
CountDownLatch latch = new CountDownLatch(100);

for (int i = 0; i < 100; i++) {
    final int taskId = i;
    new Thread(() -> {
        try {
            semaphore.acquire();
            processTask(taskId);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            semaphore.release();
            latch.countDown();
        }
    }).start();
}

latch.await(); // 等待所有任务完成

BlockingQueue<Task> queue = new LinkedBlockingQueue<>(100);
for (int i = 0; i < 100; i++) queue.put(new Task(i));

// 5 个工作线程
for (int i = 0; i < 5; i++) {
    new Thread(() -> {
        while (true) {
            Task task = queue.take(); // 阻塞获取
            processTask(task);
        }
    }).start();
}

用户 → 应用层 → 缓存(Redis) → 数据库(读写分离+分库分表)
                  ↓                    ↓
              缓存命中             主库写/从库读

-- 垂直拆分示例
-- 用户库: user_{0-9}.user_info_{0-9}
-- 订单库: order_{0-9}.order_info_{0-9}

-- 分库分表路由
public class ShardingRouter {
    // 分片键：user_id
    public String getDatabase(Long userId) {
        return "user_" + (userId % 10);
    }
    public String getTable(Long userId) {
        return "user_info_" + (userId % 10);
    }
}

ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);

// 延迟执行
executor.schedule(() -> {
    // 业务逻辑
}, 5, TimeUnit.SECONDS);

// 周期执行
executor.scheduleAtFixedRate(() -> {
    // 周期任务
}, 0, 1, TimeUnit.SECONDS);

public class DelayTaskScheduler {
    private StringRedisTemplate redis;
    private static final String KEY = "delay:tasks";

    // 添加延迟任务
    public void addTask(String taskId, long delaySec) {
        long executeTime = System.currentTimeMillis() + delaySec * 1000;
        redis.opsForZSet().add(KEY, taskId, executeTime);
    }

    // 扫描到期任务（每秒调用）
    @Scheduled(fixedRate = 1000)
    public void scanAndExecute() {
        long now = System.currentTimeMillis();
        Set<ZSetOperations.TypedTuple<String>> tasks =
            redis.opsForZSet().rangeByScore(KEY, 0, now);

        for (ZSetOperations.TypedTuple<String> task : tasks) {
            redis.opsForZSet().remove(KEY, task.getValue());
            executeTask(task.getValue());
        }
    }
}

public class TimingWheel {
    private long tickMs;          // 每 tick 时长
    private int wheelSize;         // 槽数量
    private List<List<TimerTask>> buckets;
    private int currentIndex;
    private ScheduledExecutorService executor;

    public void start() {
        executor.scheduleAtFixedRate(() -> {
            currentIndex = (currentIndex + 1) % wheelSize;
            List<TimerTask> tasks = buckets.get(currentIndex);
            for (TimerTask task : tasks) task.execute();
            tasks.clear();
        }, tickMs, tickMs, TimeUnit.MILLISECONDS);
    }

    public void addTask(TimerTask task, long delayMs) {
        long ticks = delayMs / tickMs;
        int slot = (currentIndex + (int)ticks) % wheelSize;
        buckets.get(slot).add(task);
    }
}

// user.proto
syntax = "proto3";
package com.example;

message UserRequest {
    int64 user_id = 1;
    string name = 2;
    repeated string tags = 3;
}

message UserResponse {
    int32 code = 1;
    string message = 2;
}

service UserService {
    rpc GetUser(UserRequest) returns (UserResponse);
}

用户提交 → 判题服务 → 任务队列
                            ↓
                    沙箱调度器 → Docker 容器池
                                    ↓
                            编译 → 执行 → 采集结果
                                    ↓
                            结果返回 → 容器销毁

public class CodeSandbox {

    public JudgeResult judge(SubmitRequest request) {
        // 1. 选择或创建容器
        String containerId = dockerClient.createContainer(
            request.getLanguage(),
            "1g",    // 内存限制
            "1",     // CPU限制
            5        // 超时时间
        );

        try {
            // 2. 写入用户代码
            dockerClient.copyToContainer(containerId,
                request.getCodePath(), "/app/Main.java");

            // 3. 编译代码
            ExecResult compileResult = dockerClient.execInContainer(
                containerId, "javac /app/Main.java");
            if (compileResult.getExitCode() != 0) {
                return JudgeResult.compileError(compileResult.getStderr());
            }

            // 4. 执行代码（带超时）
            ExecResult runResult = dockerClient.execInContainer(
                containerId, "timeout 5 java -Xmx256m -cp /app Main");

            // 5. 采集结果
            return JudgeResult.builder()
                .output(runResult.getStdout())
                .error(runResult.getStderr())
                .time(runResult.getExecutionTime())
                .memory(runResult.getMemoryUsage())
                .build();

        } finally {
            // 6. 销毁容器
            dockerClient.removeContainer(containerId);
        }
    }
}

用户代码 → 静态检查 → 编译 → 容器执行
                              ↓
                    cgroup(资源限制) + seccomp(系统调用过滤)
                              ↓
                    运行时监控 → 超时/超限熔断

# Docker 容器安全配置
docker run \
    --security-opt no-new-privileges \    # 禁止提权
    --cap-drop ALL \                       # 丢弃所有 capabilities
    --cap-add NET_BIND_SERVICE \           # 仅保留必要能力
    --read-only \                          # 只读根文件系统
    --tmpfs /tmp:rw,noexec,nosuid,size=64m \ # 临时文件系统
    pwn-sandbox:latest

// Docker 资源限制配置
HostConfig hostConfig = new HostConfig()
    .withMemory(256 * 1024 * 1024L)         // 内存 256MB
    .withMemorySwap(256 * 1024 * 1024L)     // 禁用 swap
    .withCpuShares(512)                     // CPU 权重
    .withNanoCPUs(1000000000L)              // 1核 CPU
    .withBlkioWeight(100)                   // IO 权重
    .withNetworkMode("none")                // 禁用网络
    .withReadonlyRootfs(true);              // 只读根文件系统

{
    "defaultAction": "SCMP_ACT_ERRNO",
    "arch": ["SCMP_ARCH_X86_64"],
    "syscalls": [
        {"names": ["read", "write", "close", "fstat"], "action": "SCMP_ACT_ALLOW"},
        {"names": ["brk", "mmap", "mprotect"], "action": "SCMP_ACT_ALLOW"},
        {"names": ["exit", "exit_group"], "action": "SCMP_ACT_ALLOW"},
        {"names": ["open", "openat", "mkdir"], "action": "SCMP_ACT_ERRNO"},
        {"names": ["socket", "connect", "bind"], "action": "SCMP_ACT_ERRNO"},
        {"names": ["fork", "clone", "execve"], "action": "SCMP_ACT_ERRNO"}
    ]
}

// JVM SecurityManager 配置
Policy.setPolicy(new Policy() {
    @Override
    public boolean implies(ProtectionDomain d, Permission p) {
        // 禁止文件写入
        if (p instanceof FilePermission &&
            ((FilePermission) p).getActions().contains("write"))
            return false;
        // 禁止网络连接
        if (p instanceof NetPermission) return false;
        // 禁止进程创建
        if (p instanceof RuntimePermission &&
            ((RuntimePermission) p).getName().startsWith("createProcess"))
            return false;
        return super.implies(d, p);
    }
});
System.setSecurityManager(new SecurityManager());

用户提交 → 调度器 → 容器池管理器
                        ↓
                ┌── 就绪容器池 ──┐
                │ Java: 20 个    │
                │ Python: 15 个  │
                │ C++: 10 个     │
                └───────────────┘
                        ↓
                  分配容器 → 执行 → 回收

public class ContainerPool {
    private Map<String, BlockingQueue<String>> pools;
    private int poolSize;

    public ContainerPool(int poolSize) {
        this.poolSize = poolSize;
        this.pools = new ConcurrentHashMap<>();
        initPool("java", poolSize);
        initPool("python", poolSize);
        initPool("cpp", poolSize);
    }

    private void initPool(String language, int size) {
        BlockingQueue<String> queue = new LinkedBlockingQueue<>();
        for (int i = 0; i < size; i++) {
            String containerId = createWarmContainer(language);
            queue.offer(containerId);
        }
        pools.put(language, queue);
    }

    // 获取容器
    public String acquire(String language) throws InterruptedException {
        BlockingQueue<String> queue = pools.get(language);
        String containerId = queue.poll(10, TimeUnit.SECONDS);
        if (containerId == null) {
            // 池空，动态创建
            return createWarmContainer(language);
        }
        return containerId;
    }

    // 归还容器
    public void release(String language, String containerId) {
        // 重置容器状态（清空文件、重置环境）
        resetContainer(containerId);
        pools.get(language).offer(containerId);
    }

    // 创建预热容器
    private String createWarmContainer(String language) {
        // 基于语言镜像创建，预装编译环境
        return dockerClient.createContainer(
            getLanguageImage(language),
            getSecurityConfig()
        );
    }

    // 重置容器（快速恢复到干净状态）
    private void resetContainer(String containerId) {
        // 方案一：执行清理脚本
        dockerClient.execInContainer(containerId, "rm -rf /app/*");
        // 方案二：提交快照并恢复（更快）
        dockerClient.commitContainer(containerId, "clean-snapshot");
    }
}

启动时预创建 N 个容器 → 池化管理 → 按需分配 → 执行后回收

基础镜像（只读层） ← 所有容器共享
容器可写层 ← 每个容器独立
执行完毕丢弃可写层

按语言分级：
- Java 容器：256MB 内存，1 核 CPU
- Python 容器：128MB 内存，0.5 核 CPU
- C++ 容器：64MB 内存，0.5 核 CPU

调度策略：
- 优先分配轻量容器
- 按资源需求排队

容器执行完毕 → 提交快照 → 恢复快照 → 快速复用
比重新创建快 10-50 倍

应用服务器 → Filebeat(轻量采集) → Kafka(消息队列) → Logstash(处理) → Elasticsearch(存储) → Kibana(检索)
                                                                                      ↓
                                                                              冷热分离(ILM)
                                                                                      ↓
                                                                              对象存储(S3/OSS)

应用日志 → 异步写入本地文件 → Filebeat 采集 → Kafka → Logstash → ES

// Elasticsearch Index Lifecycle Management
PUT _ilm/policy/my-lifecycle
{
    "policy": {
        "phases": {
            "hot": {
                "min_age": "0ms",
                "actions": {
                    "rollover": {
                        "max_primary_shard_size": "50gb"
                    },
                    "set_priority": {
                        "priority": 100
                    }
                }
            },
            "warm": {
                "min_age": "7d",
                "actions": {
                    "forcemerge": {"max_num_segments": 1},
                    "shrink": {"number_of_shards": 1},
                    "allocate": {"require": {"data": "warm"}},
                    "set_priority": {"priority": 50}
                }
            },
            "cold": {
                "min_age": "30d",
                "actions": {
                    "freeze": {},
                    "allocate": {"require": {"data": "cold"}}
                }
            },
            "delete": {
                "min_age": "180d",
                "actions": {"delete": {}}
            }
        }
    }
}

- 按天/按周创建索引，便于过期删除。
- 合理设置 shard 数（每个 shard 10-50GB）。
- 使用 mapping 优化：keyword（精确匹配）vs text（全文检索）。
- 关闭不需要的字段索引（doc_values、norms）。
- 使用 _source filtering 减少存储。

- 日志压缩：gzip 压缩传输。
- 字段降采样：非关键字段降精度。
- 历史数据归档：冷数据压缩后存对象存储。

// Elasticsearch 索引模板
PUT _template/logs_template
{
    "index_patterns": "logs-*",
    "settings": {
        "number_of_shards": 3,
        "number_of_replicas": 1,
        "refresh_interval": "30s",
        "index.translog.durability": "async"
    },
    "mappings": {
        "properties": {
            "@timestamp": {"type": "date"},
            "level": {"type": "keyword"},
            "service": {"type": "keyword"},
            "trace_id": {"type": "keyword"},
            "user_id": {"type": "keyword"},
            "message": {"type": "text", "analyzer": "ik_max_word"},
            "stack_trace": {"type": "text", "index": false},
            "host": {"type": "keyword"},
            "ip": {"type": "ip"}
        }
    }
}