Centos实时数据处理（手把手教你搭建Linux环境下的实时流式计算系统）

一、准备工作：安装CentOS并配置基础环境

首先，确保你有一台运行 CentOS 7 或 CentOS 8 的服务器（物理机或虚拟机均可）。登录后，执行以下命令更新系统并安装必要工具：

# 更新系统sudo yum update -y# 安装常用工具sudo yum install -y wget curl vim git net-tools# 安装 Java（很多流处理框架依赖 Java）sudo yum install -y java-1.8.0-openjdk-devel# 验证 Java 安装java -version

二、安装 Apache Kafka（消息队列）

Kafka 是构建Linux实时流处理系统的基石，它负责高效地接收和分发数据流。我们先下载并配置 Kafka：

# 下载 Kafka（以 3.3.1 版本为例）wget https://downloads.apache.org/kafka/3.3.1/kafka_2.13-3.3.1.tgz# 解压tar -xzf kafka_2.13-3.3.1.tgzmv kafka_2.13-3.3.1 /opt/kafka# 启动 ZooKeeper（Kafka 依赖）/opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties# 启动 Kafka Server/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

现在 Kafka 已在后台运行，监听 9092 端口。

三、创建模拟日志数据流

为了演示Centos流式计算，我们用一个简单的 Shell 脚本每秒生成一条日志，并发送到 Kafka 主题：

# 创建 Kafka 主题/opt/kafka/bin/kafka-topics.sh --create --topic logs --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1# 编写模拟日志脚本vim /tmp/generate_logs.sh

在脚本中输入以下内容：

#!/bin/bashwhile true; do  echo "$(date '+%Y-%m-%d %H:%M:%S') INFO User login from IP $(shuf -i 192.168.1.1-192.168.1.254 -n 1)"  sleep 1done | /opt/kafka/bin/kafka-console-producer.sh --topic logs --bootstrap-server localhost:9092

赋予执行权限并后台运行：

chmod +x /tmp/generate_logs.shnohup /tmp/generate_logs.sh > /dev/null 2>&1 &

四、使用 Kafka Streams 进行实时处理

接下来，我们用 Java 编写一个简单的 Kafka Streams 应用，实时统计每分钟的登录次数：

// 文件：LogAnalyzer.javaimport org.apache.kafka.streams.*;import org.apache.kafka.streams.kstream.*;import java.time.Duration;import java.util.Properties;public class LogAnalyzer {    public static void main(String[] args) {        Properties props = new Properties();        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "log-analyzer");        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());        StreamsBuilder builder = new StreamsBuilder();        KStream source = builder.stream("logs");        // 提取时间窗口内的登录事件        source            .filter((key, value) -> value.contains("User login"))            .groupByKey()            .windowedBy(TimeWindows.of(Duration.ofMinutes(1)))            .count()            .toStream()            .foreach((key, value) -> System.out.println("[实时统计] " + key + " => " + value + " 次登录"));        KafkaStreams streams = new KafkaStreams(builder.build(), props);        streams.start();        // 优雅关闭        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));    }}

编译并运行该程序，你将在终端看到每分钟的实时登录统计结果！

五、总结与进阶建议

通过本教程，你已成功在 CentOS 上搭建了一个完整的Centos实时数据处理原型系统。你可以在此基础上扩展功能，例如：

• 接入真实应用日志（如 Nginx、Tomcat）
• 使用 Flink 或 Spark Streaming 替代 Kafka Streams
• 将处理结果写入数据库（如 MySQL、Elasticsearch）
• 添加 Web 可视化界面展示实时指标

掌握Linux实时流处理技能，不仅能提升你的运维能力，还能为大数据开发打下坚实基础。赶快动手试试吧！