1. 项目概述一个面向未来的集成式物联网平台最近在整理过去几年做过的项目发现一个挺有意思的东西是之前为一个大型智慧园区项目做的物联网平台后端。当时的需求特别典型甲方既需要一个能管理成千上万个传感器比如温湿度、能耗、门禁的物联网平台又希望这个平台能和他们已有的楼宇自控系统、消防系统、甚至OA办公系统打通数据形成一个统一的“智慧大脑”。市面上要么是纯IoT平台只管设备接入和数据采集要么是传统的IBMS智能楼宇管理系统集成能力强但设备接入和扩展性往往是个短板。于是我们决定自己基于Python Django框架从头撸一个兼具两者特性的平台。现在这个项目的核心部分已经稳定运行了两年多我觉得是时候把它开源出来给有类似需求的同行们一个参考的轮子。简单来说这个项目是一个基于Django的、开源的、一体化的物联网与智能集成平台。它最核心的价值在于用一个技术栈解决了“物联”与“集成”两大难题。对于开发者而言你不需要在IoT平台和业务集成平台之间做艰难的选型或痛苦的缝合它提供了一套完整的解决方案从设备协议的解析、海量时序数据的存储与查询到多源异构系统如Modbus、BACnet、OPC UA、HTTP API等的数据接入与融合再到基于这些数据的可视化、告警、自动化工作流全部囊括其中。它非常适合用于智慧园区、智慧楼宇、智慧工厂、智慧农业等需要对物理设备进行集中监控并将设备数据与上层业务系统深度结合的复杂场景。2. 核心架构设计与技术选型思路2.1 为什么选择Django作为核心框架很多人在听到物联网平台时第一反应可能是Go、Rust或者Node.js认为它们在高并发、实时性方面有优势。我们选择Django是经过深思熟虑的主要基于以下几点考量开发效率与可维护性物联网和集成项目业务逻辑的复杂度和变化频率往往远高于单纯的设备接入。Django的“开箱即用”特性Admin后台、ORM、用户认证、表单处理和“约定优于配置”的理念能让我们快速搭建起复杂的数据模型和业务管理界面。对于一个需要频繁应对需求变更的集成类项目快速迭代和清晰的代码结构至关重要。生态成熟与稳定性Django拥有极其成熟和稳定的生态。对于物联网平台必须的组件如任务队列Celery、缓存Redis、数据库连接池等都有经过大量生产环境验证的集成方案。在涉及与多种第三方系统如数据库、消息队列、企业微信/钉钉API对接时能找到成熟库的概率非常大减少了造轮子的风险和成本。ORM的威力Django ORM不仅简化了数据库操作其强大的QuerySetAPI在构建复杂的数据查询、过滤和聚合报表时优势明显。物联网平台需要处理大量的关联查询如查询某个区域下所有设备的最近状态Django ORM能写出既高效又易读的代码。虽然有人诟病其性能但通过合理的索引、select_related/prefetch_related以及后续会讲到的读写分离、分库分表策略完全可以满足绝大多数场景。异步支持的成熟随着Django 3.1对ASGI的原生支持和Django 4.x的持续优化利用async视图和Django Channels来处理WebSocket协议实现设备的实时指令下发和状态推送已经变得非常顺畅。这弥补了传统Django在实时通信方面的短板。注意选择Django并不意味着忽视性能。物联网平台的核心性能瓶颈往往在I/O数据库、网络而非Web框架本身。正确的架构设计如引入消息队列解耦、使用高性能时序数据库、对耗时操作异步化比纠结框架的基准测试分数更有意义。2.2 整体架构分层解析我们的平台采用了经典的分层架构但每一层都针对物联网和集成场景做了特殊设计。[ 设备层 ] - 多种协议设备 (MQTT, CoAP, HTTP, TCP自定义...) | v [ 接入层 ] - 协议适配器 (Django Channels for WebSocket/MQTT, 自定义TCP服务) | | | v | [ 消息中间件 ] (Redis Streams / Apache Kafka) - 用于解耦与缓冲 | v [ 核心服务层 ] | |-- [ 设备管理服务 ] (Django App) - 设备建模、生命周期、凭证管理 |-- [ 数据服务 ] (Django App 时序数据库) - 遥测数据存储、查询、聚合 |-- [ 规则引擎服务 ] (Celery Django) - 告警规则、自动化场景 |-- [ 集成网关服务 ] (Django App) - 对接第三方系统API/协议 | v [ 数据存储层 ] |-- 关系数据库 (PostgreSQL) - 存储设备元数据、用户、配置等 |-- 时序数据库 (InfluxDB / TimescaleDB) - 存储海量设备时序数据 |-- 缓存 (Redis) - 会话、设备在线状态、频繁访问的元数据 | v [ 应用层 ] - Django Admin 扩展 / 前后端分离的Vue/React前端 / RESTful API GraphQL各层设计要点接入层没有采用单一的“轮询”或“服务器推送”模式而是支持多协议接入。对于低功耗设备使用UDP或CoAP对于需要双向通信的实时设备使用MQTT或WebSocket对于简单的数据上报使用HTTP POST。我们为每种协议编写了轻量级的“协议适配器”将不同格式的数据统一转换为平台内部的标准数据格式JSON Schema并抛入消息中间件。核心服务层这是业务逻辑的核心。我们严格遵循Django的App组织方式每个服务都是一个独立的Django App通过内部RPC如直接函数调用或轻量级消息或共享数据库需谨慎进行通信。这种“高内聚、低耦合”的设计使得未来可以相对容易地将某个服务拆分为独立的微服务。数据存储层这是性能的关键。绝对不要用关系型数据库存储时序数据。我们选择了TimescaleDB基于PostgreSQL的时序数据库扩展因为它既能利用PostgreSQL强大的关系型特性联合查询、事务又提供了针对时序数据优化的超表、连续聚合、数据压缩等功能。对于超大规模场景InfluxDB是另一个备选。Redis则用于缓存设备实时状态和热点配置极大减轻数据库压力。3. 核心功能模块深度实现3.1 设备建模与动态物模型物联网设备千差万别如何用一种灵活的方式描述它们我们借鉴了主流云平台的经验实现了动态物模型。核心思想将设备抽象为“产品”每个“产品”定义一套“物模型”属性、服务、事件。具体的设备是该“产品”的一个实例。Django模型设计# models.py in devices app class Product(models.Model): 产品模板 name models.CharField(max_length100) manufacturer models.CharField(max_length100) # 物模型定义以JSON Schema格式存储 thing_model_schema models.JSONField(defaultdict) class Device(models.Model): 设备实例 product models.ForeignKey(Product, on_deletemodels.PROTECT) name models.CharField(max_length100) device_id models.CharField(max_length64, uniqueTrue) # 唯一设备标识 secret models.CharField(max_length128) # 设备密钥用于鉴权 online_status models.BooleanField(defaultFalse) last_seen models.DateTimeField(nullTrue, blankTrue) # 动态属性存储设备上报的当前值 reported_properties models.JSONField(defaultdict) # 期望属性存储平台期望设备达到的状态用于下发控制指令 desired_properties models.JSONField(defaultdict)thing_model_schema示例:{ properties: { temperature: { identifier: temperature, name: 温度, dataType: float, unit: °C, accessMode: readOnly, specs: {min: -40, max: 125} }, powerSwitch: { identifier: powerSwitch, name: 电源开关, dataType: bool, accessMode: readWrite } }, events: { overheat: { name: 过热报警, params: [{identifier: currentTemp, dataType: float}] } } }实操要点动态字段验证当设备上报数据或平台下发指令时需要根据thing_model_schema动态验证数据格式。我们利用Django的receiver信号和自定义的clean()方法来实现。属性同步reported_properties设备上报和desired_properties平台期望的差异驱动了状态同步。平台向设备发送控制指令实质上是更新desired_properties然后通过下行通道MQTT/WebSocket将差异部分推送给设备。设备执行后上报新的reported_properties平台再更新并消除差异。元数据管理利用Django Admin我们可以非常方便地让运维人员通过网页界面创建新产品、定义物模型而无需开发人员修改代码和数据库。3.2 海量时序数据的高效存储与查询这是物联网平台的性能命脉。我们的方案是Django ORM TimescaleDB。1. 数据表设计 不在主业务库中存储遥测数据。我们为时序数据创建了独立的Django数据库路由和模型。# models.py in telemetry app class TelemetryData(models.Model): 时序数据表使用TimescaleDB超表 device models.ForeignKey(devices.Device, on_deletemodels.CASCADE, db_indexTrue) metric models.CharField(max_length64, db_indexTrue) # 指标名如temperature value models.FloatField() # 实际值可根据需要定义多种类型字段 timestamp models.DateTimeField(db_indexTrue) # 数据点时间戳 class Meta: db_table telemetry_data # 在数据库中将此表创建为TimescaleDB超表 # 通常通过迁移文件中的RunSQL操作执行 # SELECT create_hypertable(telemetry_data, timestamp); indexes [ models.Index(fields[device, metric, timestamp]), ]2. 数据写入优化 设备数据通过消息队列异步到达后批量写入是必须的。我们使用django.db.transaction.atomic和bulk_create。from django.db import transaction from .models import TelemetryData def batch_save_telemetry_data(data_points): 批量保存时序数据 objs [TelemetryData(**point) for point in data_points] with transaction.atomic(usingtimescale): # 指定时序数据库 TelemetryData.objects.using(timescale).bulk_create(objs, batch_size1000)3. 数据查询优化 利用TimescaleDB的时序查询函数和Django的Extra或RawSQL后期可封装为自定义查询集。# 查询某个设备过去24小时每5分钟的平均温度 from django.db import connection def get_device_avg_temperature(device_id, hours24, interval5 minutes): with connection.cursor() as cursor: cursor.execute( SELECT time_bucket(%s, timestamp) as bucket, avg(value) as avg_temp FROM telemetry_data WHERE device_id %s AND metric temperature AND timestamp now() - interval %s GROUP BY bucket ORDER BY bucket; , [interval, device_id, f{hours} hours]) rows cursor.fetchall() return rows # 更Django化的方式使用django-timescaledb等第三方库如果可用或自定义Manager/QuerySet。4. 数据保留与压缩 在TimescaleDB中设置数据保留策略和压缩策略自动清理旧数据并节省存储空间。这通常在数据库层面通过调度任务完成。-- 设置保留策略删除30天前的数据 SELECT add_retention_policy(telemetry_data, INTERVAL 30 days); -- 启用压缩 ALTER TABLE telemetry_data SET (timescaledb.compress, timescaledb.compress_orderby timestamp DESC); SELECT add_compression_policy(telemetry_data, INTERVAL 7 days);实操心得时序数据库的索引策略至关重要。我们的复合索引(device, metric, timestamp)能高效覆盖最常见的查询模式“查询某个设备的某个指标在一段时间内的数据”。避免在value字段上建索引除非有特殊的范围查询需求。3.3 规则引擎与自动化场景规则引擎是让物联网数据产生价值的关键。我们实现了一个基于Celery的、可灵活配置的规则引擎。核心组件触发器数据点到达、设备状态变化、定时任务、API调用等。条件对触发数据的判断逻辑大于、小于、等于、包含等。动作满足条件后执行的操作如发送告警邮件、短信、钉钉、调用设备服务、更新数据库、触发另一个工作流等。Django模型设计# models.py in rules app class Rule(models.Model): name models.CharField(max_length200) is_active models.BooleanField(defaultTrue) trigger_type models.CharField(max_length50, choicesTRIGGER_CHOICES) # e.g., telemetry, status_change, cron trigger_config models.JSONField() # 存储触发器的具体参数 condition models.JSONField(nullTrue, blankTrue) # 条件逻辑的JSON描述 actions models.JSONField() # 动作列表的JSON描述 def evaluate_and_execute(self, context_data): 评估条件并执行动作 if self._check_condition(context_data): self._execute_actions(context_data) def _check_condition(self, data): # 解析condition JSON并基于data进行评估 # 可以使用简单的表达式求值库如 asteval pass def _execute_actions(self, data): # 解析actions JSON异步执行各个动作 # 例如send_dingtalk_alert.delay(alert_config, data) pass工作流程当设备数据通过消息队列被消费后除了存入时序数据库还会发布一个“数据到达”事件。一个专门的Celery Worker监听此事件查询所有trigger_type为telemetry的活跃规则。对于每条规则用新数据作为上下文调用rule.evaluate_and_execute(context_data)。如果条件满足该规则定义的动作会被封装为Celery任务放入队列异步执行。示例温度过高告警规则触发器trigger_type: telemetry,trigger_config: {metric: temperature, device_filter: {...}}条件{operator: gt, value: 35}温度 35°C动作[{type: dingtalk, webhook: ..., template: 设备{device}温度过高{value}°C}]注意事项规则引擎的逻辑一定要异步化Celery并且做好错误处理和重试机制。避免在数据处理的同步路径中执行耗时的告警或集成动作否则会阻塞数据入库影响整体吞吐量。3.4 多源系统集成网关这是体现其“IBMS集成平台”特性的核心。我们设计了一个可插拔的适配器架构用于对接各类第三方系统。架构设计集成模型定义一个Integration基类记录集成的目标系统类型、配置信息如API地址、密钥、同步状态等。适配器模式为每种类型的系统如SQL数据库、HTTP REST API、Modbus TCP、BACnet/IP编写一个适配器类。所有适配器继承自一个公共的BaseAdapter抽象类实现connect(),fetch_data(),send_command()等方法。任务调度使用Celery Beat来调度周期性数据同步任务。每个Integration实例对应一个或多个Celery周期性任务。代码示例# integrations/adapters/base.py class BaseAdapter(ABC): def __init__(self, config): self.config config abstractmethod async def connect(self): pass abstractmethod async def fetch_data(self, point_mapping): 根据点位映射获取数据 pass abstractmethod async def send_command(self, point_id, value): 向指定点位发送控制命令 pass # integrations/adapters/modbus_adapter.py class ModbusTCPAdapter(BaseAdapter): def __init__(self, config): super().__init__(config) self.client None async def connect(self): from pymodbus.client import AsyncModbusTcpClient self.client AsyncModbusTcpClient(self.config[host], portself.config.get(port, 502)) await self.client.connect() async def fetch_data(self, point_mapping): # point_mapping: [{address: 40001, type: holding_register, scale: 0.1, ...}] results {} for point in point_mapping: address point[address] if point[type] holding_register: resp await self.client.read_holding_registers(address, count1) if not resp.isError(): raw_value resp.registers[0] results[address] raw_value * point.get(scale, 1) return results点位映射配置 在数据库中我们会存储“虚拟点位”与“真实点位”的映射关系。例如平台内有一个虚拟传感器“一楼大厅空调温度”其数据源可能映射到Modbus设备的保持寄存器40001并且需要乘以系数0.1。集成网关的任务就是定期如每5分钟通过Modbus适配器读取40001的值处理后更新到平台内这个虚拟传感器的reported_properties中并生成一条时序数据。这样上层应用完全无需关心数据来自哪里它们只和平台内统一的设备模型交互。4. 高并发与性能优化实战当设备量达到万级甚至十万级时一些在开发阶段不明显的问题就会暴露。以下是我们在实际部署中踩过的坑和优化方案。4.1 连接管理与心跳保活问题大量设备通过长连接如MQTT、WebSocket接入单纯的Django Channels应用可能会成为连接瓶颈并且连接状态管理复杂。解决方案使用专业的MQTT Broker如EMQX或Mosquitto集群而不是用Channels自带的MQTT实现来处理海量MQTT连接。Django后端仅作为MQTT Broker的一个客户端订阅特定的主题如device//data来接收数据。这样将连接管理与业务逻辑彻底解耦。设备状态同步设备在线状态online_status不再仅仅依赖于应用层的心跳。我们利用MQTT Broker的$SYS主题或其API如EMQX的HTTP API来获取客户端的连接/断开事件通过一个轻量级服务同步到平台的Device表中。同时设备端也需要发送周期性的心跳数据包作为网络不佳时的补充判断。连接数限制与负载均衡在Django Channels层对于WebSocket连接我们使用channel_layer配置为Redis进行横向扩展并通过Nginx进行负载均衡支持多个Django实例共同处理WebSocket连接。4.2 数据库读写分离与缓存策略问题平台Admin操作、API查询、设备数据写入全部集中在主库导致数据库压力大响应变慢。解决方案Django数据库路由配置多数据库将时序数据TelemetryData指向TimescaleDB将设备元数据、用户数据等指向主PostgreSQL。对于复杂的关联查询如需要同时查设备信息和其最新数据可以通过只读从库或缓存来缓解。读写分离配置一个PostgreSQL从库在Django的DATABASES设置中定义read和write配置。通过自定义数据库路由或使用django-db-read-replica这类中间件将大部分的读请求如API列表查询、报表生成导向从库。多级缓存视图缓存对不经常变动的配置页面、产品目录使用Django的缓存框架。查询缓存使用django-cacheops等库自动缓存复杂的ORM查询结果。对象缓存设备的最新状态reported_properties在写入数据库的同时也写入Redis键为device:status:{device_id}。API查询设备实时状态时优先从Redis读取毫秒级响应。Pub/Sub利用Redis的发布订阅功能实现设备状态变化的实时推送。当后端服务更新了设备状态缓存同时发布一个消息到device.status.updated频道WebSocket服务监听到后即刻推送给前端订阅了该设备的用户。4.3 异步任务队列的合理使用问题所有操作都同步执行导致API响应慢且一个耗时任务失败会影响整体。解决方案Celery Redis/RabbitMQ是Django项目的黄金搭档。必须异步化的操作设备历史数据查询与报表生成尤其是涉及大量聚合计算。向第三方系统同步数据HTTP API调用可能很慢或不稳定。发送邮件、短信、钉钉等告警通知。批量设备指令下发。规则引擎的条件判断与动作执行如前所述。任务设计要点任务幂等性确保任务被重复执行不会产生副作用。例如基于设备ID和指令ID生成唯一键在Redis中做setnx检查。任务结果跟踪对于重要的任务如控制指令使用Celery的result_backend存储结果并提供API供前端查询任务状态。错误重试与告警为任务设置合理的autoretry_for和max_retries并配置celery的失败回调将最终失败的任务信息记录到数据库或发送管理员告警。4.4 前端数据可视化与实时更新对于物联网平台一个能动态反映设备状态的可视化界面至关重要。我们采用前后端分离架构后端提供RESTful API和WebSocket前端使用Vue.js或React。实时数据推送方案WebSocket长连接每个登录的用户前端与Django Channels建立的WebSocket连接。当用户进入某个设备监控页面时前端订阅该设备的主题如device.123456.status。后端发布更新当设备数据更新或状态变化时后端服务可能是处理MQTT消息的Worker也可能是规则引擎会向Redis Channel发布消息。Channels消费并推送Django Channels的Consumer监听着Redis Channel收到消息后通过WebSocket连接推送给所有订阅了该主题的前端客户端。# consumers.py class DeviceStatusConsumer(AsyncWebsocketConsumer): async def connect(self): self.device_id self.scope[url_route][kwargs][device_id] self.room_group_name fdevice_{self.device_id} # 加入设备组 await self.channel_layer.group_add( self.room_group_name, self.channel_name ) await self.accept() async def disconnect(self, close_code): # 离开设备组 await self.channel_layer.group_discard( self.room_group_name, self.channel_name ) # 接收来自其他Channel如Redis的消息并转发给WebSocket客户端 async def device_status_update(self, event): message event[message] # 发送消息到WebSocket await self.send(text_datajson.dumps(message))这种模式实现了真正的实时更新用户体验远优于前端定时轮询API。5. 部署、监控与运维实践5.1 容器化部署我们使用Docker Compose进行一站式部署编排所有服务。# docker-compose.prod.yml version: 3.8 services: postgres-master: image: postgres:15 volumes: - pg_data:/var/lib/postgresql/data environment: - POSTGRES_DBmydb - POSTGRES_USERmyuser - POSTGRES_PASSWORDmypassword timescaledb: image: timescale/timescaledb:latest-pg15 volumes: - ts_data:/var/lib/postgresql/data environment: - POSTGRES_DBtsdb - POSTGRES_USERtsuser - POSTGRES_PASSWORDtspassword redis: image: redis:7-alpine command: redis-server --appendonly yes volumes: - redis_data:/data emqx: image: emqx:5 ports: - 1883:1883 # MQTT - 8083:8083 # MQTT over WebSocket - 18083:18083 # Dashboard volumes: - emqx_data:/opt/emqx/data django-web: build: . command: gunicorn my_iot_platform.wsgi:application --bind 0.0.0.0:8000 --workers 4 depends_on: - postgres-master - timescaledb - redis - emqx environment: - DJANGO_SETTINGS_MODULEmy_iot_platform.settings.production volumes: - static_volume:/app/staticfiles - media_volume:/app/media django-celery-worker: build: . command: celery -A my_iot_platform worker --loglevelinfo --concurrency4 depends_on: - redis - django-web environment: - DJANGO_SETTINGS_MODULEmy_iot_platform.settings.production django-celery-beat: build: . command: celery -A my_iot_platform beat --loglevelinfo depends_on: - redis - django-web environment: - DJANGO_SETTINGS_MODULEmy_iot_platform.settings.production nginx: image: nginx:alpine ports: - 80:80 - 443:443 volumes: - ./nginx/conf.d:/etc/nginx/conf.d - static_volume:/app/staticfiles depends_on: - django-web volumes: pg_data: ts_data: redis_data: emqx_data: static_volume: media_volume:5.2 监控与日志没有监控的系统就是在裸奔。我们主要监控以下几个层面基础设施监控使用Prometheus Grafana。监控服务器CPU、内存、磁盘、网络。监控PostgreSQL/TimescaleDB的连接数、查询性能、缓存命中率。监控Redis的内存使用、命中率、连接数。应用性能监控使用Django的django-prometheus库暴露指标如请求延迟、错误率、Celery队列长度、任务执行时间等。在Grafana中绘制仪表盘。业务监控这是最重要的。我们自定义了关键业务指标设备在线率总在线设备/总注册设备。数据上报成功率成功入库的数据点/接收到的数据点。规则触发频率与动作执行成功率。第三方集成同步延迟与失败率。将这些指标通过Django的logging模块记录并配置django-statsd客户端发送到StatsD最终由Prometheus收集。日志集中化所有服务的日志都通过Docker的json-file驱动输出然后由Fluentd或Loki收集统一在Grafana中查看。为Django配置结构化的JSON日志方便过滤和查询。5.3 安全考量物联网平台涉及物理设备控制安全至关重要。设备认证每个设备必须有唯一的device_id和secret或证书。MQTT连接使用username/password或TLS客户端证书认证。HTTP API使用Token签名认证如HMAC-SHA256。传输加密MQTT使用TLS/SSLMQTTS。WebSocket使用WSS。HTTP API强制使用HTTPS。权限控制使用Django REST Framework的权限类和Django Guardian进行对象级权限控制。确保用户只能访问其被授权的设备、数据。API限流使用django-ratelimit对API接口进行限流防止恶意刷接口或设备异常高频上报。固件更新安全如果支持OTA固件升级必须对固件包进行签名验证确保来源可信。6. 开源项目的规划与社区建设将这个项目开源不仅仅是扔代码到GitHub。我们希望能构建一个可持续的社区。清晰的文档包括快速上手的README.md、详细的安装部署文档、架构说明、API文档以及贡献指南。示例与演示提供docker-compose一键启动的演示环境包含模拟设备和数据看板让用户能在几分钟内看到效果。模块化设计确保核心的devices,telemetry,rules,integrations等App高度解耦。用户可以根据需要只使用设备接入功能或者只使用集成网关功能。插件化机制计划设计一套标准的“插件”接口让社区能够轻松贡献新的设备协议适配器如OPC UA、蓝牙、新的通知渠道如飞书、企业微信、新的数据可视化组件等。积极的issue与PR处理建立行为准则友善地欢迎和指导贡献者。开源这条路很长但看到自己构建的系统能够帮助其他人解决实际问题甚至激发出更好的创意和实现这种成就感是闭门造车无法比拟的。这个项目是我们团队在多个实际项目中经验和思考的结晶它可能不是最完美的但一定是一个扎实的、可用的起点。期待在开源社区与大家相遇共同完善它。