Flink State Checkpointing

🕗 发布于 2天前 flink 大数据

Flink State Checkpointing 详解

1. 基本概念

State Checkpointing（状态检查点）是 Flink 提供的容错机制，用于定期保存应用程序的状态快照，以便在发生故障时能够从最近的检查点恢复状态，确保 Exactly-Once 语义。

1.1 核心特性

容错保证：确保应用程序在故障后能够恢复到一致状态
Exactly-Once 语义：提供精确一次处理保证
自动管理：Flink 框架自动管理检查点的创建和恢复
可配置性：支持多种检查点配置选项

1.2 工作原理

State Checkpointing 通过以下方式工作：

定期触发检查点操作
通知所有算子保存当前状态
将状态快照持久化到存储系统
故障发生时从最近检查点恢复状态

2. 适用场景

2.1 精确一次处理

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
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.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 精确一次处理示例
 * 使用检查点确保数据处理的准确性
 */
public class ExactlyOnceProcessingExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 启用检查点，每5秒进行一次
        env.enableCheckpointing(5000);
        
        // 设置检查点模式为 EXACTLY_ONCE
        env.getCheckpointConfig().setCheckpointingMode(org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE);
        
        // 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        
        // 设置检查点最小间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000);
        
        // 设置最大并发检查点数
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        
        // 启用检查点外部持久化
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(
            org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        
        // 创建交易数据流
        DataStream<Transaction> transactions = env.fromElements(
            new Transaction("user1", 100.0, System.currentTimeMillis()),
            new Transaction("user2", 200.0, System.currentTimeMillis() + 1000),
            new Transaction("user1", 150.0, System.currentTimeMillis() + 2000),
            new Transaction("user3", 300.0, System.currentTimeMillis() + 3000)
        );
        
        // 处理交易并维护用户余额状态
        DataStream<String> results = transactions
            .keyBy(transaction -> transaction.userId)
            .map(new BalanceManager());
        
        results.print();
        
        env.execute("Exactly Once Processing Example");
    }
    
    /**
     * 余额管理器
     * 使用检查点确保余额计算的准确性
     */
    public static class BalanceManager extends RichMapFunction<Transaction, String> {
        private ValueState<Double> balanceState;
        
        @Override
        public void open(Configuration parameters) {
            ValueStateDescriptor<Double> descriptor = new ValueStateDescriptor<>(
                "user-balance",
                Double.class,
                0.0
            );
            balanceState = getRuntimeContext().getState(descriptor);
        }
        
        @Override
        public String map(Transaction transaction) throws Exception {
            Double currentBalance = balanceState.value();
            Double newBalance = currentBalance + transaction.amount;
            balanceState.update(newBalance);
            
            return "User " + transaction.userId + " balance: " + newBalance;
        }
    }
    
    /**
     * 交易记录
     */
    public static class Transaction {
        public String userId;
        public double amount;
        public long timestamp;
        
        public Transaction() {}
        
        public Transaction(String userId, double amount, long timestamp) {
            this.userId = userId;
            this.amount = amount;
            this.timestamp = timestamp;
        }
    }
}

2.2 状态持久化

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 状态持久化示例
 * 配置不同的状态后端以实现状态持久化
 */
public class StatePersistenceExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 启用检查点
        env.enableCheckpointing(10000);
        
        // 配置状态后端和检查点存储
        configureStateBackend(env, "rocksdb"); // 可选: "hashmap", "rocksdb"
        
        // 创建用户数据流
        DataStream<UserEvent> userEvents = env.fromElements(
            new UserEvent("user1", "login", System.currentTimeMillis()),
            new UserEvent("user2", "login", System.currentTimeMillis() + 1000),
            new UserEvent("user1", "purchase", System.currentTimeMillis() + 2000),
            new UserEvent("user3", "login", System.currentTimeMillis() + 3000)
        );
        
        // 处理用户事件并维护状态
        DataStream<String> results = userEvents
            .keyBy(event -> event.userId)
            .map(new UserActivityTracker());
        
        results.print();
        
        env.execute("State Persistence Example");
    }
    
    /**
     * 配置状态后端
     */
    public static void configureStateBackend(StreamExecutionEnvironment env, String backendType) {
        switch (backendType.toLowerCase()) {
            case "hashmap":
                env.setStateBackend(new HashMapStateBackend());
                env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoints");
                break;
                
            case "rocksdb":
                env.setStateBackend(new EmbeddedRocksDBStateBackend());
                env.getCheckpointConfig().setCheckpointStorage("hdfs://namenode:port/flink/checkpoints");
                break;
                
            default:
                throw new IllegalArgumentException("Unknown backend type: " + backendType);
        }
    }
    
    /**
     * 用户活动跟踪器
     */
    public static class UserActivityTracker extends RichMapFunction<UserEvent, String> {
        private ValueState<Integer> loginCountState;
        private ValueState<Integer> purchaseCountState;
        
        @Override
        public void open(Configuration parameters) {
            loginCountState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("login-count", Integer.class, 0)
            );
            
            purchaseCountState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("purchase-count", Integer.class, 0)
            );
        }
        
        @Override
        public String map(UserEvent event) throws Exception {
            if ("login".equals(event.eventType)) {
                Integer count = loginCountState.value();
                loginCountState.update(count + 1);
                return "User " + event.userId + " login count: " + (count + 1);
            } else if ("purchase".equals(event.eventType)) {
                Integer count = purchaseCountState.value();
                purchaseCountState.update(count + 1);
                return "User " + event.userId + " purchase count: " + (count + 1);
            }
            
            return "User " + event.userId + " performed " + event.eventType;
        }
    }
    
    /**
     * 用户事件
     */
    public static class UserEvent {
        public String userId;
        public String eventType;
        public long timestamp;
        
        public UserEvent() {}
        
        public UserEvent(String userId, String eventType, long timestamp) {
            this.userId = userId;
            this.eventType = eventType;
            this.timestamp = timestamp;
        }
    }
}

3. 检查点配置

3.1 基本配置

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 检查点基本配置示例
 */
public class BasicCheckpointConfiguration {
    
    /**
     * 配置基本检查点参数
     */
    public static StreamExecutionEnvironment configureBasicCheckpointing() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 启用检查点，每10秒进行一次
        env.enableCheckpointing(10000);
        
        // 设置检查点模式
        env.getCheckpointConfig().setCheckpointingMode(
            org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE);
        
        // 设置检查点超时时间（1分钟）
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        
        // 设置检查点最小间隔（2秒）
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000);
        
        // 设置最大并发检查点数
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        
        // 设置检查点对齐
        env.getCheckpointConfig().setAlignmentTimeout(5000);
        
        return env;
    }
}

3.2 高级配置

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 检查点高级配置示例
 */
public class AdvancedCheckpointConfiguration {
    
    /**
     * 配置高级检查点参数
     */
    public static StreamExecutionEnvironment configureAdvancedCheckpointing() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 启用检查点
        env.enableCheckpointing(5000);
        
        // 配置检查点模式为 AT_LEAST_ONCE（更高性能）
        env.getCheckpointConfig().setCheckpointingMode(
            org.apache.flink.streaming.api.CheckpointingMode.AT_LEAST_ONCE);
        
        // 配置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(30000);
        
        // 配置检查点最小间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);
        
        // 配置最大并发检查点数
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        
        // 配置未对齐检查点（Flink 1.11+）
        env.getCheckpointConfig().enableUnalignedCheckpoints();
        
        // 配置检查点外部持久化
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(
            org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        
        // 配置检查点存储
        env.getCheckpointConfig().setCheckpointStorage("hdfs://namenode:port/flink/checkpoints");
        
        // 配置本地恢复
        env.getCheckpointConfig().setLocalRecovery(true);
        
        return env;
    }
}

3.3 检查点存储配置

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 检查点存储配置示例
 */
public class CheckpointStorageConfiguration {
    
    /**
     * 配置文件系统检查点存储
     */
    public static StreamExecutionEnvironment configureFileSystemStorage() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        
        // 配置 HDFS 存储
        env.getCheckpointConfig().setCheckpointStorage("hdfs://namenode:port/flink/checkpoints");
        
        return env;
    }
    
    /**
     * 配置 RocksDB 增量检查点存储
     */
    public static StreamExecutionEnvironment configureRocksDBIncrementalStorage() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        
        // 启用增量检查点
        env.getCheckpointConfig().setCheckpointStorage(
            "filesystem://hdfs://namenode:port/flink/checkpoints");
        
        return env;
    }
    
    /**
     * 配置自定义检查点存储
     */
    public static StreamExecutionEnvironment configureCustomStorage() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        
        // 配置自定义检查点存储
        // env.getCheckpointConfig().setCheckpointStorage(new CustomCheckpointStorage());
        
        return env;
    }
}

4. 完整使用示例

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 完整的检查点使用示例
 */
public class CompleteCheckpointingExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 配置检查点
        configureCheckpointing(env);
        
        // 创建传感器数据流
        DataStream<SensorReading> sensorData = env.addSource(new SensorSource());
        
        // 处理传感器数据并维护状态
        DataStream<String> results = sensorData
            .keyBy(reading -> reading.sensorId)
            .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
            .process(new SensorDataProcessor());
        
        results.print();
        
        env.execute("Complete Checkpointing Example");
    }
    
    /**
     * 配置检查点
     */
    public static void configureCheckpointing(StreamExecutionEnvironment env) {
        // 启用检查点，每5秒进行一次
        env.enableCheckpointing(5000);
        
        // 设置检查点模式
        env.getCheckpointConfig().setCheckpointingMode(
            org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE);
        
        // 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        
        // 设置检查点最小间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000);
        
        // 设置最大并发检查点数
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        
        // 配置状态后端
        env.setStateBackend(new EmbeddedRocksDBStateBackend());
        
        // 配置检查点存储
        env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoints");
        
        // 配置检查点外部持久化
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(
            org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
    }
    
    /**
     * 传感器数据源
     */
    public static class SensorSource extends RichParallelSourceFunction<SensorReading> implements CheckpointedFunction {
        private volatile boolean isRunning = true;
        private long sequenceNumber = 0;
        
        @Override
        public void run(SourceContext<SensorReading> ctx) throws Exception {
            while (isRunning) {
                synchronized (ctx.getCheckpointLock()) {
                    // 生成传感器读数
                    SensorReading reading = new SensorReading(
                        "sensor-" + (getRuntimeContext().getIndexOfThisSubtask() + 1),
                        System.currentTimeMillis(),
                        Math.random() * 100
                    );
                    
                    ctx.collect(reading);
                    sequenceNumber++;
                    
                    Thread.sleep(1000); // 每秒生成一个读数
                }
            }
        }
        
        @Override
        public void cancel() {
            isRunning = false;
        }
        
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            // 在检查点时保存状态
            System.out.println("Checkpoint taken at sequence number: " + sequenceNumber);
        }
        
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            // 初始化状态
            if (context.isRestored()) {
                System.out.println("Restoring from checkpoint");
            }
        }
    }
    
    /**
     * 传感器数据处理器
     */
    public static class SensorDataProcessor extends ProcessWindowFunction<SensorReading, String, String, TimeWindow> {
        @Override
        public void process(String sensorId, Context context, Iterable<SensorReading> readings, Collector<String> out) {
            double sum = 0;
            int count = 0;
            
            for (SensorReading reading : readings) {
                sum += reading.value;
                count++;
            }
            
            double average = count > 0 ? sum / count : 0;
            
            out.collect("Sensor " + sensorId + " average: " + String.format("%.2f", average) + 
                       " (based on " + count + " readings)");
        }
    }
    
    /**
     * 传感器读数
     */
    public static class SensorReading {
        public String sensorId;
        public long timestamp;
        public double value;
        
        public SensorReading() {}
        
        public SensorReading(String sensorId, long timestamp, double value) {
            this.sensorId = sensorId;
            this.timestamp = timestamp;
            this.value = value;
        }
        
        @Override
        public String toString() {
            return "SensorReading{sensorId='" + sensorId + "', timestamp=" + timestamp + ", value=" + value + "}";
        }
    }
}

5. 检查点恢复

5.1 从检查点恢复

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 从检查点恢复示例
 */
public class CheckpointRecoveryExample {
    
    /**
     * 从保存的检查点恢复作业
     */
    public static void recoverFromCheckpoint() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 启用检查点
        env.enableCheckpointing(5000);
        
        // 配置从检查点恢复
        env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoints");
        
        // 如果存在保存的检查点，Flink 会自动从中恢复
        
        // 创建数据流处理逻辑
        // ... 作业逻辑 ...
        
        env.execute("Recovered Job");
    }
}

5.2 保存点恢复

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 从保存点恢复示例
 */
public class SavepointRecoveryExample {
    
    /**
     * 从保存点恢复作业
     */
    public static void recoverFromSavepoint(String savepointPath) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 从指定保存点恢复
        env.executeFromSavepoint(savepointPath);
        
        // 创建数据流处理逻辑
        // ... 作业逻辑 ...
        
        env.execute("Job from Savepoint");
    }
}

6. 监控和调优

6.1 检查点监控

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 检查点监控示例
 */
public class CheckpointMonitoringExample {
    
    /**
     * 监控检查点性能
     */
    public static void monitorCheckpoints() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        
        // 配置检查点监听器
        // env.getCheckpointConfig().setCheckpointListener(new CustomCheckpointListener());
        
        // 执行作业
        JobExecutionResult result = env.execute("Monitoring Job");
        
        // 获取检查点统计信息
        // CheckpointStats checkpointStats = result.getCheckpointStats();
        // System.out.println("Checkpoint Stats: " + checkpointStats);
    }
}

6.2 性能调优

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 检查点性能调优示例
 */
public class CheckpointPerformanceTuning {
    
    /**
     * 优化检查点性能
     */
    public static StreamExecutionEnvironment tuneCheckpointPerformance() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 启用检查点
        env.enableCheckpointing(10000);
        
        // 优化配置
        env.getCheckpointConfig().setCheckpointingMode(
            org.apache.flink.streaming.api.CheckpointingMode.AT_LEAST_ONCE);  // 使用 AT_LEAST_ONCE 模式提高性能
        
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);  // 增加检查点间隔
        
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);  // 允许并发检查点
        
        env.getCheckpointConfig().setCheckpointTimeout(120000);  // 增加超时时间
        
        // 启用未对齐检查点（如果支持）
        env.getCheckpointConfig().enableUnalignedCheckpoints();
        
        // 配置本地恢复
        env.getCheckpointConfig().setLocalRecovery(true);
        
        return env;
    }
}

7. 最佳实践建议

7.1 配置建议

检查点间隔：
- 根据业务需求和恢复时间要求设置合适的间隔
- 一般建议在秒级到分钟级之间
超时时间：
- 设置足够长的超时时间以避免检查点失败
- 考虑网络延迟和存储性能
并发检查点：
- 根据系统资源合理设置并发检查点数
- 避免过多并发影响正常处理

7.2 性能优化

状态后端选择：
- 小状态使用 HashMapStateBackend
- 大状态使用 EmbeddedRocksDBStateBackend
- 启用增量检查点以提高性能
检查点模式：
- 对于高吞吐量场景可考虑 AT_LEAST_ONCE
- 对于数据准确性要求高的场景使用 EXACTLY_ONCE
存储配置：
- 使用高性能存储系统
- 配置合适的存储路径和权限

7.3 监控和维护

监控指标：
- 监控检查点完成时间和频率
- 监控检查点失败率
- 监控状态大小和增长趋势
故障处理：
- 定期备份重要检查点
- 准备恢复方案和流程
- 测试恢复过程的有效性
容量规划：
- 根据检查点大小规划存储容量
- 监控存储使用情况
- 定期清理过期检查点

通过合理配置和使用 State Checkpointing，可以确保 Flink 应用程序在发生故障时能够快速恢复到一致状态，提供高可用性和数据一致性保证。

原文地址：https://blog.csdn.net/wudonglianga/article/details/154178820

免责声明：本站文章内容转载自网络资源，如侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：超越“盒子”：虚拟机在云计算与AI时代的颠覆性未来应用展望
下一篇：AI大模型的算力革命：从架构优化到智能自演化的未来路径

Java HashMap深度解析：数据结构、原理与实战指南
本文详细解析了Java HashMap的核心概念与使用方法。HashMap采用数组+链表/红黑树的数据结构，JDK8后引入红黑树优化查询效率。它具有键唯一、值可重复、无序存储、允许null键值等特点，
阅读更多2025-11-07
无zookeeper Kafka 4.1.0 Raft 集群搭建
实现高可用，集群若允许N个controller失败，则需要2N+1个controller组成集群。下面搭建一个3节点的Kafka集群，3个controller，3个broker。Kafka kraft
阅读更多2025-11-07
Kafka4.1.0 队列模式尝鲜
摘要：Kafka 4.1.0版本引入了新的队列消费模型（KIP-932提案），通过共享组协议实现多消费者并发处理消息。文中演示了如何在Docker中配置启用该预览功能，并使用Spring-Kafka框
阅读更多2025-11-07
SNK施努卡机器人玻璃涂胶
机器人能严格按照预设轨迹运行，确保胶条宽度、厚度和出胶速度的绝对一致性，彻底消除了人工操作中因手抖、疲劳导致的断胶、不均等问题，极大提升了产品密封性和外观质量。通过离线编程和视觉引导技术，机器人能够快
阅读更多2025-11-07
监视你的脚本：自动 Linux 活动审计
用于监控关键操作的设置，描述了一个用于分析日志的脚本，并探讨了向 Telegram 添加警报的可能性。您可以自行决定此工具的用途（如果有）。我很乐意听取您对此方法的实用价值的评论并进行讨论。在本文中，
阅读更多2025-11-07
kafka服务端与客户端如何协作？生产者发送消息分区策略是什么？消费者组分区策略？集群与ACK机制？
kafka服务端与客户端如何协作？生产者发送消息分区策略是什么？消费者组分区策略？集群与ACK机制？
阅读更多2025-11-07
2025年具身智能安全前沿：守护机器人时代的防失控策略
具身智能作为人工智能与机器人技术的融合体，在2025年已广泛应用于医疗、制造、交通等领域，但随之而来的安全风险，尤其是机器人失控问题，成为全球关注的焦点。本文深入探讨具身智能的安全性，分析潜在失控原因
阅读更多2025-11-07
中国机器人产业：迅猛崛起与未来征程
2025年前三季度，我国机器人行业营收同比增长29.5%，工业机器人产量达59.5万台，服务机器人产量突破1350万套，均已超2024年全年总量。该平台支持多方在线开展设计评审与标记，其"3
阅读更多2025-11-07
机器学习、深度学习、大模型是什么关系？
人工智能（AI）是一个综合性领域，涵盖了机器学习（ML）、深度学习（DL）和大模型三个层次，其中大模型是深度学习的扩展形态。AI不仅包括学习技术，还涉及感知、推理、决策等多个方面。AI源于计算机科学，
阅读更多2025-11-07
C++_chapter10_C++IO流类库
数据的输入和输出（input/output简写为I/O）；对标准输入设备和标准输出设备的输入输出简称为标准I/O；对在外存磁盘上文件的输入输出简称为文件I/O；对内存中指定的字符串存储空间的输入输出简
阅读更多2025-11-07