简介这是一份关于北京市智能交通系统设计的PDF文档源自2003年《公路交通科技》期刊论文适合交通工程、智慧城市及智能交通领域的研究者与从业者参考。文档围绕交通流数据采集、处理/分析和信息发布三大模块展开介绍了北京当时环形线圈、微波与视频检测三种采集方式以及可变信息显示屏、交通广播、公交网和停车诱导系统等发布渠道并探讨了与信号控制、号牌识别等系统的集成思路。资源为单文件PDF大小仅227KB便于快速阅读与检索目前已有70人在线学习。读者可从中获得北京早期交通流信息系统的原始设计方案、关键数据流框架及实践细节对理解智能交通系统演进或撰写相关方案具有参考价值。1. 为什么交通流系统要先定数据边界再谈技术选型城市交通流数据采集与处理系统最容易被低估的是数据边界问题。北京这样的超大城市单日来自微波、地磁、视频检测器的数据量轻松过亿加上浮动车GPS轨迹数据形态从结构化记录到半结构化报文都有。很多团队一上来就铺Kafka、Spark、时序数据库结果半年后卡在数据对不齐、时间不同步、重复上报和传感器漂移上。这类系统设计分享实际是在讲一套从采集到发布的完整数据链路先定业务指标再倒推采集频率、处理窗口和发布格式。核心原则是每一步都明确数据边界——什么数据必须实时什么数据允许延迟什么数据宁可丢弃也不污染指标。这套方案面向一线系统设计、数据开发和运维人员偏重可落地的实现与排错思路不讨论纯学术算法。2. 采集层设计从断面检测器到浮动车数据的接入规范2.1 数据源分类与采集频率的取舍交通流数据源大致分两类定点断面数据线圈、微波、雷达、视频和移动源数据出租车GPS、网约车轨迹、公交到站数据。定点设备给出精准的断面流量、速度、占有率但覆盖有限浮动车覆盖面广却存在采样率不匀、漂移和城市峡谷信号丢失问题。设计采集规范时我一般先列一个数据源属性表明确每个源的采集频率、上报协议、时间同步方式和质量等级。数据源典型采集频率上报方式主要字段质量风险微波/雷达检测器20秒-1分钟TCP长连接断面流量、平均速度、时间占有率时钟漂移、断点补报地磁线圈30秒HTTP POST车流量、道路占用状态线圈损坏产生零值视频AI检测1-5秒MQTT车牌、车型、坐标光照影响识别率出租车GPS10-30秒Kafka/HTTP经纬度、瞬时速度、载客状态轨迹漂移、重复上报这里的关键是不要对所有数据源用同一套采集线程。定点数据追求确定性用固定周期拉取浮动车数据到达时间分布长尾用队列蓄洪并允许少量乱序。合理的策略是设置两级缓存接入网关先做格式校验和时间戳标准化再进入消息队列同时为每个数据源维护一个设备状态表记录最近上报时间、计数和异常率方便后续的数据质量评估。2.2 用Kafka承接高并发流量的最小配置常见做法是Kafka作为采集总线生产者是各数据采集代理消费者是后续的处理管道。以北京市典型规模估算每秒峰值可能达到数万条记录单分区顺序写就能应付但为了后续并行处理需要按设备ID或地理位置分区。下面是一段Python生产者的最小实现用于从检测器读取数据并写入Kafka。from kafka import KafkaProducer import json, time, random producer KafkaProducer( bootstrap_servers[10.10.0.11:9092, 10.10.0.12:9092], key_serializerlambda k: str(k).encode(), value_serializerlambda v: json.dumps(v).encode(), acksall, # 等所有副本确认避免设备数据丢失 retries3, # 瞬时错误重试3次 linger_ms20, # 攒20ms再发提升吞吐 batch_size32768 # 单批32KB ) while True: packet { device_id: fD035{random.randint(1, 999):03d}, ts: int(time.time() * 1000), volume: random.randint(0, 120), speed: random.randint(10, 80), occupancy: round(random.uniform(0.1, 0.9), 2) } producer.send(traffic-packet, keypacket[device_id], valuepacket) time.sleep(0.02) # 模拟20ms一条acksall和retries3是交通流数据的标配因为感应线圈数据一旦丢了几秒就没法补回来。linger_ms不要设太大控制在 50ms 以下否则会明显增加端到端延迟。batch_size根据单条报文大小调一般 16KB 到 64KB 就够了。注意acksall只保证Kafka副本不丢不等于端到端不丢。生产者进程崩溃前的最后一批数据仍然可能丢失所以消费者侧要配合幂等写或唯一键去重。消费者侧要注意提交偏移量的时机。如果先处理再提交崩溃时会重复处理如果先提交再处理会丢数据。我的做法是业务幂等处理后手动提交在 Java 代码里把enable_auto_commit设为 false在下一批拉取之前提交上一批偏移量。2.3 采集层的容错与重试机制采集网关宕机是常态不能指望硬件不掉链子。要设计独立的死信队列把格式错误、解码失败或时间戳超过当前时间两小时的数据单独存放而不是直接丢弃。每隔一段时间回放死信队列分析哪些设备在持续产生脏数据然后主动把该设备踢出采集链路并通知运维。另外需要统一时钟。工业设备的时钟经常不准我见过一台微波检测器比标准时间快 12 秒导致上下游数据时间错位。最稳妥的解决办法是网关收到数据后不信任设备时间而是以网关接收时间为准并把设备时间一并保存用于诊断。这样即使设备时间错乱后续的计算窗口也不会被污染。3. 处理与分析清洗、聚合与路况计算3.1 流式数据清洗规则与异常值过滤原始数据不能直接进入指标计算清洗是处理层的第一道关。常见问题包括速度值超过路段限速两倍、流量为负、坐标点在河道或公园、同一设备ID在极短时间内出现荒谬的跳变。清洗规则要用配置化方式管理而不是硬编码在程序里。下面是用 Python 和 Pandas 做离线索引数据清洗的一个示例实际流式处理可以改为 Spark Structured Streaming 的 foreachBatch。import pandas as pd def clean_traffic(df): # 剔除缺失关键字段的样本 df df.dropna(subset[device_id, ts, speed]) # 1. 速度范围检查高速公路0-160普通道路0-120超过视为异常置为NA df.loc[(df[road_level] highway) (df[speed] 160), speed] pd.NA df.loc[(df[road_level] urban) (df[speed] 120), speed] pd.NA # 2. 流量与占有率合理性流量为负或占有率大于1直接剔除 df df[(df[volume] 0) (df[occupancy].between(0, 1))] # 3. 时间戳校准距离网关时间超过600秒的数据丢弃 df df[abs(df[ts] - df[gateway_ts]) 600_000] # 4. 去重同一设备、同一时间窗口内保留最后一条 df df.drop_duplicates(subset[device_id, ts], keeplast) return df这个清洗函数里有四个关键参数速度阈值、时间偏差容忍度、去重键和保留策略。速度阈值要按道路等级分别配置因为北京快速路和胡同能跑出的正常速度范围完全不同。时间偏差容忍度设为 10 分钟是经验值太长会让上游断点数据反复进入处理管道太短又会丢弃一些真的因为信号问题延迟上报的有效数据。3.2 路况计算模型从5分钟粒度到拥堵指数路况计算通常以 15 分钟或 5 分钟为窗口将清洗后的点数据聚合成路段平均速度和饱和度。一个常用公式是时间占有率occ和平均速度v结合判断拥堵等级。用流量加权平均速度能避免低速车少时拉低均值。这里给出一个 Spark Structured Streaming 中基于窗口聚类的核心代码。from pyspark.sql import functions as F streaming_df spark.readStream.format(kafka) \ .option(kafka.bootstrap.servers, 10.10.0.11:9092) \ .option(subscribe, traffic-cleaned) \ .load() parsed streaming_df.selectExpr(CAST(value AS STRING) as json) \ .select(F.from_json(json, schema).alias(data)) \ .select(data.*) # 按路段ID和5分钟窗口聚合 agg parsed.withWatermark(ts, 60 seconds) \ .groupBy(segment_id, F.window(ts, 5 minutes)) \ .agg( F.avg(speed).alias(avg_speed), F.avg(occupancy).alias(avg_occupancy), F.count(*).alias(sample_cnt) )withWatermark设置 60 秒是为了容忍乱序数据超过 60 秒晚到的数据直接忽略。window是滚动窗口北京路况发布常用 5 分钟粒度如果要做早晚高峰更细的发布可以改成 3 分钟但会放大单点噪声。从聚合结果计算道路拥堵指数时我习惯于把平均速度映射到 0-10 的指数映射公式按道路等级设定。比如快速路指数10*(1 - v/80) 并限制在 0.5-10 之间普通道路则用期望速度v0作为分母。这个指数要保留两位小数并在字段里附带计算所用样本数便于下游判断置信度。样本数小于 5 的路段直接标记为无有效数据。3.3 数据存储选型时序数据库与空间索引处理后的指标需要同时支持实时查询和离线分析单一数据库很难两全。常见做法是双写路况指标写入时序数据库原始和清洗后的明细写入数据湖或列式存储。时序数据库我这边用得较多的是 InfluxDB 或 TDengine前者生态成熟后者在国产环境下部署更轻。对于需要按经纬度范围查询的路口则配合 PostGIS 做空间索引表。存储对象建议选型关键索引保留周期断面原始记录HDFS/Parquet设备ID时间30天清洗后明细Kafka topic 备份-7天5分钟路况指标TDenginesegment_id, ts90天路段静态信息PostgreSQLPostGISspatial index永久选 TDengine 做实时指标的原因是其超级表模型非常适合按路段分组聚合一个路段模型下挂多个设备查询时按时间窗口拉取即可。数据写入要设置标签包括路段ID、区划、道路等级和限速值这些标签直接在超级表上做过滤能省掉很多 join 操作。4. 信息发布系统设计从API到推送通道4.1 发布接口的时效与权限控制信息发布系统面向的对象包括交通诱导屏、手机App、导航软件和政府网页不同渠道对数据的时效要求和授权级别不同。不能把所有数据以同一个接口暴露出去。我会把接口分为内部高精度数据和外部脱敏数据两层。内部接口返回原始平均速度和拥堵指数供可变情报板控制软件使用走内部网关外部接口只返回拥堵等级、建议绕行提示和旅行时间差供导航合作方调用。接口都要在响应头里注明数据生成时间客户端能根据这个时间判断数据是否够新避免缓存了昨晚的旧数据还当作实时路况。下面是一个 REST 接口的伪代码用 Spring Boot 风格实现同事用 Gin 和 FastAPI 也能实现同等逻辑。RestController RequestMapping(/api/v1/traffic) public class TrafficPublisher { GetMapping(/segment/{segmentId}) public Response publishSegment(PathVariable String segmentId) { // 从缓存取保证响应时间在50ms以内 SegmentStatus s cache.get(status: segmentId); if (s null) { s taosService.queryLatest(segmentId); cache.set(status: segmentId, s, 10, TimeUnit.SECONDS); } return Response.ok(s); } }接口里最关键的参数是缓存过期时间。对外发布的数据请求量会瞬间暴增比如早晚高峰有人频繁刷新因此缓存 5-10 秒即可不需要太长。过长的缓存会让用户在拥堵已缓解时仍看到红色路况容易引发投诉。另一个参数是限流阈值按 ApiKey 和 IP 两级限制单 Key 100 QPS、单 IP 10 QPS 是我常用的初始值。4.2 可变情报板与App推送的数据格式设计不同发布渠道需要不同的数据视图。可变情报板只需要一个拥堵等级和一句话文案手机 App 需要路段列表和预计通行时间而导航合作方需要带坐标的矢量数据。不要在数据库里为每个渠道建表而是从统一路况服务层拼装不同的 DTO。以可变情报板为例其推送报文可以设计成下面的 JSON 格式。{ version: 1.0, timestamp: 2025-06-14T08:30:0008:00, deviceId: VMS-03-020, roadName: 北三环中路, direction: eastbound, level: 3, levelDesc: 缓行, averageSpeed: 32, travelTimeDiff: 3.5, suggestion: 建议绕行北四环 }level取值 1-4 或 0-10 要和采集端的指数映射保持一致否则会出现内部算 3 级、发布端显示 2 级的错位。travelTimeDiff代表比畅通状态下多花的时间单位分钟用于诱导屏上显示“比平时多花 X 分钟”。suggestion字段不要写成固定文案应由后台根据路网状态生成避免前端写死导致信息与实时路况冲突。4.3 多级缓存降低查询压力发布层的查询压力主要来自外部渠道的轮询请求。我采用三级缓存本地 JVM 缓存、Redis 分布式缓存和数据库。请求先查本地再查 Redis最后查数据库。数据库查询只发生在缓存全部失效且数据尚未更新的瞬间。Redis 的 key 可以设计为traffic:segment:{segmentId}value 存完整的 JSON过期时间 85 秒。这样设计的原因是要比采集周期稍长但又不能太长。采集是 20 秒一条处理是 1 分钟一轮缓存设 85 秒可以有效吸收重复查询同时保证最终能看到新的数据。写缓存时用SET EX命令并且在路况计算完成时主动更新而不是等待过期。另外要设计一个发布回调机制。当某条路段的拥堵等级发生跳变比如从畅通直接跳到拥堵时及时推送消息给订阅客户端。使用 Redis 的 Pub/Sub 或 WebSocket 通道都可以但注意要限制推送频率同一路段等级变化的推送间隔最短 30 秒防止频繁抖动把客户端刷死。5. 上线前必须做的三件事压测、回放校验与降级预案5.1 基于历史数据回放验证算法准确性交通流处理算法最怕拍脑袋调参数。上线前我会取过去两周的历史报文按原时间戳顺序重新送入处理管道跑一遍全流程然后把输出的路况指数和历史人工标注对照。回放工具直接用 Python 读取 Kafka 备份 topic按原时间戳重新发送。注意回放倍速不要太高10 倍速是上限否则生产端的linger_ms和消费端窗口聚合会失真。回放后看两个指标拥堵等级准确率要求 85% 以上和平均速度 MAE要求小于 8 km/h。达不到就调整速度阈值和指数映射参数不要改代码硬套。回放输出要存一份作为基准以后每次改完算法都跑同一份数据方便对比回归。这一份基准文件要纳入版本库和算法配置一起管理。5.2 用JMeter压测发布API的吞吐量发布 API 的压测要模拟真实请求分布90% 的请求集中在拥堵路段10% 随机请求。用 JMeter 创建线程组设置 200 并发、持续 15 分钟观察 P99 响应时间和错误率。压测时要做故障注入故意停掉 Redis确认本地缓存能否扛住一分钟。不能的话要在 API 网关层加熔断服务降级时直接返回最近一次的缓存副本而不是去查库。记录吞吐量、P99 延迟、错误率、CPU 和内存峰值。北京路况发布系统的合理目标是单实例支撑 1000 QPS 以上P99 小于 200ms错误率低于 0.1%。5.3 降级预案与数据质量监控传感器故障、网络中断和上游报文格式变化迟早会出现要提前定好降级策略。我的做法是把每条路段的最近一次有效状态保存在本地文件当实时计算输出为空时发布接口返回本地快照数据并在响应头加X-Stale: true标记。数据质量监控不要依赖人工盯屏在处理和发布两个环节各挂一个告警器。处理环节对每个 5 分钟窗口检查输入条数和输出条数如果某个分区输入条数与上周同时间段相比下降超过 50%立即触发告警发布环节对每条推送记录抽样校验字段完整性和速度值范围。告警消息发送到企业微信或钉钉机器人里面要带上设备 ID、路段时间段和异常类型便于快速定位。回放数据文件、压测报告和降级剧本都需要纳入版本库管理每次变更先重跑回放再跑压测确认没有退化后才能上线。本文还有配套的精品资源点击获取
城市交通流数据处理系统实战：从Kafka采集到实时路况发布
简介这是一份关于北京市智能交通系统设计的PDF文档源自2003年《公路交通科技》期刊论文适合交通工程、智慧城市及智能交通领域的研究者与从业者参考。文档围绕交通流数据采集、处理/分析和信息发布三大模块展开介绍了北京当时环形线圈、微波与视频检测三种采集方式以及可变信息显示屏、交通广播、公交网和停车诱导系统等发布渠道并探讨了与信号控制、号牌识别等系统的集成思路。资源为单文件PDF大小仅227KB便于快速阅读与检索目前已有70人在线学习。读者可从中获得北京早期交通流信息系统的原始设计方案、关键数据流框架及实践细节对理解智能交通系统演进或撰写相关方案具有参考价值。1. 为什么交通流系统要先定数据边界再谈技术选型城市交通流数据采集与处理系统最容易被低估的是数据边界问题。北京这样的超大城市单日来自微波、地磁、视频检测器的数据量轻松过亿加上浮动车GPS轨迹数据形态从结构化记录到半结构化报文都有。很多团队一上来就铺Kafka、Spark、时序数据库结果半年后卡在数据对不齐、时间不同步、重复上报和传感器漂移上。这类系统设计分享实际是在讲一套从采集到发布的完整数据链路先定业务指标再倒推采集频率、处理窗口和发布格式。核心原则是每一步都明确数据边界——什么数据必须实时什么数据允许延迟什么数据宁可丢弃也不污染指标。这套方案面向一线系统设计、数据开发和运维人员偏重可落地的实现与排错思路不讨论纯学术算法。2. 采集层设计从断面检测器到浮动车数据的接入规范2.1 数据源分类与采集频率的取舍交通流数据源大致分两类定点断面数据线圈、微波、雷达、视频和移动源数据出租车GPS、网约车轨迹、公交到站数据。定点设备给出精准的断面流量、速度、占有率但覆盖有限浮动车覆盖面广却存在采样率不匀、漂移和城市峡谷信号丢失问题。设计采集规范时我一般先列一个数据源属性表明确每个源的采集频率、上报协议、时间同步方式和质量等级。数据源典型采集频率上报方式主要字段质量风险微波/雷达检测器20秒-1分钟TCP长连接断面流量、平均速度、时间占有率时钟漂移、断点补报地磁线圈30秒HTTP POST车流量、道路占用状态线圈损坏产生零值视频AI检测1-5秒MQTT车牌、车型、坐标光照影响识别率出租车GPS10-30秒Kafka/HTTP经纬度、瞬时速度、载客状态轨迹漂移、重复上报这里的关键是不要对所有数据源用同一套采集线程。定点数据追求确定性用固定周期拉取浮动车数据到达时间分布长尾用队列蓄洪并允许少量乱序。合理的策略是设置两级缓存接入网关先做格式校验和时间戳标准化再进入消息队列同时为每个数据源维护一个设备状态表记录最近上报时间、计数和异常率方便后续的数据质量评估。2.2 用Kafka承接高并发流量的最小配置常见做法是Kafka作为采集总线生产者是各数据采集代理消费者是后续的处理管道。以北京市典型规模估算每秒峰值可能达到数万条记录单分区顺序写就能应付但为了后续并行处理需要按设备ID或地理位置分区。下面是一段Python生产者的最小实现用于从检测器读取数据并写入Kafka。from kafka import KafkaProducer import json, time, random producer KafkaProducer( bootstrap_servers[10.10.0.11:9092, 10.10.0.12:9092], key_serializerlambda k: str(k).encode(), value_serializerlambda v: json.dumps(v).encode(), acksall, # 等所有副本确认避免设备数据丢失 retries3, # 瞬时错误重试3次 linger_ms20, # 攒20ms再发提升吞吐 batch_size32768 # 单批32KB ) while True: packet { device_id: fD035{random.randint(1, 999):03d}, ts: int(time.time() * 1000), volume: random.randint(0, 120), speed: random.randint(10, 80), occupancy: round(random.uniform(0.1, 0.9), 2) } producer.send(traffic-packet, keypacket[device_id], valuepacket) time.sleep(0.02) # 模拟20ms一条acksall和retries3是交通流数据的标配因为感应线圈数据一旦丢了几秒就没法补回来。linger_ms不要设太大控制在 50ms 以下否则会明显增加端到端延迟。batch_size根据单条报文大小调一般 16KB 到 64KB 就够了。注意acksall只保证Kafka副本不丢不等于端到端不丢。生产者进程崩溃前的最后一批数据仍然可能丢失所以消费者侧要配合幂等写或唯一键去重。消费者侧要注意提交偏移量的时机。如果先处理再提交崩溃时会重复处理如果先提交再处理会丢数据。我的做法是业务幂等处理后手动提交在 Java 代码里把enable_auto_commit设为 false在下一批拉取之前提交上一批偏移量。2.3 采集层的容错与重试机制采集网关宕机是常态不能指望硬件不掉链子。要设计独立的死信队列把格式错误、解码失败或时间戳超过当前时间两小时的数据单独存放而不是直接丢弃。每隔一段时间回放死信队列分析哪些设备在持续产生脏数据然后主动把该设备踢出采集链路并通知运维。另外需要统一时钟。工业设备的时钟经常不准我见过一台微波检测器比标准时间快 12 秒导致上下游数据时间错位。最稳妥的解决办法是网关收到数据后不信任设备时间而是以网关接收时间为准并把设备时间一并保存用于诊断。这样即使设备时间错乱后续的计算窗口也不会被污染。3. 处理与分析清洗、聚合与路况计算3.1 流式数据清洗规则与异常值过滤原始数据不能直接进入指标计算清洗是处理层的第一道关。常见问题包括速度值超过路段限速两倍、流量为负、坐标点在河道或公园、同一设备ID在极短时间内出现荒谬的跳变。清洗规则要用配置化方式管理而不是硬编码在程序里。下面是用 Python 和 Pandas 做离线索引数据清洗的一个示例实际流式处理可以改为 Spark Structured Streaming 的 foreachBatch。import pandas as pd def clean_traffic(df): # 剔除缺失关键字段的样本 df df.dropna(subset[device_id, ts, speed]) # 1. 速度范围检查高速公路0-160普通道路0-120超过视为异常置为NA df.loc[(df[road_level] highway) (df[speed] 160), speed] pd.NA df.loc[(df[road_level] urban) (df[speed] 120), speed] pd.NA # 2. 流量与占有率合理性流量为负或占有率大于1直接剔除 df df[(df[volume] 0) (df[occupancy].between(0, 1))] # 3. 时间戳校准距离网关时间超过600秒的数据丢弃 df df[abs(df[ts] - df[gateway_ts]) 600_000] # 4. 去重同一设备、同一时间窗口内保留最后一条 df df.drop_duplicates(subset[device_id, ts], keeplast) return df这个清洗函数里有四个关键参数速度阈值、时间偏差容忍度、去重键和保留策略。速度阈值要按道路等级分别配置因为北京快速路和胡同能跑出的正常速度范围完全不同。时间偏差容忍度设为 10 分钟是经验值太长会让上游断点数据反复进入处理管道太短又会丢弃一些真的因为信号问题延迟上报的有效数据。3.2 路况计算模型从5分钟粒度到拥堵指数路况计算通常以 15 分钟或 5 分钟为窗口将清洗后的点数据聚合成路段平均速度和饱和度。一个常用公式是时间占有率occ和平均速度v结合判断拥堵等级。用流量加权平均速度能避免低速车少时拉低均值。这里给出一个 Spark Structured Streaming 中基于窗口聚类的核心代码。from pyspark.sql import functions as F streaming_df spark.readStream.format(kafka) \ .option(kafka.bootstrap.servers, 10.10.0.11:9092) \ .option(subscribe, traffic-cleaned) \ .load() parsed streaming_df.selectExpr(CAST(value AS STRING) as json) \ .select(F.from_json(json, schema).alias(data)) \ .select(data.*) # 按路段ID和5分钟窗口聚合 agg parsed.withWatermark(ts, 60 seconds) \ .groupBy(segment_id, F.window(ts, 5 minutes)) \ .agg( F.avg(speed).alias(avg_speed), F.avg(occupancy).alias(avg_occupancy), F.count(*).alias(sample_cnt) )withWatermark设置 60 秒是为了容忍乱序数据超过 60 秒晚到的数据直接忽略。window是滚动窗口北京路况发布常用 5 分钟粒度如果要做早晚高峰更细的发布可以改成 3 分钟但会放大单点噪声。从聚合结果计算道路拥堵指数时我习惯于把平均速度映射到 0-10 的指数映射公式按道路等级设定。比如快速路指数10*(1 - v/80) 并限制在 0.5-10 之间普通道路则用期望速度v0作为分母。这个指数要保留两位小数并在字段里附带计算所用样本数便于下游判断置信度。样本数小于 5 的路段直接标记为无有效数据。3.3 数据存储选型时序数据库与空间索引处理后的指标需要同时支持实时查询和离线分析单一数据库很难两全。常见做法是双写路况指标写入时序数据库原始和清洗后的明细写入数据湖或列式存储。时序数据库我这边用得较多的是 InfluxDB 或 TDengine前者生态成熟后者在国产环境下部署更轻。对于需要按经纬度范围查询的路口则配合 PostGIS 做空间索引表。存储对象建议选型关键索引保留周期断面原始记录HDFS/Parquet设备ID时间30天清洗后明细Kafka topic 备份-7天5分钟路况指标TDenginesegment_id, ts90天路段静态信息PostgreSQLPostGISspatial index永久选 TDengine 做实时指标的原因是其超级表模型非常适合按路段分组聚合一个路段模型下挂多个设备查询时按时间窗口拉取即可。数据写入要设置标签包括路段ID、区划、道路等级和限速值这些标签直接在超级表上做过滤能省掉很多 join 操作。4. 信息发布系统设计从API到推送通道4.1 发布接口的时效与权限控制信息发布系统面向的对象包括交通诱导屏、手机App、导航软件和政府网页不同渠道对数据的时效要求和授权级别不同。不能把所有数据以同一个接口暴露出去。我会把接口分为内部高精度数据和外部脱敏数据两层。内部接口返回原始平均速度和拥堵指数供可变情报板控制软件使用走内部网关外部接口只返回拥堵等级、建议绕行提示和旅行时间差供导航合作方调用。接口都要在响应头里注明数据生成时间客户端能根据这个时间判断数据是否够新避免缓存了昨晚的旧数据还当作实时路况。下面是一个 REST 接口的伪代码用 Spring Boot 风格实现同事用 Gin 和 FastAPI 也能实现同等逻辑。RestController RequestMapping(/api/v1/traffic) public class TrafficPublisher { GetMapping(/segment/{segmentId}) public Response publishSegment(PathVariable String segmentId) { // 从缓存取保证响应时间在50ms以内 SegmentStatus s cache.get(status: segmentId); if (s null) { s taosService.queryLatest(segmentId); cache.set(status: segmentId, s, 10, TimeUnit.SECONDS); } return Response.ok(s); } }接口里最关键的参数是缓存过期时间。对外发布的数据请求量会瞬间暴增比如早晚高峰有人频繁刷新因此缓存 5-10 秒即可不需要太长。过长的缓存会让用户在拥堵已缓解时仍看到红色路况容易引发投诉。另一个参数是限流阈值按 ApiKey 和 IP 两级限制单 Key 100 QPS、单 IP 10 QPS 是我常用的初始值。4.2 可变情报板与App推送的数据格式设计不同发布渠道需要不同的数据视图。可变情报板只需要一个拥堵等级和一句话文案手机 App 需要路段列表和预计通行时间而导航合作方需要带坐标的矢量数据。不要在数据库里为每个渠道建表而是从统一路况服务层拼装不同的 DTO。以可变情报板为例其推送报文可以设计成下面的 JSON 格式。{ version: 1.0, timestamp: 2025-06-14T08:30:0008:00, deviceId: VMS-03-020, roadName: 北三环中路, direction: eastbound, level: 3, levelDesc: 缓行, averageSpeed: 32, travelTimeDiff: 3.5, suggestion: 建议绕行北四环 }level取值 1-4 或 0-10 要和采集端的指数映射保持一致否则会出现内部算 3 级、发布端显示 2 级的错位。travelTimeDiff代表比畅通状态下多花的时间单位分钟用于诱导屏上显示“比平时多花 X 分钟”。suggestion字段不要写成固定文案应由后台根据路网状态生成避免前端写死导致信息与实时路况冲突。4.3 多级缓存降低查询压力发布层的查询压力主要来自外部渠道的轮询请求。我采用三级缓存本地 JVM 缓存、Redis 分布式缓存和数据库。请求先查本地再查 Redis最后查数据库。数据库查询只发生在缓存全部失效且数据尚未更新的瞬间。Redis 的 key 可以设计为traffic:segment:{segmentId}value 存完整的 JSON过期时间 85 秒。这样设计的原因是要比采集周期稍长但又不能太长。采集是 20 秒一条处理是 1 分钟一轮缓存设 85 秒可以有效吸收重复查询同时保证最终能看到新的数据。写缓存时用SET EX命令并且在路况计算完成时主动更新而不是等待过期。另外要设计一个发布回调机制。当某条路段的拥堵等级发生跳变比如从畅通直接跳到拥堵时及时推送消息给订阅客户端。使用 Redis 的 Pub/Sub 或 WebSocket 通道都可以但注意要限制推送频率同一路段等级变化的推送间隔最短 30 秒防止频繁抖动把客户端刷死。5. 上线前必须做的三件事压测、回放校验与降级预案5.1 基于历史数据回放验证算法准确性交通流处理算法最怕拍脑袋调参数。上线前我会取过去两周的历史报文按原时间戳顺序重新送入处理管道跑一遍全流程然后把输出的路况指数和历史人工标注对照。回放工具直接用 Python 读取 Kafka 备份 topic按原时间戳重新发送。注意回放倍速不要太高10 倍速是上限否则生产端的linger_ms和消费端窗口聚合会失真。回放后看两个指标拥堵等级准确率要求 85% 以上和平均速度 MAE要求小于 8 km/h。达不到就调整速度阈值和指数映射参数不要改代码硬套。回放输出要存一份作为基准以后每次改完算法都跑同一份数据方便对比回归。这一份基准文件要纳入版本库和算法配置一起管理。5.2 用JMeter压测发布API的吞吐量发布 API 的压测要模拟真实请求分布90% 的请求集中在拥堵路段10% 随机请求。用 JMeter 创建线程组设置 200 并发、持续 15 分钟观察 P99 响应时间和错误率。压测时要做故障注入故意停掉 Redis确认本地缓存能否扛住一分钟。不能的话要在 API 网关层加熔断服务降级时直接返回最近一次的缓存副本而不是去查库。记录吞吐量、P99 延迟、错误率、CPU 和内存峰值。北京路况发布系统的合理目标是单实例支撑 1000 QPS 以上P99 小于 200ms错误率低于 0.1%。5.3 降级预案与数据质量监控传感器故障、网络中断和上游报文格式变化迟早会出现要提前定好降级策略。我的做法是把每条路段的最近一次有效状态保存在本地文件当实时计算输出为空时发布接口返回本地快照数据并在响应头加X-Stale: true标记。数据质量监控不要依赖人工盯屏在处理和发布两个环节各挂一个告警器。处理环节对每个 5 分钟窗口检查输入条数和输出条数如果某个分区输入条数与上周同时间段相比下降超过 50%立即触发告警发布环节对每条推送记录抽样校验字段完整性和速度值范围。告警消息发送到企业微信或钉钉机器人里面要带上设备 ID、路段时间段和异常类型便于快速定位。回放数据文件、压测报告和降级剧本都需要纳入版本库管理每次变更先重跑回放再跑压测确认没有退化后才能上线。本文还有配套的精品资源点击获取
