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

Flink Stream API 源码走读 - socketTextStream

ops 2025/8/15 21:15:56

概述

本文深入分析了 Flink 中 socketTextStream() 方法的源码实现，从用户API调用到最终返回 DataStream 的完整流程。

核心知识点

1. socketTextStream 方法重载链

// 用户调用入口
env.socketTextStream("hostname", 9999)↓ 补充分隔符参数
env.socketTextStream("hostname", 9999, "\n") ↓ 补充重试次数参数
env.socketTextStream("hostname", 9999, "\n", 0)↓ 创建 SocketTextStreamFunction
addSource(new SocketTextStreamFunction(hostname, port, delimiter, maxRetry), "Socket Stream")

重载过程分析：

第一层：补充分隔符参数（默认 “\n”）
第二层：补充重试次数参数（默认 0）
最终：创建 SocketTextStreamFunction 并调用 addSource

2. SourceFunction 的重要说明

@Deprecated
public class SocketTextStreamFunction implements SourceFunction<String>

⚠️ 重要提醒：

SourceFunction 已被标记为 @Deprecated（过时）
官方建议使用新的 Source API
基于 SourceFunction 的架构是老架构
新架构基于 org.apache.flink.api.connector.source.Source

3. addSource 方法的重载链

参数补充过程：

addSource(function, "Socket Stream")
addSource(function, "Socket Stream", null) - 补充 TypeInformation 为 null
addSource(function, "Socket Stream", null, CONTINUOUS_UNBOUNDED) - 补充有界性

4. 核心处理逻辑分析

private <OUT> DataStreamSource<OUT> addSource(final SourceFunction<OUT> function,final String sourceName,@Nullable final TypeInformation<OUT> typeInfo,final Boundedness boundedness) {// 1. 非空检查checkNotNull(function);checkNotNull(sourceName);checkNotNull(boundedness);// 2. 抽取类型信息TypeInformation<OUT> resolvedTypeInfo = getTypeInfo(function, sourceName, SourceFunction.class, typeInfo);// 3. 判断是否并行boolean isParallel = function instanceof ParallelSourceFunction;// 4. 序列化检查clean(function);// 5. Function → Operatorfinal StreamSource<OUT, ?> sourceOperator = new StreamSource<>(function);// 6. 返回 DataStreamSourcereturn new DataStreamSource<>(this, resolvedTypeInfo, sourceOperator, isParallel, sourceName, boundedness);
}

5. 四个核心概念的转换

概念解释：

Function: 用户的业务逻辑封装
- SocketTextStreamFunction - Socket连接和数据读取逻辑
- 继承自 SourceFunction<String>
Operator: 算子的抽象
- StreamSource<OUT, ?> - 将Function包装成算子
- 继承自 AbstractUdfStreamOperator
Transformation: 转换操作的封装
- LegacySourceTransformation - 包装Operator和相关元信息
- 包含类型信息、并行度、有界性等
DataStream: 面向用户的流式API
- DataStreamSource - 继承自 DataStream
- 支持链式调用（map、filter、keyBy等）

6. 重要参数说明

TypeInformation（类型信息）

// 为什么需要 TypeInformation？
// Java 泛型在编译后会被类型擦除，Flink需要显式的类型信息来：
// 1. 创建序列化器/反序列化器
// 2. 根据不同类型产生不同的序列化机制
TypeInformation<OUT> resolvedTypeInfo = getTypeInfo(function, sourceName, SourceFunction.class, typeInfo);

想更好理解，可以去看我这篇文章：Java 泛型类型擦除

Boundedness（有界性）

// CONTINUOUS_UNBOUNDED 表示无界流
// 在翻译成物理执行计划时会用到这个信息
// 有界流和无界流会生成不同的执行计划
Boundedness.CONTINUOUS_UNBOUNDED

并行性检查

// 检查是否为并行源函数
boolean isParallel = function instanceof ParallelSourceFunction;
// SocketTextStreamFunction 不是 ParallelSourceFunction，所以 isParallel = false

7. DataStreamSource 的构造

public DataStreamSource(StreamExecutionEnvironment environment,TypeInformation<T> outTypeInfo, StreamSource<T, ?> operator,boolean isParallel,String sourceName,Boundedness boundedness) {// 调用父类构造，创建 LegacySourceTransformationsuper(environment, new LegacySourceTransformation<>(sourceName, operator, outTypeInfo, environment.getParallelism(), boundedness));// 如果不是并行的，设置并行度为1if (!isParallel) {setParallelism(1);}
}

8. 继承关系分析

重要理解：

DataStreamSource 本质上就是一个 DataStream
所有的链式调用API都定义在 DataStream 中
SingleOutputStreamOperator 这个命名容易误导，它实际上是个 DataStream

9. DataStream 的内部结构

public class DataStream<T> {// 两个最重要的成员protected final StreamExecutionEnvironment environment;  // 执行环境protected final Transformation<T> transformation;        // 转换操作
}

关系链：

DataStream 包含 Transformation
Transformation 包含 Operator
Operator 包含 Function

10. 链式调用的实现

DataStream<String> stream = env.socketTextStream("localhost", 9999).map(...)           // 返回 SingleOutputStreamOperator (实际是DataStream).filter(...)        // 返回 SingleOutputStreamOperator  .keyBy(...)         // 返回 KeyedStream.window(...)        // 返回 WindowedStream.sum(...)           // 返回 SingleOutputStreamOperator.print();          // 返回 DataStreamSink