当前位置：首页 > news >正文

实时数据处理：Apache Kafka与Flink实战

news 2026/5/22 13:57:59

实时数据处理：Apache Kafka与Flink实战

大家好，我是欧阳瑞（Rich Own）。今天想和大家聊聊实时数据处理这个重要话题。作为一个全栈开发者，实时数据处理已经成为现代应用的核心能力。今天就来分享一下Apache Kafka和Flink的实战经验。

实时数据处理概述

应用场景

场景	说明
实时监控	实时日志分析、监控告警
实时推荐	个性化推荐系统
实时计算	实时统计、实时报表
实时风控	欺诈检测、异常识别

技术选型

消息队列 → Kafka/RabbitMQ 实时计算 → Flink/Spark Streaming 消息存储 → Kafka/Pulsar

Apache Kafka

核心概念

概念	说明
Topic	消息主题
Partition	分区
Producer	生产者
Consumer	消费者
Consumer Group	消费者组

生产者配置

const { Kafka } = require('kafkajs'); const kafka = new Kafka({ clientId: 'my-app', brokers: ['localhost:9092'] }); const producer = kafka.producer(); async function produce() { await producer.connect(); await producer.send({ topic: 'user-events', messages: [ { value: JSON.stringify({ userId: '1', event: 'login' }) }, { value: JSON.stringify({ userId: '2', event: 'purchase' }) } ] }); await producer.disconnect(); }

消费者配置

const consumer = kafka.consumer({ groupId: 'my-group' }); async function consume() { await consumer.connect(); await consumer.subscribe({ topic: 'user-events', fromBeginning: true }); await consumer.run({ eachMessage: async ({ topic, partition, message }) => { console.log({ value: message.value.toString() }); } }); }

Apache Flink

核心概念

概念	说明
DataStream	数据流
Window	窗口
Operator	算子
State	状态

Flink作业示例

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; public class KafkaFlinkJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<String> stream = env.addSource(new FlinkKafkaConsumer<>( "user-events", new SimpleStringSchema(), properties )); stream .map(json -> { UserEvent event = parseJson(json); return event; }) .keyBy(event -> event.userId) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .count() .print(); env.execute("Kafka Flink Job"); } }

实时计算案例

// 使用flink-streaming-java的JavaScript API const { StreamExecutionEnvironment } = require('flink-streaming-java'); const env = StreamExecutionEnvironment.getExecutionEnvironment(); env.fromCollection([1, 2, 3, 4, 5]) .map(x => x * 2) .filter(x => x > 5) .print(); env.execute('Simple Job');

实战案例：实时用户行为分析

Kafka → Flink → Redis → Dashboard 1. 用户行为数据写入Kafka 2. Flink消费Kafka，计算实时统计 3. 将结果写入Redis 4. Dashboard从Redis读取数据展示

// Flink处理逻辑 const stream = env.addSource(kafkaConsumer); stream .map(record => JSON.parse(record.value())) .keyBy(record => record.userId) .window(SlidingEventTimeWindows.of(Time.seconds(30), Time.seconds(10))) .aggregate( () => ({ count: 0, events: [] }), (acc, record) => { acc.count++; acc.events.push(record); return acc; }, (key, window, aggregates) => { return { userId: key, count: aggregates.count }; } ) .addSink(redisSink);