目录大纲

• Part01-基础概念与理论知识
  • 1.1 低延迟实时计算定义与分类
  • 1.2 实时计算架构模式对比
  • 1.3 延迟指标与性能基准
• Part02-生产环境规划与建议
  • 2.1 实时计算集群架构设计
  • 2.2 资源规划与容量评估
  • 2.3 网络与存储优化策略
• Part03-生产环境项目实施方案
  • 3.1 Flink低延迟实时计算实现
  • 3.2 Spark Streaming微批处理优化
  • 3.3 Kafka流式数据处理实战
• Part04-生产案例与实战讲解
  • 4.1 实时风控系统案例
  • 4.2 实时推荐系统案例
  • 4.3 实时监控告警系统案例
• Part05-风哥经验总结与分享
  • 5.1 低延迟计算常见问题
  • 5.2 性能调优最佳实践
  • 5.3 故障排查与处理经验

Part01-基础概念与理论知识

1.1 低延迟实时计算定义与分类

低延迟实时计算是指在数据产生后极短时间内完成处理并输出结果的技术能力，通常延迟范围在毫秒级到秒级。根据延迟要求，实时计算可分为：

• 超低延迟计算：延迟<100ms，适用于高频交易、实时风控等场景
• 低延迟计算：延迟100ms-1s，适用于实时推荐、实时监控等场景
• 近实时计算：延迟1s-10s，适用于实时报表、实时分析等场景

更多视频教程www.fgedu.net.cn

1.2 实时计算架构模式对比

主流实时计算架构包括：

1.3 延迟指标与性能基准

关键延迟指标包括：

• 端到端延迟：从数据产生到结果输出的总时间
• 处理延迟：数据进入系统到处理完成的时间
• 网络延迟：数据在网络传输中的时间
• P99延迟：99%请求的延迟上限

风哥提示：生产环境中应重点关注P99延迟，而非平均延迟，因为尾部延迟对用户体验影响更大。

Part02-生产环境规划与建议

2.1 实时计算集群架构设计

生产级实时计算集群应采用分层架构：

• 接入层：Kafka集群，负责数据接入和缓冲
• 计算层：Flink/Spark集群，负责实时数据处理
• 存储层：HBase/Redis，负责结果存储和查询
• 服务层：API网关，负责结果查询和展示

更多视频教程www.fgedu.net.cn

2.2 资源规划与容量评估

资源规划需考虑以下因素：

• 数据吞吐量：每秒处理的数据量（TPS/QPS）
• 计算复杂度：每个事件的处理CPU开销
• 内存需求：状态存储和窗口计算内存占用
• 网络带宽：节点间数据传输带宽需求

2.3 网络与存储优化策略

网络优化策略：

• 使用万兆网络，减少网络延迟
• 优化网络拓扑，减少跨机架通信
• 启用网卡多队列和RSS，提升网络吞吐

存储优化策略：

• 使用NVMe SSD，提升I/O性能
• 优化文件系统，使用ext4或xfs
• 合理配置磁盘RAID级别

更多视频教程www.fgedu.net.cn

Part03-生产环境项目实施方案

3.1 Flink低延迟实时计算实现

Flink是实现低延迟实时计算的理想选择，以下是一个实时风控系统的实现示例：

// Flink实时风控系统实现
public class RealTimeFraudDetection {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置低延迟配置
        env.setParallelism(4);
        env.enableCheckpointing(1000); // 1秒一次checkpoint
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        // 从Kafka读取交易数据
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
        kafkaProps.setProperty("group.id", "fraud-detection-group");
        kafkaProps.setProperty("auto.offset.reset", "latest");

        FlinkKafkaConsumer kafkaSource = new FlinkKafkaConsumer<>(
            "transactions",
            new TransactionDeserializer(),
            kafkaProps
        );

        DataStream transactions = env.addSource(kafkaSource);

        // 实时风控处理
        DataStream alerts = transactions
            .keyBy(Transaction::getUserId)
            .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
            .process(new FraudDetectionProcessFunction());

        // 输出告警到Kafka
        FlinkKafkaProducer kafkaSink = new FlinkKafkaProducer<>(
            "fraud-alerts",
            new AlertSerializer(),
            kafkaProps,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );

        alerts.addSink(kafkaSink);

        // 执行作业
        env.execute("Real-Time Fraud Detection");
    }
}

// 风险检测处理函数
public class FraudDetectionProcessFunction extends ProcessWindowFunction<
    Transaction, Alert, String, TimeWindow> {

    @Override
    public void process(String userId,
                       Context ctx,
                       Iterable transactions,
                       Collector out) {
        List txList = new ArrayList<>();
        double totalAmount = 0;
        int txCount = 0;

        for (Transaction tx : transactions) {
            txList.add(tx);
            totalAmount += tx.getAmount();
            txCount++;
        }

        // 风险规则1：短时间内大额交易
        if (totalAmount > 10000) {
            Alert alert = new Alert();
            alert.setUserId(userId);
            alert.setAlertType("LARGE_AMOUNT");
            alert.setAmount(totalAmount);
            alert.setTimestamp(System.currentTimeMillis());
            alert.setDescription("10分钟内交易金额超过1万元");
            out.collect(alert);
        }

        // 风险规则2：高频交易
        if (txCount > 5) {
            Alert alert = new Alert();
            alert.setUserId(userId);
            alert.setAlertType("HIGH_FREQUENCY");
            alert.setTxCount(txCount);
            alert.setTimestamp(System.currentTimeMillis());
            alert.setDescription("10分钟内交易次数超过5次");
            out.collect(alert);
        }

        // 风险规则3：异常地点交易
        Set locations = txList.stream()
            .map(Transaction::getLocation)
            .collect(Collectors.toSet());
        if (locations.size() > 2) {
            Alert alert = new Alert();
            alert.setUserId(userId);
            alert.setAlertType("ABNORMAL_LOCATION");
            alert.setLocationCount(locations.size());
            alert.setTimestamp(System.currentTimeMillis());
            alert.setDescription("10分钟内在多个不同地点交易");
            out.collect(alert);
        }
    }
}

更多视频教程www.fgedu.net.cn

3.2 Spark Streaming微批处理优化

Spark Streaming通过微批处理实现近实时计算，以下是一个实时推荐系统的实现：

// Spark Streaming实时推荐系统
public class RealTimeRecommendation {
    public static void main(String[] args) throws InterruptedException {
        // 创建Spark配置
        SparkConf conf = new SparkConf()
            .setAppName("RealTimeRecommendation")
            .setMaster("yarn")
            .set("spark.streaming.backpressure.enabled", "true")
            .set("spark.streaming.backpressure.initialRate", "10000")
            .set("spark.streaming.kafka.maxRatePerPartition", "5000")
            .set("spark.streaming.blockInterval", "200ms");

        // 创建StreamingContext，批处理间隔1秒
        JavaStreamingContext ssc = new JavaStreamingContext(conf, Durations.seconds(1));

        // 设置checkpoint目录
        ssc.checkpoint("hdfs://fgedu.net.cn:9000/checkpoint/recommendation");

        // Kafka参数配置
        Map kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", "recommendation-group");
        kafkaParams.put("auto.offset.reset", "latest");
        kafkaParams.put("enable.auto.commit", "false");

        Collection topics = Arrays.asList("user-behavior");

        // 创建Kafka Direct Stream
        JavaInputDStream> stream =
            KafkaUtils.createDirectStream(
                ssc,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.Subscribe(topics, kafkaParams)
            );

        // 解析用户行为数据
        JavaDStream behaviors = stream.map(record -> {
            JSONObject json = new JSONObject(record.value());
            UserBehavior behavior = new UserBehavior();
            behavior.setUserId(json.getString("userId"));
            behavior.setItemId(json.getString("itemId"));
            behavior.setAction(json.getString("action"));
            behavior.setTimestamp(json.getLong("timestamp"));
            return behavior;
        });

        // 实时推荐计算
        JavaPairDStream itemScores = behaviors
            .filter(b -> "view".equals(b.getAction()) || "click".equals(b.getAction()))
            .mapToPair(b -> new Tuple2<>(b.getItemId(), 1))
            .reduceByKeyAndWindow((a, b) -> a + b, Durations.minutes(5));

        // 生成推荐结果
        JavaDStream recommendations = itemScores
            .transform(rdd -> {
                // 获取Top10热门商品
                List> topItems = rdd.takeOrdered(10,
                    (a, b) -> b._2.compareTo(a._2));

                return rdd.context().parallelize(topItems)
                    .map(item -> {
                        Recommendation rec = new Recommendation();
                        rec.setItemId(item._1);
                        rec.setScore(item._2);
                        rec.setTimestamp(System.currentTimeMillis());
                        return rec;
                    });
            });

        // 输出推荐结果到Redis
        recommendations.foreachRDD(rdd -> {
            rdd.foreachPartition(partition -> {
                JedisPool pool = new JedisPool("redis1", 6379);
                try (Jedis jedis = pool.getResource()) {
                    while (partition.hasNext()) {
                        Recommendation rec = partition.next();
                        String key = "recommendation:hot:" + rec.getItemId();
                        jedis.zadd("recommendation:hot_items", rec.getScore(), rec.getItemId());
                    }
                }
            });
        });

        // 启动流处理
        ssc.start();
        ssc.awaitTermination();
    }
}

风哥提示：Spark Streaming的批处理间隔直接影响延迟，生产环境中通常设置为1-5秒，需要在延迟和吞吐量之间找到平衡。

3.3 Kafka流式数据处理实战

Kafka Streams提供轻量级的流式处理能力，适合简单的实时数据处理场景：

// Kafka Streams实时监控告警系统
public class RealTimeMonitoring {
    public static void main(String[] args) {
        // Kafka Streams配置
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "monitoring-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9092,kafka3:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000); // 1秒提交一次
        props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0); // 禁用缓存以降低延迟
        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);

        // 创建StreamsBuilder
        StreamsBuilder builder = new StreamsBuilder();

        // 从metrics主题读取监控数据
        KStream metricsStream = builder.stream("metrics");

        // 解析监控指标
        KStream parsedMetrics = metricsStream
            .mapValues(value -> {
                JSONObject json = new JSONObject(value);
                Metric metric = new Metric();
                metric.setServiceName(json.getString("serviceName"));
                metric.setMetricName(json.getString("metricName"));
                metric.setValue(json.getDouble("value"));
                metric.setTimestamp(json.getLong("timestamp"));
                return metric;
            });

        // 按服务分组计算
        KTable serviceStats = parsedMetrics
            .groupBy((key, metric) -> metric.getServiceName(),
                Grouped.with(Serdes.String(), new JsonSerde<>(Metric.class)))
            .aggregate(
                MetricStats::new,
                (key, metric, stats) -> {
                    stats.setServiceName(metric.getServiceName());
                    stats.setCount(stats.getCount() + 1);
                    stats.setSum(stats.getSum() + metric.getValue());
                    stats.setAvg(stats.getSum() / stats.getCount());
                    stats.setMax(Math.max(stats.getMax(), metric.getValue()));
                    stats.setMin(Math.min(stats.getMin(), metric.getValue()));
                    stats.setTimestamp(System.currentTimeMillis());
                    return stats;
                },
                Materialized.with(Serdes.String(), new JsonSerde<>(MetricStats.class))
            );

        // 异常检测
        KStream alerts = parsedMetrics
            .filter((key, metric) -> metric.getValue() > 1000)
            .mapValues(metric -> {
                Alert alert = new Alert();
                alert.setServiceName(metric.getServiceName());
                alert.setMetricName(metric.getMetricName());
                alert.setValue(metric.getValue());
                alert.setAlertType("HIGH_VALUE");
                alert.setTimestamp(System.currentTimeMillis());
                alert.setDescription("指标值超过阈值1000");
                return alert;
            });

        // 输出告警到alerts主题
        alerts.to("alerts",
            Produced.with(Serdes.String(), new JsonSerde<>(Alert.class)));

        // 创建Kafka Streams应用
        KafkaStreams streams = new KafkaStreams(builder.build(), props);

        // 添加关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

        // 启动应用
        streams.start();

        System.out.println("监控告警系统已启动");
    }
}

更多视频教程www.fgedu.net.cn

Part04-生产案例与实战讲解

4.1 实时风控系统案例

某电商平台实时风控系统，要求在交易发生后500ms内完成风险检测。系统架构如下：

• 数据接入：Kafka集群，3节点，每秒处理10万笔交易
• 实时计算：Flink集群，10个TaskManager，每个4核16G
• 规则引擎：Drools规则引擎，支持动态规则配置
• 结果存储：Redis集群，存储风险评分和告警信息

关键优化措施：

• 使用Flink的ProcessFunction实现低延迟处理
• 优化网络拓扑，减少跨节点数据传输
• 使用RocksDB状态后端，提升状态访问性能
• 合理设置checkpoint间隔，平衡一致性和延迟

生产环境中，风控系统的延迟要求非常严格，需要持续监控P99延迟，确保99%的交易在500ms内完成处理。

4.2 实时推荐系统案例

某内容平台实时推荐系统，要求在用户行为发生后2秒内更新推荐结果。系统架构如下：

• 行为采集：Kafka集群，5节点，每秒处理50万用户行为
• 实时计算：Spark Streaming集群，20个Executor，每个8核32G
• 特征存储：HBase集群，存储用户和商品特征
• 推荐引擎：TensorFlow Serving，实时计算推荐分数

性能优化策略：

• 使用Kryo序列化，减少网络传输开销
• 优化批处理间隔，设置为2秒
• 使用DataFrame API，提升计算性能
• 合理设置分区数，避免数据倾斜

更多视频教程www.fgedu.net.cn

4.3 实时监控告警系统案例

某互联网公司实时监控告警系统，要求在指标异常后1秒内发出告警。系统架构如下：

• 指标采集：Prometheus + Kafka，每秒采集100万指标
• 实时计算：Kafka Streams，5个实例，每个4核8G
• 告警存储：Elasticsearch，存储告警历史
• 告警通知：钉钉/企业微信，实时推送告警

技术选型考虑：

• Kafka Streams轻量级，部署简单
• 本地状态存储，访问延迟低
• 支持 Exactly-Once 语义，保证数据一致性
• 与Kafka深度集成，运维成本低

风哥提示：监控告警系统的准确性比延迟更重要，需要设置合理的告警阈值，避免误报和漏报。

Part05-风哥经验总结与分享

5.1 低延迟计算常见问题

在生产环境中，低延迟计算常见问题包括：

• 延迟波动：由于GC、网络抖动等原因导致延迟不稳定
• 数据倾斜：某些key的数据量过大，导致处理延迟
• 状态膨胀：状态数据过大，导致访问延迟增加
• 背压问题：下游处理能力不足，导致上游阻塞

更多视频教程www.fgedu.net.cn

5.2 性能调优最佳实践

低延迟计算性能调优最佳实践：

• 合理设置并行度：根据数据量和计算复杂度设置合适的并行度
• 优化序列化：使用高效的序列化方式，如Kryo、Avro
• 减少网络传输：尽量减少shuffle操作，使用本地聚合
• 优化状态访问：选择合适的状态后端，如RocksDB
• 调整JVM参数：优化GC参数，减少GC停顿时间
• 监控关键指标：持续监控延迟、吞吐量、资源使用率

5.3 故障排查与处理经验

低延迟计算故障排查经验：

• 延迟突增：检查是否有数据倾斜、GC频繁、网络问题
• 吞吐量下降：检查资源使用率、是否有任务卡住
• 数据丢失：检查offset提交、checkpoint配置
• 结果不准确：检查业务逻辑、时间窗口设置

风哥提示：建立完善的监控告警体系，在问题影响业务之前及时发现和处理，是保障低延迟计算系统稳定运行的关键。

更多视频教程www.fgedu.net.cn