<properties><maven.compiler.source>11</maven.compiler.source><maven.compiler.target>11</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><flink.version>2.0.0</flink.version></properties><dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-core</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><version>${flink.version}</version></dependency></dependencies>

創建WordCount 類

package com.neilparker;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;public class WordCount {public static void main(String[] args) throws Exception {StreamExecutionEnvironment executionEnvironment =  StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> source = executionEnvironment.socketTextStream("localhost1",7777,"\n");SingleOutputStreamOperator<Tuple2<String, Long>> dataStream = source.flatMap(new FlatMapFunction<String, Tuple2<String,Long>>(){@Overridepublic void flatMap (String string, Collector<Tuple2<String, Long>> collector) {String[] splits = string.split("\\s");for (String word : splits) {collector.collect(Tuple2.of(word, 1L));}}}).keyBy(value -> value.f0).sum(1);dataStream.print();executionEnvironment.execute("wordcount batch process");}}

啟動nc 命令 模擬一個 Socket Server ，

然后運行java 代碼，

然后再nc 命令行發送數據

然后就可以看到nc 命令行如下：

[root@localhost1 ~]# nc -lp 7777
hello neil hello jack
hello mike hello walker
hello sun

Java代碼控制臺看到單詞統計結果：

5> (hello,1)
15> (neil,1)
14> (jack,1)
5> (hello,2)
4> (mike,1)
9> (walker,1)
5> (hello,3)
5> (hello,4)
15> (sun,1)
5> (hello,5)

 <build><plugins><!-- 打jar插件 --><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><version>2.4.3</version><executions><execution><phase>package</phase><goals><goal>shade</goal></goals><configuration><filters><filter><artifact>*:*</artifact><excludes><exclude>META-INF/*.SF</exclude><exclude>META-INF/*.DSA</exclude><exclude>META-INF/*.RSA</exclude></excludes></filter></filters></configuration></execution></executions></plugin></plugins></build>

然后maven package