使用树莓派制作的远程开门器

收藏 2025-11-17

攀叶逝吭核心依赖

dependencies {
// Flink核心依赖
implementation 'org.apache.flink:flink_core:1.20.1'
implementation 'org.apache.flink:flink-streaming-java:1.20.1'
implementation 'org.apache.flink:flink-clients:1.20.1'
}

三、SocketWordCount示例详解

功能介绍
SocketWordCount 是 Flink 中的一个经典实例，它经由 Socket 接收实时数据流，统计数据流中词汇的数量，并即时输出结果。此示例尽管简洁，却涵盖了 Flink 流处理的关键组成部分：数据源链接、数据转换、并行处理和结果输出。

完整代码实现

package com.cn.daimajiangxin.flink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.util.Collector;
import java.time.Duration;

public class SocketWordCount {
    public static void main(String[] args) throws Exception {
        // 1. 建立执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 启用检查点，确保故障恢复
        env.enableCheckpointing(5000); // 每5秒创建一个检查点
        // 配置并行度
        env.setParallelism(2);
        // 2. 从Socket读取数据
        String hostname = "localhost";
        int port = 9999;
        // 支持命令行参数输入
        if (args.length > 0) {
            hostname = args[0];
        }
        if (args.length > 1) {
            port = Integer.parseInt(args[1]);
        }
        DataStream<String> text = env.socketTextStream(
            hostname,
            port,
            "\n", // 行分隔符
            0);   // 最大重试次数
        // 3. 数据转换
        DataStream<Tuple2<String, Integer>> wordCounts = text
            .flatMap(new Tokenizer())
            .keyBy(value -> value.f0)
            // 添加基于处理时间的滚动窗口计算
            .window(TumblingProcessingTimeWindows.of(Duration.ofSeconds(5)))
            // 应用sum聚合算子
            .sum(1);
        // 4. 输出结果
        wordCounts.print("Word Count");
        // 5. 启动任务
        env.execute("Socket Word Count");
    }
    // 可选：采用传统的FlatMapFunction实现方法
    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        private static final long serialVersionUID = 1L;
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            String[] words = value.toLowerCase().split("\\W+");
            for (String word : words) {
                if (word.length() > 0) {

out.collect(Tuple2.of(word, 1));
}
}
}
}
}

3. 代码解析

3.1 执行环境创建

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
此段代码构建了Flink的执行环境，并配置了并行级别为2。执行环境是所有Flink程序的起点，负责管理和调度任务的执行。

3.2 数据源连接

DataStream<String> text = env.socketTextStream(hostname, port);
此处通过socketTextStream方法从Socket接口读取文本信息。这是一类Flink提供的内置数据源连接方式，适合用于测试和展示。

3.3 数据转换

DataStream<Tuple2<String, Integer>> wordCounts = text
.flatMap(new Tokenizer())
.keyBy(value -> value.f0) // 按词汇分组
.sum(1); // 累积计数
数据转换涉及三个主要环节：
分词：利用flatMap操作符将每行文本拆分为词汇，并为每个词汇生成(word, 1)的组合
分组：采用keyBy操作符依据词汇进行分组
聚合：应用sum操作符对各词汇的数量进行累积

3.4 结果输出

wordCounts.print("Word Count");
利用print方法将结果展示在控制台上，这是一种内置的输出手段，特别适用于调试和演示。

3.5 作业启动

env.execute("Socket Word Count");
最终，通过调用execute方法启动任务。需注意，Flink程序采取懒执行模式，仅当调用execute方法时才实际开始计算过程。

四、Flink并行流处理机制

1. 并行度概念

并行度代表Flink程序中每个操作员能同时处理的任务数目。在SocketWordCount例子中，设定了全局并行度为2，即每个操作员会有2个并行实例。

2. 数据流分区策略

Flink提供了多种数据流分区策略，包括：
Forward Partitioning：维持数据分区，一个输入分区对应一个输出分区
Shuffle Partitioning：随机分配数据至下游操作员的分区
Rebalance Partitioning：循环分配数据至下游操作员的分区
Rescale Partitioning：类似于rebalance，但在本地节点内部循环
Broadcast Partitioning：向所有下游分区广播数据
Key Group Partitioning：根据键的哈希值决定分区
在SocketWordCount中，keyBy操作采用了Key Group Partitioning策略，保证相同词汇的数据被分配到同一分区处理。

3. 并行执行图解

sadmermaid-diagram
该图直观地展现了Flink并行执行的过程，涵盖：
Socket数据源连接
FlatMap操作（并行度为2）及其两个子任务
KeyBy/Sum操作（并行度为2）及其两个子任务
Print输出操作（并行度为2）

五、运行SocketWordCount

1. 准备Socket服务器

在运行SocketWordCount程序前，需先启动一个Socket服务器作为数据来源。以下是几种常见的Socket服务器建立方法：

1.1 使用netcat工具

Linux/Mac系统：
nc -lk 9999
参数解释：
-l: 表示监听模式，等待连接
-k: 表示保持连接，允许多次连接（对持续测试非常有用）
9999: 端口编号
Windows系统：
Windows有多种获得netcat的方法：
如果安装了Git，可以通过Git Bash：
nc -l -p 9999
如果安装了Windows Subsystem for Linux (WSL)：
nc -lk 9999
参数解释：
-l: 表示监听模式，等待连接
-k: 表示保持连接，允许多次连接（对持续测试非常有用）
9999: 端口编号

1.2 使用Java实现Socket服务端

如果你希望使用Java代码创建一个更为灵活的Socket服务器，可以参考以下示例：
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;
public class SimpleSocketServer {
public static void main(String[] args) {

int port = 9999;

try (ServerSocket serverSocket = new ServerSocket(port)) {

System.out.println("Socket服务已启动，监控端口: " + port);

while (true) {

try (Socket clientSocket = serverSocket.accept();

PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), true);

BufferedReader in = new BufferedReader(new InputStreamReader(System.in))) {

System.out.println("客户端已接入，输入要传输的信息（输入'exit'退出）：");

String inputLine;

while ((inputLine = in.readLine()) != null) {

if (inputLine.equalsIgnoreCase("exit")) {

break;

}

out.println(inputLine);

}

} catch (IOException e) {

System.err.println("客户端连接出现故障: " + e.getMessage());

}

} catch (IOException e) {

System.err.println("无法启动服务: " + e.getMessage());

}

该Java实现的Socket服务具备以下特性：

启动后持续监控9999端口
接收客户端连接并允许数据传输
支持通过输入'exit'断开当前客户端连接
异常处理更为全面

1.3 测试Socket连接

在启动Socket服务后，可以采用以下方式测试连接是否正常：

使用telnet客户端测试：
telnet localhost 9999
使用netcat作为客户端测试：
nc localhost 9999

1.4 常见问题与解决策略

端口被占用：
错误信息：地址已在使用或类似提示
解决策略：更改端口号，或使用lsof -i :9999（Linux/Mac）查找占用端口的进程
防火墙阻碍：
症状：服务启动但客户端无法连接
解决策略：检查系统防火墙设定，确保端口9999已开放
权限问题（Linux/Mac）：
症状：普通用户无法绑定低端口（<1024）
解决策略：使用sudo权限或选择1024以上的端口
Windows特殊情况：
如果nc命令不可用，可以使用上述PowerShell脚本或安装第三方netcat工具
确保Windows Defender防火墙允许连接

六、高级功能扩展

1. 添加窗口计算

引入基于处理时间的滚动窗口计算：

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
DataStream> wordCounts = text
.flatMap(new Tokenizer())
.keyBy(value -> value.f0)
.window(TumblingProcessingTimeWindows.of(Duration.ofSeconds(5)))
.sum(1);

sad20251007145023

七、常见问题与解决方案

连接被拒绝错误
问题：程序抛出Connection refused错误。
解决方案：确保Socket服务已启动，并且监控在正确的端口上。
结果不符合预期
问题：输出的单词计数结果不符合预期。
解决方案：检查分词逻辑是否准确，确保单词的大小写处理和分隔符使用恰当。
性能问题
问题：程序处理速度较慢。
解决方案：调整并行度，增加资源配置，或优化数据转换逻辑。

八、最佳实践

1. 生产环境配置

设置合理的并行度：根据集群资源和任务特点设置并行度
启用检查点：对于生产环境，启用检查点机制以确保容错能力
配置状态后端：根据数据量大小选择适当的状态后端

2. 代码优化建议

避免使用全局变量：确保函数是无状态的或正确管理状态
合理设置并行度：避免因过度并行化导致的资源浪费

九、总结与展望

扫码加我拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

栏目导航