# RustIotPlat

**Repository Path**: kdisi_admin/RustIotPlat

## Basic Information

- **Project Name**: RustIotPlat
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: MIT
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-05-30
- **Last Updated**: 2026-07-08

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# HIoT Platform — 高性能物联网接入平台

基于 Rust 构建的高性能物联网接入平台，支持百万级设备并发接入、遥测数据处理、规则引擎和设备管理。

## 架构概览

```
设备 ──MQTT──▶ Device Hub ──Kafka──▶ Engine ──▶ TDengine/PostgreSQL
                (网关)        │       (遥测管线)
                              │
                              ▼
                        Rule Engine ──▶ 规则链执行
                        (规则引擎)
                              ▲
API ◀──────────────────────────┘
(REST/WebSocket)
```

### 核心服务

| 服务 | 端口 | 说明 |
|------|------|------|
| **API** | 8000 | REST API 服务（Rocket），设备管理、用户认证、规则链 CRUD |
| **Engine** | 8092 | 遥测处理管线，Kafka 消费 → TDengine 存储 → 规则转发 |
| **Device Hub** | 1883/8093 | MQTT 网关（rumqttd），设备认证、连接管理、协议适配 |
| **Rule Engine** | 8091 | 规则链执行引擎，水平扩展的 Kafka 消费者组 |

### 技术栈

- **运行时**: Tokio 异步运行时
- **HTTP**: Rocket 0.5
- **MQTT**: rumqttd（内嵌 MQTT broker）
- **消息队列**: Kafka / Redpanda
- **关系数据库**: PostgreSQL（SQLx）
- **时序数据库**: TDengine
- **缓存**: Redis
- **协议支持**: MQTT, Sparkplug B, CoAP, LwM2M（规划中）

## 部署模式

平台支持**单体**和**微服务**两种部署模式，共享同一套代码。

### 单体模式（Standalone）

所有服务运行在一个进程内，通过进程内消息总线（InProcessBus）通信，不需要 Kafka。

**适用场景**: 小规模接入、开发测试、边缘部署、PoC 验证

```bash
# 构建并运行
cargo run --bin standalone -- --config standalone/app-standalone.yml

# 或使用 Docker
docker build -f docker/Dockerfile --build-arg SERVICE=standalone -t hiot-standalone .
docker run -p 8000:8000 -p 1883:1883 hiot-standalone
```

**依赖**: PostgreSQL, Redis, TDengine（可选）

### 微服务模式（Distributed）

各服务独立部署，通过 Kafka 通信，支持水平扩展。

**适用场景**: 生产环境、大规模设备接入、高可用要求

```bash
# 分别启动各服务
cargo run --bin api -- --config api/app-local.yml
cargo run --bin engine -- --config api/app-local.yml
cargo run --bin device-hub -- --config api/app-local.yml
cargo run --bin rule-engine-service

# 或使用 Docker Compose
docker compose -f scripts/dev-stack.yml up -d
```

**依赖**: PostgreSQL, Redis, TDengine, Kafka

### 架构对比

```
┌─────────────────────────────────┐     ┌─────────┐ ┌─────────┐ ┌──────────┐ ┌────────────┐
│         Standalone              │     │   API   │ │ Engine  │ │Device Hub│ │ Rule Engine│
│                                 │     │(独立进程)│ │(独立)   │ │ (独立)   │ │  (独立)    │
│  API + Engine + DeviceHub       │     └────┬────┘ └────┬────┘ └────┬─────┘ └─────┬──────┘
│  + RuleEngine                   │          │           │           │              │
│  + InProcessBus                 │          └───────────┴───────────┴──────────────┘
│                                 │                          Kafka
│  共享: PG, Redis, TDengine      │
└─────────────────────────────────┘
```

## 快速开始

### 环境要求

- Rust 1.96+
- PostgreSQL 15+
- Redis 7+
- TDengine 3.x（可选，不配置时降级运行）
- Kafka 3.x（仅微服务模式需要）

### 1. 克隆项目

```bash
git clone git@gitee.com:kdisi_admin/RustIotPlat.git
cd RustIotPlat
```

### 2. 配置

```bash
# 复制示例配置
cp standalone/app-standalone.yml app-local.yml

# 编辑配置（数据库连接、Redis 地址等）
vim app-local.yml
```

### 3. 启动基础设施

```bash
# 使用 Docker Compose 启动依赖服务
docker compose -f scripts/dev-stack.yml up -d
```

### 4. 运行

```bash
# 单体模式（推荐开发测试）
cargo run --bin standalone

# 微服务模式
cargo run --bin api &
cargo run --bin engine &
cargo run --bin device-hub &
cargo run --bin rule-engine-service &
```

### 5. 验证

```bash
# API 健康检查
curl http://localhost:8000/health

# Swagger UI
open http://localhost:8000/swagger-ui/

# MQTT 测试（需要 mosquitto-clients）
mosquitto_pub -h localhost -p 1883 -t "v1/devices/me/telemetry" \
  -u "YOUR_DEVICE_TOKEN" -m '{"temperature": 25.5}'
```

## 项目结构

```
rustPlatform/
├── api/                    # REST API 服务
│   ├── src/
│   │   ├── biz/            # 业务逻辑层（95+ 模块）
│   │   ├── controller/     # HTTP 路由处理器（91+ 模块）
│   │   ├── routes/         # 路由注册
│   │   ├── state/          # Rocket 状态管理
│   │   └── lib.rs          # build_rocket() + 共享导出
│   └── migrations/         # PostgreSQL 数据库迁移
├── engine/                 # 遥测处理管线
│   └── src/
│       ├── kafka_consumer.rs   # Kafka 消费 + 消息处理
│       ├── tdengine_writer.rs  # TDengine 批量写入
│       ├── batch_worker.rs     # Postgres 任务队列 → MQTT
│       └── lib.rs              # run_engine() 入口
├── device-hub/             # MQTT 网关
│   └── src/
│       ├── auth.rs             # Redis 设备认证
│       ├── connection_map.rs   # 分布式连接映射
│       ├── kafka_producer.rs   # 上行消息生产
│       ├── kafka_consumer.rs   # 下行命令消费
│       └── lib.rs              # run_device_hub() 入口
├── rule-engine-service/    # 规则引擎服务
│   └── src/
│       ├── consumer.rs         # Kafka 消费 + 规则执行
│       ├── dlq.rs              # 死信队列处理
│       └── lib.rs              # run_rule_engine() 入口
├── common/                 # 共享库（重型依赖）
│   └── src/
│       ├── kafka_bus.rs        # KafkaBus 实现
│       ├── config.rs           # 配置加载
│       ├── pg_utils.rs         # PostgreSQL 工具
│       └── redis_handler.rs    # Redis 工具
├── common-core/            # 轻量共享库（无重型依赖）
│   └── src/
│       ├── message_bus.rs      # MessageBus trait + InProcessBus
│       └── config.rs           # 基础配置
├── standalone/             # 单体模式入口
│   ├── src/main.rs             # 4 服务并发启动
│   └── app-standalone.yml      # 单体模式配置
├── cli/                    # CLI 管理工具
├── vendor/rumqttd/         # MQTT broker（fork）
├── docker/Dockerfile       # 统一 Docker 构建
└── k8s/                    # Kubernetes 部署清单
```

## 消息总线抽象

平台通过 `MessageBus` trait 实现服务间通信的抽象，支持两种实现：

```rust
#[async_trait]
pub trait MessageBus: Send + Sync + 'static {
    async fn publish(&self, topic: &str, key: &str, payload: &[u8]) -> Result<()>;
    fn subscribe(&self, topic: &str) -> Box<dyn MessageConsumer>;
    async fn publish_dlq(&self, topic: &str, payload: &[u8]) -> Result<()>;
}
```

| 实现 | 使用场景 | 底层技术 |
|------|---------|---------|
| `InProcessBus` | 单体模式 | `tokio::sync::broadcast` |
| `KafkaBus` | 微服务模式 | `rdkafka` |

## 配置说明

### 环境变量覆盖

所有配置项均可通过环境变量覆盖：

| 环境变量 | 说明 | 默认值 |
|---------|------|--------|
| `APP_CONFIG` | 配置文件路径 | `app-local.yml` |
| `JWT_SECRET` | JWT 签名密钥 | - |
| `REDIS_HOST` | Redis 地址 | `127.0.0.1` |
| `REDIS_PORT` | Redis 端口 | `6379` |
| `PG_HOST` | PostgreSQL 地址 | `127.0.0.1` |
| `PG_PORT` | PostgreSQL 端口 | `5432` |
| `PG_DBNAME` | 数据库名 | `hiot` |
| `PG_USERNAME` | 数据库用户 | `postgres` |
| `PG_PASSWORD` | 数据库密码 | - |
| `TDENGINE_DSN` | TDengine 连接串 | `taos://root:taos@127.0.0.1:6030` |
| `KAFKA_BROKERS` | Kafka 地址 | `localhost:9092` |
| `NODE_HOST` | 节点监听地址 | `0.0.0.0` |
| `NODE_PORT` | API 端口 | `8000` |

### Kafka Topic

| Topic | 方向 | 说明 |
|-------|------|------|
| `gateway.publish` | Device Hub → Engine | 上行遥测数据 |
| `gateway.subscribe` | Engine → Device Hub | 下行命令 |
| `gateway.subscribe.rule` | Engine → Rule Engine | 规则引擎输入 |
| `cmd.dead_letter` | Engine | 死信队列 |

## 性能优化

- **TDengine 批量写入**: 攒批写入，减少 TDengine 连接开销
- **Kafka 分区路由**: FNV-1a hash 确保同设备消息保序
- **Warning 工作池**: 固定大小 worker pool 替代 per-message spawn
- **独立运行时**: batch_worker 使用独立 Tokio 运行时，与遥测管线隔离
- **轻量表达式引擎**: 规则表达式使用 evalexpr/Rhai，减少脚本运行时开销

## 开发

```bash
# 检查编译
cargo check

# 运行测试
cargo test --workspace

# 代码格式化
cargo fmt

# 代码检查
cargo clippy --workspace

# 开发命令（just）
just dev          # 启动开发环境
just test         # 运行测试
just build        # 构建发布版本
```

## 相关文档

- [规则引擎分析](docs/rule-engine-analysis.md)
- [性能调优指南](docs/performance-tuning-guide.md)
- [节点能力矩阵](docs/node-capability-matrix.md)
- [百万设备优化](docs/rule-engine-million-device-optimization.md)

## 许可证

内部项目，未公开授权。