Apache Flink 源码剖析 - 实战经验和最佳实践

/**
 * 并行度设置的最佳实践
 */
public class ParallelismBestPractices {
    
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 1. 全局并行度设置
        // 建议：CPU 核数的 1-2 倍，考虑 I/O 密集型任务可以设置更高
        env.setParallelism(Runtime.getRuntime().availableProcessors() * 2);
        
        // 2. 算子级别并行度设置
        DataStream<String> source = env.addSource(new MySourceFunction())
            .setParallelism(4); // 数据源通常设置较低的并行度
        
        DataStream<ProcessedData> processed = source
            .map(new MyMapFunction())
            .setParallelism(8) // CPU 密集型操作可以设置较高并行度
            .keyBy(ProcessedData::getKey)
            .window(TumblingEventTimeWindows.of(Time.minutes(1)))
            .aggregate(new MyAggregateFunction())
            .setParallelism(4); // 聚合操作通常需要较少的并行度
        
        processed.addSink(new MySinkFunction())
            .setParallelism(2); // Sink 操作根据下游系统能力设置
        
        env.execute("Parallelism Optimization Example");
    }
}

/**
 * 动态并行度调整策略
 */
public class DynamicParallelismStrategy {
    
    /**
     * 根据数据量动态调整并行度
     */
    public static int calculateOptimalParallelism(long dataRate, int cpuCores) {
        // 基于数据处理速率计算最优并行度
        if (dataRate < 1000) {
            return Math.max(1, cpuCores / 2);
        } else if (dataRate < 10000) {
            return cpuCores;
        } else {
            return cpuCores * 2;
        }
    }
    
    /**
     * 基于背压情况调整并行度
     */
    public static void adjustParallelismBasedOnBackpressure(
            double backpressureRatio, 
            JobManagerGateway jobManager, 
            JobID jobId) {
        
        if (backpressureRatio > 0.8) {
            // 高背压，建议增加并行度
            LOG.warn("High backpressure detected: {}. Consider increasing parallelism.", 
                     backpressureRatio);
        } else if (backpressureRatio < 0.2) {
            // 低背压，可能资源浪费
            LOG.info("Low backpressure: {}. Consider decreasing parallelism.", 
                     backpressureRatio);
        }
    }
}

/**
 * Slot 共享组优化
 */
public class SlotSharingOptimization {
    
    public static void optimizeSlotSharing(StreamExecutionEnvironment env) {
        
        // 1. CPU 密集型操作使用独立的 slot 共享组
        DataStream<String> cpuIntensiveStream = env.addSource(new MySource())
            .map(new CpuIntensiveMapFunction())
            .slotSharingGroup("cpu-intensive"); // 独立的 slot 共享组
        
        // 2. I/O 密集型操作使用另一个 slot 共享组
        DataStream<String> ioIntensiveStream = env.addSource(new MySource())
            .flatMap(new IoIntensiveFlatMapFunction())
            .slotSharingGroup("io-intensive");
        
        // 3. 轻量级操作可以共享默认 slot 组
        DataStream<String> lightweightStream = env.addSource(new MySource())
            .filter(new LightweightFilterFunction());
        // 使用默认 slot 共享组
        
        // 4. 关键路径操作使用专用 slot 共享组
        cpuIntensiveStream
            .keyBy(x -> x)
            .window(TumblingEventTimeWindows.of(Time.minutes(1)))
            .aggregate(new CriticalAggregateFunction())
            .slotSharingGroup("critical-path"); // 关键路径专用组
    }
}

# flink-conf.yaml 内存配置最佳实践
taskmanager.memory.process.size: 4gb

# JVM 堆内存配置
taskmanager.memory.task.heap.size: 1gb
taskmanager.memory.framework.heap.size: 128mb

# 托管内存配置（用于状态后端和批处理）
taskmanager.memory.managed.size: 1gb
taskmanager.memory.managed.fraction: 0.4

# 网络内存配置
taskmanager.memory.network.fraction: 0.1
taskmanager.memory.network.min: 64mb
taskmanager.memory.network.max: 1gb

# JVM 元空间配置
taskmanager.memory.jvm-metaspace.size: 256mb

# JVM 开销配置
taskmanager.memory.jvm-overhead.fraction: 0.1
taskmanager.memory.jvm-overhead.min: 192mb
taskmanager.memory.jvm-overhead.max: 1gb

/**
 * 对象重用优化
 */
public class ObjectReuseOptimization {
    
    /**
     * 启用对象重用
     */
    public static void enableObjectReuse(StreamExecutionEnvironment env) {
        // 启用对象重用以减少 GC 压力
        env.getConfig().enableObjectReuse();
    }
    
    /**
     * 自定义可重用对象的 MapFunction
     */
    public static class ReuseableMapFunction extends RichMapFunction<InputType, OutputType> {
        
        // 重用的输出对象
        private transient OutputType reuse;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            this.reuse = new OutputType();
        }
        
        @Override
        public OutputType map(InputType input) throws Exception {
            // 重用对象而不是创建新对象
            reuse.setField1(input.getField1());
            reuse.setField2(input.getField2());
            reuse.setTimestamp(System.currentTimeMillis());
            return reuse;
        }
    }
    
    /**
     * 使用对象池的优化策略
     */
    public static class ObjectPoolMapFunction extends RichMapFunction<InputType, OutputType> {
        
        private transient ObjectPool<OutputType> objectPool;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            // 初始化对象池
            this.objectPool = new GenericObjectPool<>(new OutputTypeFactory());
        }
        
        @Override
        public OutputType map(InputType input) throws Exception {
            OutputType output = null;
            try {
                // 从对象池获取对象
                output = objectPool.borrowObject();
                output.setField1(input.getField1());
                output.setField2(input.getField2());
                return output;
            } finally {
                if (output != null) {
                    // 归还对象到池中
                    objectPool.returnObject(output);
                }
            }
        }
        
        @Override
        public void close() throws Exception {
            super.close();
            if (objectPool != null) {
                objectPool.close();
            }
        }
    }
}

/**
 * 网络缓冲区优化配置
 */
public class NetworkOptimization {
    
    /**
     * 配置网络缓冲区
     */
    public static void configureNetworkBuffers() {
        Configuration config = new Configuration();
        
        // 网络缓冲区大小（默认 32KB）
        config.setString("taskmanager.memory.segment-size", "64kb");
        
        // 每个网络连接的缓冲区数量
        config.setInteger("taskmanager.network.numberOfBuffers", 8192);
        
        // 网络缓冲区超时时间
        config.setLong("taskmanager.network.buffer-timeout", 10);
        
        // 启用网络压缩
        config.setBoolean("taskmanager.network.compression.enable", true);
        
        // 网络压缩算法
        config.setString("taskmanager.network.compression.codec", "LZ4");
    }
    
    /**
     * 批量发送优化
     */
    public static class BatchingSinkFunction extends RichSinkFunction<MyData> {
        
        private static final int BATCH_SIZE = 1000;
        private static final long BATCH_TIMEOUT = 5000; // 5 seconds
        
        private transient List<MyData> batch;
        private transient long lastBatchTime;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            this.batch = new ArrayList<>(BATCH_SIZE);
            this.lastBatchTime = System.currentTimeMillis();
        }
        
        @Override
        public void invoke(MyData value, Context context) throws Exception {
            batch.add(value);
            
            // 检查是否需要发送批次
            if (batch.size() >= BATCH_SIZE || 
                System.currentTimeMillis() - lastBatchTime > BATCH_TIMEOUT) {
                sendBatch();
            }
        }
        
        private void sendBatch() throws Exception {
            if (!batch.isEmpty()) {
                // 批量发送数据
                externalSystem.sendBatch(batch);
                batch.clear();
                lastBatchTime = System.currentTimeMillis();
            }
        }
        
        @Override
        public void close() throws Exception {
            // 发送剩余的数据
            sendBatch();
            super.close();
        }
    }
}

/**
 * 状态后端选择策略
 */
public class StateBackendSelection {
    
    /**
     * 根据使用场景选择状态后端
     */
    public static void configureStateBackend(StreamExecutionEnvironment env, 
                                           StateBackendType type) {
        switch (type) {
            case MEMORY:
                // 适用于：开发测试、小状态、低延迟要求
                env.setStateBackend(new MemoryStateBackend(100 * 1024 * 1024)); // 100MB
                break;
                
            case FILESYSTEM:
                // 适用于：中等状态大小、需要持久化
                env.setStateBackend(new FsStateBackend("hdfs://namenode:port/flink-checkpoints"));
                break;
                
            case ROCKSDB:
                // 适用于：大状态、高吞吐量、可以容忍稍高延迟
                RocksDBStateBackend rocksDB = new RocksDBStateBackend(
                    "hdfs://namenode:port/flink-checkpoints");
                
                // RocksDB 优化配置
                rocksDB.setDbStoragePath("/tmp/flink/rocksdb");
                rocksDB.setPredefinedOptions(PredefinedOptions.SPINNING_DISK_OPTIMIZED);
                rocksDB.enableTtlCompactionFilter();
                
                env.setStateBackend(rocksDB);
                break;
        }
    }
    
    enum StateBackendType {
        MEMORY, FILESYSTEM, ROCKSDB
    }
}

/**
 * 状态 TTL 最佳实践
 */
public class StateTtlBestPractices {
    
    /**
     * 配置状态 TTL
     */
    public static class TtlProcessFunction extends KeyedProcessFunction<String, Event, Result> {
        
        // 用户会话状态，1小时过期
        private ValueState<UserSession> sessionState;
        
        // 用户行为计数，24小时过期
        private ValueState<Long> behaviorCountState;
        
        // 临时缓存状态，5分钟过期
        private MapState<String, CacheEntry> cacheState;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            
            // 会话状态配置
            StateTtlConfig sessionTtlConfig = StateTtlConfig
                .newBuilder(Time.hours(1))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .cleanupFullSnapshot()
                .build();
            
            ValueStateDescriptor<UserSession> sessionDescriptor = 
                new ValueStateDescriptor<>("user-session", UserSession.class);
            sessionDescriptor.enableTimeToLive(sessionTtlConfig);
            sessionState = getRuntimeContext().getState(sessionDescriptor);
            
            // 行为计数状态配置
            StateTtlConfig behaviorTtlConfig = StateTtlConfig
                .newBuilder(Time.hours(24))
                .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)
                .cleanupIncrementally(10, true) // 增量清理
                .build();
            
            ValueStateDescriptor<Long> behaviorDescriptor = 
                new ValueStateDescriptor<>("behavior-count", Long.class);
            behaviorDescriptor.enableTimeToLive(behaviorTtlConfig);
            behaviorCountState = getRuntimeContext().getState(behaviorDescriptor);
            
            // 缓存状态配置
            StateTtlConfig cacheTtlConfig = StateTtlConfig
                .newBuilder(Time.minutes(5))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .cleanupInRocksdbCompactFilter(1000) // RocksDB 压缩时清理
                .build();
            
            MapStateDescriptor<String, CacheEntry> cacheDescriptor = 
                new MapStateDescriptor<>("cache", String.class, CacheEntry.class);
            cacheDescriptor.enableTimeToLive(cacheTtlConfig);
            cacheState = getRuntimeContext().getMapState(cacheDescriptor);
        }
        
        @Override
        public void processElement(Event event, Context ctx, Collector<Result> out) 
                throws Exception {
            
            // 更新会话状态
            UserSession session = sessionState.value();
            if (session == null) {
                session = new UserSession(event.getUserId(), ctx.timestamp());
            }
            session.updateLastActivity(ctx.timestamp());
            sessionState.update(session);
            
            // 更新行为计数
            Long count = behaviorCountState.value();
            behaviorCountState.update(count == null ? 1L : count + 1);
            
            // 使用缓存
            CacheEntry cached = cacheState.get(event.getKey());
            if (cached == null) {
                cached = computeExpensiveValue(event);
                cacheState.put(event.getKey(), cached);
            }
            
            out.collect(new Result(event, session, cached));
        }
        
        private CacheEntry computeExpensiveValue(Event event) {
            // 模拟昂贵的计算
            return new CacheEntry(event.getKey(), System.currentTimeMillis());
        }
    }
}

/**
 * 状态大小监控和告警
 */
public class StateMonitoring {
    
    /**
     * 状态大小监控函数
     */
    public static class StateMonitoringFunction extends KeyedProcessFunction<String, Event, Event> {
        
        private static final Logger LOG = LoggerFactory.getLogger(StateMonitoringFunction.class);
        private static final long STATE_SIZE_THRESHOLD = 100 * 1024 * 1024; // 100MB
        
        private ValueState<UserData> userDataState;
        private transient Counter stateSizeCounter;
        private transient Histogram stateSizeHistogram;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            
            ValueStateDescriptor<UserData> descriptor = 
                new ValueStateDescriptor<>("user-data", UserData.class);
            userDataState = getRuntimeContext().getState(descriptor);
            
            // 注册指标
            MetricGroup metricGroup = getRuntimeContext().getMetricGroup();
            stateSizeCounter = metricGroup.counter("state_size_bytes");
            stateSizeHistogram = metricGroup.histogram("state_size_distribution", 
                new DescriptiveStatisticsHistogram(1000));
        }
        
        @Override
        public void processElement(Event event, Context ctx, Collector<Event> out) 
                throws Exception {
            
            UserData userData = userDataState.value();
            if (userData == null) {
                userData = new UserData();
            }
            
            userData.addEvent(event);
            userDataState.update(userData);
            
            // 监控状态大小
            long stateSize = estimateStateSize(userData);
            stateSizeCounter.inc(stateSize);
            stateSizeHistogram.update(stateSize);
            
            // 状态大小告警
            if (stateSize > STATE_SIZE_THRESHOLD) {
                LOG.warn("Large state detected for key {}: {} bytes", 
                         ctx.getCurrentKey(), stateSize);
                
                // 可以触发状态清理或发送告警
                triggerStateCleanup(ctx.getCurrentKey(), userData);
            }
            
            out.collect(event);
        }
        
        private long estimateStateSize(UserData userData) {
            // 估算状态大小的简单方法
            return userData.getEvents().size() * 100; // 假设每个事件约100字节
        }
        
        private void triggerStateCleanup(String key, UserData userData) {
            // 清理旧数据
            long cutoffTime = System.currentTimeMillis() - TimeUnit.HOURS.toMillis(24);
            userData.removeEventsBefore(cutoffTime);
            
            try {
                userDataState.update(userData);
                LOG.info("Cleaned up state for key: {}", key);
            } catch (Exception e) {
                LOG.error("Failed to clean up state for key: " + key, e);
            }
        }
    }
}

/**
 * 检查点配置最佳实践
 */
public class CheckpointOptimization {
    
    /**
     * 优化检查点配置
     */
    public static void configureCheckpointing(StreamExecutionEnvironment env) {
        
        // 1. 基本检查点配置
        env.enableCheckpointing(60000); // 1分钟检查点间隔
        
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        
        // 2. 检查点模式配置
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        
        // 3. 检查点超时配置
        checkpointConfig.setCheckpointTimeout(300000); // 5分钟超时
        
        // 4. 并发检查点配置
        checkpointConfig.setMaxConcurrentCheckpoints(1); // 通常设置为1
        
        // 5. 检查点间最小间隔
        checkpointConfig.setMinPauseBetweenCheckpoints(30000); // 30秒
        
        // 6. 检查点失败容忍度
        checkpointConfig.setTolerableCheckpointFailureNumber(3);
        
        // 7. 外部化检查点配置
        checkpointConfig.enableExternalizedCheckpoints(
            CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        
        // 8. 非对齐检查点（适用于高背压场景）
        checkpointConfig.enableUnalignedCheckpoints(true);
        
        // 9. 检查点压缩
        checkpointConfig.setCheckpointStorage("hdfs://namenode:port/flink-checkpoints");
        
        // 10. 状态后端配置
        RocksDBStateBackend rocksDB = new RocksDBStateBackend(
            "hdfs://namenode:port/flink-checkpoints");
        
        // 启用增量检查点
        rocksDB.enableIncrementalCheckpointing(true);
        env.setStateBackend(rocksDB);
    }
    
    /**
     * 自定义检查点监听器
     */
    public static class CustomCheckpointListener implements CheckpointListener {
        
        private static final Logger LOG = LoggerFactory.getLogger(CustomCheckpointListener.class);
        
        @Override
        public void notifyCheckpointComplete(long checkpointId) throws Exception {
            LOG.info("Checkpoint {} completed successfully", checkpointId);
            
            // 可以在这里执行检查点完成后的清理工作
            cleanupOldData(checkpointId);
        }
        
        @Override
        public void notifyCheckpointAborted(long checkpointId) throws Exception {
            LOG.warn("Checkpoint {} was aborted", checkpointId);
            
            // 可以在这里记录检查点失败的指标
            recordCheckpointFailure(checkpointId);
        }
        
        private void cleanupOldData(long checkpointId) {
            // 清理旧的临时数据
        }
        
        private void recordCheckpointFailure(long checkpointId) {
            // 记录检查点失败指标
        }
    }
}

/**
 * 重启策略最佳实践
 */
public class RestartStrategyOptimization {
    
    /**
     * 配置重启策略
     */
    public static void configureRestartStrategy(StreamExecutionEnvironment env, 
                                              RestartStrategyType type) {
        
        switch (type) {
            case FIXED_DELAY:
                // 固定延迟重启策略 - 适用于临时故障
                env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                    3, // 重启次数
                    Time.of(10, TimeUnit.SECONDS) // 重启间隔
                ));
                break;
                
            case EXPONENTIAL_DELAY:
                // 指数退避重启策略 - 适用于可能的系统性问题
                env.setRestartStrategy(RestartStrategies.exponentialDelayRestart(
                    Time.of(1, TimeUnit.SECONDS), // 初始延迟
                    Time.of(60, TimeUnit.SECONDS), // 最大延迟
                    2.0, // 退避乘数
                    Time.of(10, TimeUnit.MINUTES), // 重置间隔
                    0.1 // 抖动因子
                ));
                break;
                
            case FAILURE_RATE:
                // 失败率重启策略 - 适用于生产环境
                env.setRestartStrategy(RestartStrategies.failureRateRestart(
                    3, // 时间间隔内最大失败次数
                    Time.of(5, TimeUnit.MINUTES), // 时间间隔
                    Time.of(10, TimeUnit.SECONDS) // 重启延迟
                ));
                break;
                
            case NO_RESTART:
                // 不重启策略 - 适用于批处理或测试
                env.setRestartStrategy(RestartStrategies.noRestart());
                break;
        }
    }
    
    enum RestartStrategyType {
        FIXED_DELAY, EXPONENTIAL_DELAY, FAILURE_RATE, NO_RESTART
    }
    
    /**
     * 自定义故障处理逻辑
     */
    public static class RobustProcessFunction extends KeyedProcessFunction<String, Event, Result> {
        
        private static final Logger LOG = LoggerFactory.getLogger(RobustProcessFunction.class);
        private static final int MAX_RETRY_ATTEMPTS = 3;
        
        private transient Counter errorCounter;
        private transient Meter errorRate;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            
            MetricGroup metricGroup = getRuntimeContext().getMetricGroup();
            errorCounter = metricGroup.counter("processing_errors");
            errorRate = metricGroup.meter("error_rate", new MeterView(errorCounter, 60));
        }
        
        @Override
        public void processElement(Event event, Context ctx, Collector<Result> out) 
                throws Exception {
            
            int attempts = 0;
            Exception lastException = null;
            
            while (attempts < MAX_RETRY_ATTEMPTS) {
                try {
                    Result result = processEventWithRetry(event);
                    out.collect(result);
                    return; // 成功处理，退出重试循环
                    
                } catch (RetryableException e) {
                    lastException = e;
                    attempts++;
                    
                    LOG.warn("Retryable error processing event {} (attempt {}): {}", 
                             event.getId(), attempts, e.getMessage());
                    
                    // 指数退避
                    try {
                        Thread.sleep(Math.min(1000 * (1L << attempts), 10000));
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        throw new RuntimeException("Interrupted during retry", ie);
                    }
                    
                } catch (NonRetryableException e) {
                    // 不可重试的异常，记录并跳过
                    LOG.error("Non-retryable error processing event {}: {}", 
                              event.getId(), e.getMessage());
                    errorCounter.inc();
                    return;
                }
            }
            
            // 重试次数用完，记录错误并可能触发故障转移
            LOG.error("Failed to process event {} after {} attempts", 
                      event.getId(), MAX_RETRY_ATTEMPTS, lastException);
            errorCounter.inc();
            
            // 可以选择抛出异常触发重启，或者发送到死信队列
            sendToDeadLetterQueue(event, lastException);
        }
        
        private Result processEventWithRetry(Event event) throws Exception {
            // 实际的事件处理逻辑
            if (Math.random() < 0.1) { // 模拟 10% 的可重试错误
                throw new RetryableException("Temporary processing error");
            }
            
            if (Math.random() < 0.01) { // 模拟 1% 的不可重试错误
                throw new NonRetryableException("Permanent processing error");
            }
            
            return new Result(event.getId(), "processed");
        }
        
        private void sendToDeadLetterQueue(Event event, Exception error) {
            // 发送到死信队列的逻辑
            LOG.info("Sending event {} to dead letter queue", event.getId());
        }
    }
    
    // 自定义异常类型
    static class RetryableException extends Exception {
        public RetryableException(String message) { super(message); }
    }
    
    static class NonRetryableException extends Exception {
        public NonRetryableException(String message) { super(message); }
    }
}

/**
 * 指标监控最佳实践
 */
public class MetricsMonitoring {
    
    /**
     * 自定义指标监控函数
     */
    public static class MetricsProcessFunction extends KeyedProcessFunction<String, Event, Event> {
        
        // 计数器指标
        private transient Counter processedCounter;
        private transient Counter errorCounter;
        
        // 计量器指标
        private transient Meter throughputMeter;
        
        // 直方图指标
        private transient Histogram processingTimeHistogram;
        
        // 仪表盘指标
        private transient Gauge<Long> queueSizeGauge;
        
        // 自定义指标
        private transient Counter businessMetricCounter;
        
        private transient Queue<Event> eventQueue;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            
            MetricGroup metricGroup = getRuntimeContext().getMetricGroup();
            
            // 注册基础指标
            processedCounter = metricGroup.counter("events_processed");
            errorCounter = metricGroup.counter("processing_errors");
            throughputMeter = metricGroup.meter("throughput", new MeterView(processedCounter, 60));
            
            // 注册处理时间直方图
            processingTimeHistogram = metricGroup.histogram("processing_time_ms", 
                new DescriptiveStatisticsHistogram(1000));
            
            // 注册队列大小仪表盘
            eventQueue = new LinkedList<>();
            queueSizeGauge = metricGroup.gauge("queue_size", () -> (long) eventQueue.size());
            
            // 注册业务指标
            MetricGroup businessGroup = metricGroup.addGroup("business");
            businessMetricCounter = businessGroup.counter("important_events");
        }
        
        @Override
        public void processElement(Event event, Context ctx, Collector<Event> out) 
                throws Exception {
            
            long startTime = System.currentTimeMillis();
            
            try {
                // 添加到队列
                eventQueue.offer(event);
                
                // 处理事件
                processEvent(event);
                
                // 更新成功指标
                processedCounter.inc();
                
                // 业务指标
                if (event.isImportant()) {
                    businessMetricCounter.inc();
                }
                
                out.collect(event);
                
            } catch (Exception e) {
                // 更新错误指标
                errorCounter.inc();
                throw e;
                
            } finally {
                // 记录处理时间
                long processingTime = System.currentTimeMillis() - startTime;
                processingTimeHistogram.update(processingTime);
                
                // 从队列移除
                eventQueue.poll();
            }
        }
        
        private void processEvent(Event event) throws Exception {
            // 模拟事件处理
            Thread.sleep(10); // 模拟处理时间
            
            if (Math.random() < 0.05) { // 5% 错误率
                throw new RuntimeException("Processing error");
            }
        }
    }
    
    /**
     * 系统指标监控
     */
    public static class SystemMetricsReporter extends RichFunction {
        
        private transient ScheduledExecutorService scheduler;
        private transient Gauge<Double> cpuUsageGauge;
        private transient Gauge<Long> memoryUsageGauge;
        private transient Gauge<Long> gcTimeGauge;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            
            MetricGroup systemGroup = getRuntimeContext().getMetricGroup().addGroup("system");
            
            // CPU 使用率
            cpuUsageGauge = systemGroup.gauge("cpu_usage", this::getCpuUsage);
            
            // 内存使用量
            memoryUsageGauge = systemGroup.gauge("memory_usage_bytes", this::getMemoryUsage);
            
            // GC 时间
            gcTimeGauge = systemGroup.gauge("gc_time_ms", this::getGcTime);
            
            // 定期更新系统指标
            scheduler = Executors.newSingleThreadScheduledExecutor();
            scheduler.scheduleAtFixedRate(this::updateMetrics, 0, 30, TimeUnit.SECONDS);
        }
        
        private double getCpuUsage() {
            OperatingSystemMXBean osBean = ManagementFactory.getOperatingSystemMXBean();
            if (osBean instanceof com.sun.management.OperatingSystemMXBean) {
                return ((com.sun.management.OperatingSystemMXBean) osBean).getProcessCpuLoad();
            }
            return -1.0;
        }
        
        private long getMemoryUsage() {
            MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean();
            return memoryBean.getHeapMemoryUsage().getUsed();
        }
        
        private long getGcTime() {
            long totalGcTime = 0;
            for (GarbageCollectorMXBean gcBean : ManagementFactory.getGarbageCollectorMXBeans()) {
                totalGcTime += gcBean.getCollectionTime();
            }
            return totalGcTime;
        }
        
        private void updateMetrics() {
            // 触发指标更新
        }
        
        @Override
        public void close() throws Exception {
            super.close();
            if (scheduler != null) {
                scheduler.shutdown();
            }
        }
    }
}

/**
 * 日志和调试最佳实践
 */
public class LoggingAndDebugging {
    
    /**
     * 结构化日志记录
     */
    public static class StructuredLoggingFunction extends RichMapFunction<Event, Event> {
        
        private static final Logger LOG = LoggerFactory.getLogger(StructuredLoggingFunction.class);
        private static final Marker BUSINESS_MARKER = MarkerFactory.getMarker("BUSINESS");
        private static final Marker PERFORMANCE_MARKER = MarkerFactory.getMarker("PERFORMANCE");
        
        private transient ObjectMapper objectMapper;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            this.objectMapper = new ObjectMapper();
        }
        
        @Override
        public Event map(Event event) throws Exception {
            long startTime = System.nanoTime();
            
            try {
                // 业务日志
                if (LOG.isInfoEnabled()) {
                    Map<String, Object> logData = new HashMap<>();
                    logData.put("eventId", event.getId());
                    logData.put("eventType", event.getType());
                    logData.put("userId", event.getUserId());
                    logData.put("timestamp", event.getTimestamp());
                    logData.put("subtaskIndex", getRuntimeContext().getIndexOfThisSubtask());
                    
                    LOG.info(BUSINESS_MARKER, "Processing event: {}", 
                             objectMapper.writeValueAsString(logData));
                }
                
                // 处理事件
                Event processedEvent = processEvent(event);
                
                // 性能日志
                long processingTime = System.nanoTime() - startTime;
                if (processingTime > 1_000_000) { // 超过1ms记录性能日志
                    LOG.warn(PERFORMANCE_MARKER, 
                             "Slow processing detected: eventId={}, processingTime={}ms", 
                             event.getId(), processingTime / 1_000_000.0);
                }
                
                return processedEvent;
                
            } catch (Exception e) {
                // 错误日志
                LOG.error("Failed to process event: eventId={}, error={}", 
                          event.getId(), e.getMessage(), e);
                throw e;
            }
        }
        
        private Event processEvent(Event event) {
            // 事件处理逻辑
            return event;
        }
    }
    
    /**
     * 调试工具函数
     */
    public static class DebuggingUtils {
        
        /**
         * 数据流调试函数
         */
        public static <T> SingleOutputStreamOperator<T> debug(DataStream<T> stream, 
                                                              String debugName) {
            return stream.map(new DebugMapFunction<>(debugName));
        }
        
        /**
         * 调试 MapFunction
         */
        private static class DebugMapFunction<T> extends RichMapFunction<T, T> {
            
            private final String debugName;
            private transient long elementCount;
            private transient long lastLogTime;
            
            public DebugMapFunction(String debugName) {
                this.debugName = debugName;
            }
            
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                this.elementCount = 0;
                this.lastLogTime = System.currentTimeMillis();
            }
            
            @Override
            public T map(T element) throws Exception {
                elementCount++;
                
                long currentTime = System.currentTimeMillis();
                if (currentTime - lastLogTime > 10000) { // 每10秒输出一次
                    LOG.info("Debug [{}]: Processed {} elements, current element: {}", 
                             debugName, elementCount, element);
                    lastLogTime = currentTime;
                }
                
                return element;
            }
        }
        
        /**
         * 数据采样调试
         */
        public static <T> SingleOutputStreamOperator<T> sample(DataStream<T> stream, 
                                                               double sampleRate, 
                                                               String sampleName) {
            return stream.filter(new SampleFilterFunction<>(sampleRate, sampleName));
        }
        
        private static class SampleFilterFunction<T> extends RichFilterFunction<T> {
            
            private final double sampleRate;
            private final String sampleName;
            private transient Random random;
            
            public SampleFilterFunction(double sampleRate, String sampleName) {
                this.sampleRate = sampleRate;
                this.sampleName = sampleName;
            }
            
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                this.random = new Random();
            }
            
            @Override
            public boolean filter(T element) throws Exception {
                boolean sampled = random.nextDouble() < sampleRate;
                if (sampled) {
                    LOG.info("Sample [{}]: {}", sampleName, element);
                }
                return sampled;
            }
        }
    }
}