说明

socketTextStream-从套接字读取

分析

Flink适合处理流式数据，即：数据源源不断流入，socket套接字适合这种场景

但是，不能并发，吞吐量小，不够安全，如果网络断开，来不及处理的数据，将全部丢失

很少使用

api

DataStreamSource<String> socketTextStream(String hostname, int port)

DataStreamSource<String> socketTextStream(String hostname, int port, String delimiter)

DataStreamSource<String> socketTextStream(String hostname, int port, String delimiter, long maxRetry)

形参：

• hostname：主机名、ip
• port：端口号
• delimiter：设置定界符，由定界符分隔一个一个数据，默认是 \n
• maxRetry：重试次数，默认是0，负数表示一直重试

例子

DataStream<String> ds = env.socketTextStream("hadoop1", 9999);

底层代码

env.socketTextStream() 底层代码调用的是下面的代码：

env.addSource(
    new SocketTextStreamFunction(
        hostname,
        port,
        delimiter,
        maxRetry
    ),
    "Socket Stream");

SocketTextStreamFunction 类实现 SourceFunction 接口，详见 链接

例子

socket 服务器

需要启动 socket TCP 服务器

登录 hadoop1 ，通过 nc 命令启动 TCP服务器，如下：

nc -l 9999

需要先安装 nc，详见 链接

注意：要先启动 socket服务端，因为启动 java 要连接服务器，连不上会报错

java代码

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    DataStream<String> ds = env
            .socketTextStream("hadoop1", 9999);
    ds.print();
    env.execute("test");
}

测试

在 nc 环境下，分别输入：

hadoop flink

flink yarn

hadoop yarn

在idea控制台中，显示：

hadoop flink

flink yarn

hadoop yarn

原文出处：/show_1IX42IdFArmW.html