# IOT-Data-Process

**Repository Path**: Frank9981/iot-data-process

## Basic Information

- **Project Name**: IOT-Data-Process
- **Description**: 本项目是一款基于物联网三层架构与 C++11 高性能网络编程开发的智能环境数据采集与可视化系统。系统实现了从环境数据采集、高并发 TCP 传输、JSON 解析、持久化存储到 Web 前端可视化展示的完整链路。项目从初期的单机一体化架构，迭代演进至基于 Nginx 负载均衡与 Broker 消息转发的分布式架构，成功解决了高并发场景下的丢包、TCP粘包、数据解析异常及入库效率低等核心痛点
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 13
- **Forks**: 0
- **Created**: 2026-09-13
- **Last Updated**: 2026-10-17

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

 ** **# IOT-Data-Process（分布式IoT数据采集与存储系统）** ** 

#### 📌 项目简介

本项目是一款基于物联网三层架构开发的智能环境数据采集与可视化系统，采用 C++ 高性能网络编程技术，实现环境温度、湿度数据的终端采集、高并发传输、持久化存储与 Web 可视化展示。系统从单机一体化架构迭代至分布式架构，解决了传统 IoT 数据采集系统高并发丢包、数据解析异常、入库效率低等核心问题，具备高并发、高可靠、可扩展的特性，适用于小型物联网环境监测场景。

> **注意**：前期开发的场景，是编写了一套 C++ 客户端，去代替传感器。

系统严格遵循物联网经典三层架构设计，各层级技术实现如下：

**1. 感知层：多源传感器数据采集**

依托环境温度、湿度传感器，实时采集环境监测数据，封装设备编号、采集时间、温度值、湿度值等核心字段，为上层数据传输提供原始数据支撑。

**2. 传输层：基于 muduo 库的高性能 TCP 通信**

采用成熟的 muduo 高性能网络库搭建 TCP 通信架构，基于 Reactor 事件驱动模型实现非阻塞、高并发网络通信，完美适配多设备同时上报数据的场景。

- **通信模式**：传感器终端作为 TCP 客户端，与服务端建立长连接，保障数据传输的稳定性
- **数据上报**：将采集的环境数据封装为标准 JSON 数据包，通过 TCP Socket 完成实时上报

**3. 应用层：基于 httplib 的 Web 可视化服务**

搭建轻量级 HTTP Web 服务，实现数据持久化、历史数据查询、前端可视化展示等核心业务能力。

- **服务搭建**：基于 cpp-httplib 单头文件库构建轻量化 HTTP 服务器，部署简单、运行高效
- **数据存储**：对接 MySQL 8.0 数据库，持久化存储传感器采集数据，支持历史数据检索、溯源
- **前端交互**：Web 客户端通过 HTTP 接口请求后端实时、历史数据，完成数据可视化渲染
    

#### ✨ 项目亮点

- 🚀 分布式架构：Nginx 负载均衡 + Broker 轮询转发 + WriteDB 双实例
- 📈 高并发：100 并发下入库率 99.45%，真实 QPS 7769
- 🛡️ 高可靠：单点故障不影响整体服务
- 🔧 完整链路：采集 → 传输 → 解析 → 存储 → 可视化

#### 🔄 架构演进

 ##🥇V1.0 单机架构

    初期采用一体化设计，TCP收发、数据解析、数据库入库、HTTP服务全部集成在单一服务中，架构简单、部署便捷，适合基础功能验证。

```
    【传感器】 -->> 【muduo TCP客户端】 -->> 【muduo TCP服务端】-->>【Web客户端】 -->>【httplib HTTP服务器】 -->>【数据库存储MySQL】

```

 ##🥈V2.0 分布式架构
    针对单机架构高并发性能瓶颈，进行服务拆分重构，拆解网关、转发、数据存储能力，实现服务解耦、分工明确，大幅提升并发处理能力与系统稳定性。


```
                          ┌──────────────────────┐
                          │   传感器客户端 × N    │
                          └──────────┬───────────┘
                                     │  TCP
                                     ▼
                          ┌──────────────────────┐
                          │   Nginx (8888)       │
                          │   TCP 负载均衡        │
                          └──────────┬───────────┘
                                     │ 轮询
                     ┌───────────────┴───────────────┐
                     ▼                               ▼
          ┌────────────────────┐          ┌────────────────────┐
          │ ChatServer_1(8889) │          │ ChatServer_2(8890) │
          │    网关实例 1       │          │    网关实例 2       │
          └─────────┬──────────┘          └──────────┬─────────┘
                    │                                │
                    └──────────────┬─────────────────┘
                                   │  TCP
                                   ▼
                        ┌──────────────────────┐
                        │  Broker (7777)       │
                        │  消息转发（轮询）      │
                        └──────────┬───────────┘
                                   │
                     ┌─────────────┴─────────────┐
                     ▼                           ▼
          ┌────────────────────┐      ┌────────────────────┐
          │ write_db_server_1  │      │ write_db_server_2  │
          │    写库实例 1       │      │    写库实例 2       │
          └─────────┬──────────┘      └──────────┬─────────┘
                    │                            │
                    └──────────────┬─────────────┘
                                   │  SQL
                                   ▼
                        ┌──────────────────────┐
                        │      MySQL 8.0       │
                        │      数据持久化       │
                        └──────────┬───────────┘
                                   │
                                   ▼
                        ┌──────────────────────┐
                        │   httplib Web 服务    │
                        │      (8080)          │
                        └──────────┬───────────┘
                                   │  HTTP
                                   ▼
                        ┌──────────────────────┐
                        │     Web 客户端        │
                        │   浏览器可视化展示     │
                        └──────────────────────┘

```

 **架构说明：** 
   
-     1. Nginx 作为 TCP 负载均衡，将大量传感器客户端连接分发到多个 ChatServer 网关实例，分担网关压力
-     2. ChatServer 多实例接收设备上报 JSON 数据，统一转发给 Broker
-     3. Broker**轮询转发**（替代广播），一条消息只发给一个 WriteDB 实例，避免多写库进程重复入库
-     4. WriteDB 多实例并行消费消息，写入 MySQL，提升入库吞吐能力

#### 📊 性能压测

### 🖥️ 测试环境

    | 项目    | 配置                                            |
    |---------|-----------------------------------------------|
    | CPU   |                                               |
    | 内存    |                                               |
    | 操作系统  | Ubuntu22.04                                   |
    | MySQL | 8.0                                           |
    | 部署方式  | 单机Broker x 1，WriteDB Server x 2,ChatServer x1 |
    | 压测客户端 | Python客户端（模拟N台设备并发上报）                         |

### 🧪 测试方法
    使用python 多线程脚本模拟多态设备，持续向 ChatServer 上报传感器数据，压测结束后核对MySQL 入库条数

### 🚀 测试步骤
    
    V1.0:如果测试版本1，就把端口号改成8081，在保存编译
        `python3 bench.py`

    V2.0:如果测试版本2，就把端口号改成8088，在保存编译
        `python3 bench.py`

### 🏗️ 系统架构
    
    优化前：压测脚本 - TCP - MServer - MySQL
    优化后：压测脚本 - TCP - MServer - Broker - WriteDB Server x 2 - MySQL

### 📈 压测结果

    | 版本       | 并发客户端 | 发送条数  | 发送耗时（s） | 发送 QPS | 入库条数 | 入库率   |
    |----------|-------|-------|---------|--------|------|-------|
    | ❌ 未优化版（单机）| 100   | 10000 | 0.68    | 14813  | 395  | 3.95%  |
    | ✅ 优化版（分布式） | 100   | 10000 | 1.28    | 7807   | 9945 | 99.45% |

### 🔍 优化过程记录

---

#### 🥇 第一阶段：单机架构，未处理拆包

**架构**

压测脚本 ─ TCP ─▶ MServer ─▶ MySQL（单进程，收包与写库耦合）

**问题**

初次压测发现 MySQL 入库条目远少于发送条目，丢失率高达 **96%**。

**排查**

查看 MServer 日志，发现大量 JSON 解析失败。

**根因**

客户端连续发送多条 JSON，TCP 层发生**粘包**。服务端 `onMessage` 中使用 `retrieveAllAsString()` 一次性读取整个缓冲区，导致多条消息被拼成一条，JSON 解析失败后直接丢弃。

**现象**

发送 QPS 虚高至 **14813**，实为"漏得多所以跑得快"，真实入库 QPS 仅 **581**。

---

#### 🥈 第二阶段：分布式架构改造，统一消息边界

**架构**

压测脚本 ─ TCP ─▶ MServer ─▶ Broker ─▶ WriteDB Server × 2 ─▶ MySQL

**优化**

1. 全结构改造消息分割机制：客户端与服务端统一使用 `'\n'` 作为消息边界，服务端 `onMessage` 改为**按行循环拆包**。
2. 将收包与写库解耦：引入 **Broker** 做消息转发，MServer 不再被 MySQL 拖慢。
3. WriteDB Server 独立部署**两个实例**，由 Broker 轮询负载均衡转发。

**说明**

设计改造的进程包括 **MClient、MServer、Broker、WriteDB Server** 四处，全部统一分隔符。

---

### 🎯 优化效果对比
    | 指标   | ❌ 优化前        | ✅ 优化后        | 🚀 提升    |
    |------|------------|------------|-------|
    | 入库条数 | 395/10000 | 9945/10000 |  **+2418%**  |
    | 入库率  | 3.95%%        | 99.45%      | +95.5个百分点      |
    | 真实 QPS | 581          | 7769          |  **+1237%**  |
    | 发送 QPS | 14813（虚高）    | 7807          | ——      |

### 🔮 后续优化方向 
     1. 💾 **MySQL 单条插入为当前瓶颈**，考虑引入批量 insert（攒批提交）进一步提升写入吞吐
     2. 📬 引入消息队列持久化（如 **Redis**），避免高并发下上游堆积丢包


#### 📁 目录结构


```
├── build                # muduo 第三方库
│  
├── include              # 头文件目录(.hpp)，所有类声明放这里
│   │ 
│   ├── Jsonn.hpp        # json序列化/反序列化工具
│   │ 
│   ├── Mutex.hpp        # 互斥锁保护资源
│   │ 
│   ├── Log.hpp          # 日志报错
│   │ 
│   ├── Mysql.hpp        # MySQL数据库操作封装
│   │ 
│   ├── Captcha.hpp      # 生成登陆时候的验证码
│   │ 
│   ├── PasswordHash.hpp # 注册的时候，使用盐和哈希加密
│  
├── log                  # 运行时日志文件存放目录
│  
├── srcv1.0              # 业务源码目录，主程序实现(.cc) ---- ✨✨✨V1.0单机架构的实现
│   ├── makefile         # 编译脚本
│   │ 
│   ├── main.cc          # HTTP Web 服务主程序
│   │ 
│   ├── MClient.cc       # muduo TCP客户端实现
│   │ 
│   └── MServer.cc       # muduo TCP服务端实现
│  
├── srcv2.0              # 业务源码目录，主程序实现(.cc) ---- ✨✨✨V2.0分布式架构的实现
│   ├── makefile                  # 编译脚本
│   │ 
│   ├── main.cc                   # HTTP Web 服务主程序
│   │ 
│   ├── broker.cc                 # borker 消息转发服务
│   │ 
│   ├── write_db_server.cc        # writeDB 写库服务
│   │ 
│   ├── MClient.cc                # muduo TCP客户端实现
│   │ 
│   └── MServer.cc                # muduo TCP服务端实现
│  
├── testing              # 单元/功能测试代码目录（用来测试，实际对项目没有帮助）
│  
├── third_party          # 第三方库
│   └── httplib.h        # 单头文件HTTP库，用于C++搭建web服务
│  
├── wwwroot              # Web静态资源目录（网页根目录）
│   ├── index.html       # 首页
│   │ 
│   ├── login.html       # 登录页面
│   │ 
│   └── register.html    # 注册页面
│  
│  
└── bench.py             # 性能测试的Python文件
│  
└── IoTSql.txt           # IoT数据库 SQL 脚本文件
│ 
└── README.md

```


#### 🛠 技术栈与第三方库安装说明
### 1️⃣ 基础环境要求

    1.  系统：Ubuntu22.04.1
    2.  编译器：g++≥11.4
    3.  C++标准：C++11
    4.  Mysql：8.0.46
    5.  jsoncpp：1.9.5-3

### 2️⃣ 基础技术栈要求

    1、语言：C++11
    2、数据库：MYSQL
    3、序列和反序列化：Json
    4、网络编程：httplib、muduo
    5、前端：html、css、js

### 3️⃣ 安装三方库

---

#### （1）更新源

```bash
sudo apt update
```

---

#### （2）安装 MySQL

```bash
# 安装 MySQL 服务端
sudo apt-get install mysql-server
```

安装完成后，设置 root 密码为 `123456`：

```bash
# 打开 MySQL（root 用户）
mysql -u root -p

# 修改 root 密码
ALTER USER 'root'@'localhost' IDENTIFIED WITH mysql_native_password BY '123456';
```

安装 MySQL 客户端开发库（编译时需要）：

```bash
sudo apt install libmysqlclient-dev
```

---

#### （3）安装 JSON 库

```bash
sudo apt-get install libjsoncpp-dev
```

---

### ⭐♥⭐ 配置 Nginx（V2 版本继续配置）

---

分布式架构中，即使一台服务器挂掉，另一台仍能继续服务，这依赖 **负载均衡 + 轮询** 机制。轮询即多个后端轮流工作，为完成负载均衡，需要配置 Nginx 来实现轮询转发。

---

#### 1️⃣ 安装 Nginx

```bash
sudo apt-get update
sudo apt-get install -y nginx
```

---

#### 2️⃣ 验证安装

```bash
nginx -v        # 打印版本号
```

---

#### 3️⃣ 确认 Nginx 带 Stream 模块

```bash
nginx -v 2>&1 | grep -o with-stream
```

- 输出 `with-stream` → **支持**，可直接配置 TCP 负载均衡
- 无输出 → 需要安装 `nginx-full`

---

#### 4️⃣ 编写 Nginx 配置

**① 打开主配置文件**

```bash
sudo vim /etc/nginx/nginx.conf
```

**② 在文件最外层（与 `http {}` 同级）加入以下内容**

```nginx
stream {
    upstream collectors {
        server 127.0.0.1:8889;
        server 127.0.0.1:8890;
    }
    server {
        listen 8888;
        proxy_pass collectors;
    }
}
```

**说明**

- `upstream collectors`：定义后端服务器列表（8889 / 8890 两个 ChatServer 网关）
- `listen 8888`：Nginx 对外监听的 TCP 端口
- `proxy_pass collectors`：将客户端连接轮询转发给后端

---

#### 5️⃣ 检查并重载配置

```bash
sudo nginx -t            # 检查配置，应打印 syntax is ok / test is successful
sudo nginx -s reload     # 重载配置
```

---

### 🚀 快速开始

---

#### 🥇 V1.0 单机架构运行部署步骤

---

 **1️⃣ 项目编译操作** 

**① 进入 V1.0 源码目录**

```bash
cd srcv1.0
```

**② 一键编译**

```bash
make
```

通过项目内置 Makefile 一键编译所有服务源码，自动生成可执行程序。

---

 **2️⃣ 多终端启动后台服务** 

V1.0 单机架构需开启 **三个独立终端**，分别启动 TCP 服务端、TCP 客户端、HTTP Web 服务，三者协同运行才可实现完整功能。

---

**终端1：启动 TCP 服务端（数据接收 + 数据库存储）**

```bash
./MServer
```

作用：开启 TCP 监听端口，负责接收传感器模拟设备上报的温度、湿度数据，完成 JSON 解析、MySQL 数据入库。

---

**终端2：启动 TCP 客户端（模拟传感器设备）**

```bash
./MClient
```

作用：模拟终端传感器设备，持续向服务端上报环境检测数据。

---

**终端3：启动 HTTP 后台主服务（Web 接口服务）**

```bash
./main
```

作用：开启 8080 端口 Web 服务，提供前端数据查询接口。

---

 **3️⃣ 前端页面访问与登录** 

确认三个服务全部启动成功后，即可通过浏览器访问系统前端页面。

**1、浏览器访问地址（服务器部署）**

```
服务器ip:8080/login
```

**2、输入系统默认登录账号密码**

- 账号：`admin`
- 密码：`123456`

**3、登录成功**

自动跳转至传感器数据可视化界面，可实时查看设备上报的温度、湿度监测数据。

---

 **4️⃣ 补充说明** 

若出现网页访问解析失败、无法打开页面的情况，优先检查三个终端服务是否全部正常运行、服务器 8080 端口是否放行，排查无误后重新访问即可。

---

#### 🥈 V2.0 分布式架构运行部署步骤

---

 **1️⃣ 项目编译操作** 

**① 进入 V2.0 源码目录**

```bash
cd srcv2.0
```

**② 一键编译**

```bash
make
```

通过项目内置 Makefile 一键编译所有服务源码，自动生成可执行程序。

---

 **2️⃣ 多终端启动后台服务** 

V2.0 分布式架构需开启 **七个独立终端**，分别启动 TCP 服务端、TCP 客户端、WriteDB 写库服务、HTTP Web 服务，各进程协同运行才可实现完整功能。

---

**终端1：启动 Broker（消息转发中间服务）**

```bash
./broker
```

作用：开启 TCP 监听端口 **7777**，接收 ChatServer 转发来的传感器数据，按轮询负载均衡分发给下游 WriteDB Server。

---

**终端2、终端3：启动两个 write_db_server（写库服务）**

```bash
./write_db_server     # 实例 1
./write_db_server     # 实例 2
```

作用：作为 Broker 的消费者（consumer），接收 Broker 转发的数据，解析 JSON 并写入 MySQL，支持双实例互为备份。

---

**终端4、终端5：启动两个 MServer（ChatServer 网关）**

```bash
./MServer 8889        # 网关实例 1
./MServer 8890        # 网关实例 2
```

作用：开启 TCP 监听端口 **8889 / 8890**，接收 MClient 上报的传感器数据，转发给 Broker，并广播给所有在线客户端。由 Nginx 在 **8888** 端口做轮询负载均衡。

---

**终端6：启动 TCP 客户端（模拟传感器设备）**

```bash
./MClient
```

作用：模拟终端传感器设备，持续向 ChatServer 上报温度、湿度等环境数据。

---

**终端7：启动 HTTP 后台主服务（Web 接口服务）**

```bash
./main
```

作用：开启 **8080** 端口 Web 服务，提供前端数据查询接口。

---

 **3️⃣ 前端页面访问与登录** 

确认所有服务全部启动成功后，即可通过浏览器访问系统前端页面。

**1、浏览器访问地址（服务器部署）**

```
服务器ip:8080/login
```

**2、输入系统默认登录账号密码**

- 账号：`admin`
- 密码：`123456`

**3、登录成功**

自动跳转至传感器数据可视化界面，可实时查看设备上报的温度、湿度监测数据。

---

 **4️⃣ 补充说明** 

若出现网页访问解析失败、无法打开页面的情况，优先检查所有终端服务是否全部正常运行、服务器 8080 端口是否放行，排查无误后重新访问即可。

---

### 🔌 端口速查

| 服务 | 端口 | 说明 |
|---|---|---|
| Nginx | 8888 | TCP 负载均衡入口 |
| MServer × 2 | 8889 / 8890 | ChatServer 网关 |
| Broker | 7777 | 消息转发 |
| MySQL | 3306 | 数据库 |
| Web (main) | 8080 | HTTP 服务 |

### 🧠 项目核心运行原理

 **1. 数据采集与传输** 
    基于muduo Reactor事件驱动模型搭建TCP长连接服务，持续监听传感器终端连接请求。设备端定时采集光照、湿度数据，封装设备编号、采集时间等信息，生成标准JSON数据包上报至服务端。

 **2. 数据解析与持久化** 
    服务端接收数据后，完成JSON反序列化解析有效数据，通过自定义阈值规则判断环境数据是否异常，支持后续告警拓展。同时通过线程队列异步将合规数据写入MySQL数据库，避免网络线程阻塞，提升并发处理能力。

 **3. 数据查询与可视化** 
    前端通过Ajax定时调用后端/api/sensor接口，发起HTTP数据查询请求；后端接收请求后查询数据库，组装JSON格式数据返回前端，前端解析数据并渲染折线图、数据表格，实现环境数据实时可视化展示。

### 💡 前端说明

    - HTML + CSS + JS 实现可视化页面
    - 网页定时 Ajax 请求调用 `/api/sensor` 接口
    - 后端返回 JSON
    - 前端解析，渲染折线图、表格，实现传感器实时数据可视化


### 📝 后续规划
    - 中间件优化：引入消息队列持久化Redis,避免高并发下上游推积丢包
    - 异步优化：内存队列 + 线程池实现数据库异步写入，避免网络线程阻塞
    - 日志模块：异步日志，分级日志输出、日志滚动