☰
Django+MySQL智慧交通公交系统:从ORM建模到实时位置模拟实战
2026/10/13 10:54:36 网站建设 项目流程

简介:这是一份基于Python+Django+MySQL的智慧交通公交系统完整项目包,适合Web开发学习者、毕业设计开发者以及对城市智能出行感兴趣的技术人员。项目围绕公交线路、车辆、站点与时刻表等核心数据,完整实现用户管理、线路管理、车辆管理、时刻表、地图展示、数据分析与API接口等模块,覆盖从数据库设计、后端业务逻辑到前端交互的整套流程,可作为快速掌握Django+MySQL联动开发的实战范例。资源共1510个文件,其中HTML/CSS/JS构成前端页面,Python/PYC实现后端逻辑,SQL提供数据库初始化与表结构,PNG/JPG等图片为界面素材,另含文档、BPMN流程模型与概念数据模型等设计文件;压缩包约114.15MB,分层目录便于按模块查阅,代码与设计文档配合可快速理解业务建模与二次开发。目前已有2105人学习,适合需要完整项目参考和动手实践的技术人群。

1. 智慧交通公交系统:先搞清楚 zip 里装的到底是什么

很多开发者的网盘里都躺着这样一个python+django+mysql智慧交通公交系统.zip,名字听着高大上,但解压之后你会发现,它没有多玄妙的算法,本质就是一套公交车信息管理平台:用户能查线路、查站牌、查某辆车现在跑到哪个路口,管理员能维护线路、站点、班次和车辆。Django 负责业务逻辑和后台管理,MySQL 负责把线路、站点、车辆位置这些结构化数据存好。它适合两类人:一类是做课程设计或毕业设计,需要一套能讲清楚前后端如何连接、数据库如何设计的完整项目;另一类是想把某个城市的公交数据从 Excel 搬到线上、做一个能演示的调度大屏的开发者。这篇文章从数据模型一路写到避坑,按我自己落地的方式把这个系统复现一遍。

2. 把公交系统拆成数据模型:Django ORM 与 MySQL 的表设计

2.1 选型理由:Django ORM 为什么适合公交业务快速迭代

公交系统的核心实体是线路、站点、车辆、班次,它们之间是典型的强关系模型:一条线路经过多个站点,一辆车在某个时间点归属一条线路,一个站点被多条线路共享。这种场景用关系型数据库处理最自然,而 Django 自带的 ORM 和迁移机制能把这个过程压缩到几行命令。

常见做法是选 Django 加 MySQL,而不是 Flask 加 MongoDB。原因不复杂:Django 自带 Admin 后台,班级或小团队做运营维护时不用额外写管理界面;ORM 的迁移文件能记录每一次表结构变更,出问题可以回滚;MySQL 在 Linux 服务器上部署方便,索引和事务机制成熟。如果你不是要处理 GPS 轨迹的千万级吞吐,这套组合足够应付每小时几万次查询。

我一般会建议新手把mysqlclient作为驱动,而不是pymysql,因为前者是 C 语言编译的原生驱动,连接速度和稳定性更好。Windows 上装不了mysqlclient时,再退回到pymysql,但需要加上一行pymysql.install_as_MySQLdb()才能被 Django 识别。

2.2 核心模型:车站、线路、班次、实时位置的四张表怎么建模

实际项目中不止四张表,但核心定位是六张:线路、站点、线路站点关系、班次、车辆、实时位置。线路和站点之间是多对多关系,但因为需要保存“第几站”和“上行下行”这类属性,所以不能直接用一个简单的ManyToManyField,必须通过中间表来链接。

下面是我会用的 Django 模型代码,放在bus_app/models.py里:

from django.db import models class Station(models.Model): """公交站点,存经纬度方便后续做地图展示""" name = models.CharField(max_length=64, db_index=True, verbose_name='站点名称') lng = models.DecimalField(max_digits=9, decimal_places=6, verbose_name='经度') lat = models.DecimalField(max_digits=9, decimal_places=6, verbose_name='纬度') def __str__(self): return self.name class Meta: db_table = 'bus_station' class Route(models.Model): """公交线路,例如 一路、二路,包含首末班时间""" name = models.CharField(max_length=32, unique=True, verbose_name='线路名称') code = models.CharField(max_length=16, unique=True, verbose_name='线路编号') start_station = models.ForeignKey(Station, on_delete=models.PROTECT, related_name='routes_as_start', verbose_name='始发站') end_station = models.ForeignKey(Station, on_delete=models.PROTECT, related_name='routes_as_end', verbose_name='终点站') start_time = models.TimeField(verbose_name='首班时间') end_time = models.TimeField(verbose_name='末班时间') def __str__(self): return f'{self.code} {self.name}' class Meta: db_table = 'bus_route' class RouteStation(models.Model): """线路站点关系表,用 order_no 控制站点顺序""" route = models.ForeignKey(Route, on_delete=models.CASCADE, related_name='route_stations', verbose_name='线路') station = models.ForeignKey(Station, on_delete=models.CASCADE, related_name='route_station_links', verbose_name='站点') order_no = models.PositiveIntegerField(default=0, verbose_name='站点顺序') updown = models.SmallIntegerField(default=0, choices=[(0, '上行'), (1, '下行')], verbose_name='上下行') distance_to_next = models.FloatField(default=0, verbose_name='到下一站距离(km)') class Meta: db_table = 'bus_route_station' ordering = ['route', 'updown', 'order_no'] unique_together = [('route', 'station', 'updown', 'order_no')]

我在Route的外键上用了on_delete=models.PROTECT,这是故意为之。如果误删了一条线路,系统会拒绝删除,避免把站点和班次一起带没。对于RouteStation中间表,则用CASCADE,因为中间记录本身没有独立价值,线路删了,中间关系自然应该删掉。

2.3 用 Django Migrations 在 MySQL 里建表:两条必用的命令

模型写完后,需要先确认settings.py里的数据库连接。我的配置通常是:

DATABASES = { 'default': { 'ENGINE': 'django.db.backends.mysql', 'NAME': 'smart_bus', 'USER': 'bus_admin', 'PASSWORD': 'your_password', 'HOST': '127.0.0.1', 'PORT': '3306', 'OPTIONS': { 'charset': 'utf8mb4', 'init_command': "SET sql_mode='STRICT_TRANS_TABLES'", }, } }

然后执行生成迁移和同步数据库两条命令:

python manage.py makemigrations bus_app python manage.py migrate

makemigrations只会生成迁移文件,不会碰数据库;migrate才是真正在 MySQL 里建表。执行完后可以用python manage.py shell进入 Python 解释器,通过from bus_app.models import Route快速验证表能不能操作。需要留意的是,MySQL 8.0 以上版本对字符集的要求是utf8mb4,如果建库时用了utf8,存中文没问题,但存某些生僻字或 Emoji 会报错。

3. 查询引擎与接口:让用户查线路、查站牌、查实时车在哪

3.1 核心查询逻辑:线路查站点、站点查线路的 ORM 写法

公交系统最常见的两个查询场景:一个是给线路 ID,按顺序返回所有站点;一个是给站点 ID,返回所有经过该站点的线路。用 ORM 写起来并不复杂,但如果用错查询时机,会掉进 N+1 的坑。

查线路站点时,我通常会先取出线路,再通过route_stations过滤并按order_no排序:

from bus_app.models import Route def get_route_stations(route_id, updown=0): route = Route.objects.prefetch_related( 'route_stations__station' ).get(id=route_id) stations = [ rs.station.name for rs in route.route_stations.all() if rs.updown == updown ] return stations

这里的关键是prefetch_related('route_stations__station')。如果只写route.route_stations.station,Django 会在查询每个站点时额外发送一条 SQL,30 个站点就是 30 条查询。加上prefetch_related后,所有站点一次聚合成 IN 查询,接口响应时间从几百毫秒降到几十毫秒。

反查站点线路也是一样的思路,用到Station的反向外键:

from django.db.models import Prefetch from bus_app.models import Station, RouteStation def get_routes_by_station(station_id): qs = Route.objects.prefetch_related( Prefetch( 'route_stations', queryset=RouteStation.objects.filter(station_id=station_id), to_attr='hit_stations' ) ).all() return [route.name for route in qs if route.hit_stations]

Prefetch可以自定义子查询条件,to_attr把过滤结果放进临时属性,避免所有站点都被加载。

3.2 用 Django REST Framework 暴露公交查询接口

光有 ORM 不够,前端要拿数据,还得有 HTTP 接口。常规做法是引入djangorestframework,把上面的查询逻辑放进 ViewSet 中。先定义一个简单的序列化器:

from rest_framework import serializers from bus_app.models import Station class StationSerializer(serializers.ModelSerializer): class Meta: model = Station fields = ['id', 'name', 'lng', 'lat']

再定义 ViewSet:

from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.viewsets import ReadOnlyModelViewSet from bus_app.models import Route from bus_app.serializers import StationSerializer from bus_app.utils import get_route_stations, get_routes_by_station class RouteViewSet(ReadOnlyModelViewSet): queryset = Route.objects.all() serializer_class = StationSerializer # 仅示例 @action(detail=True, methods=['get']) def stations(self, request, pk=None): """GET /api/routes/{id}/stations/?updown=0""" updown = request.query_params.get('updown', 0) data = get_route_stations(pk, updown=int(updown)) return Response({'route_id': pk, 'stations': data})

路由配置放在urls.py:

from rest_framework.routers import DefaultRouter from bus_app.views import RouteViewSet router = DefaultRouter() router.register(r'routes', RouteViewSet) urlpatterns = router.urls

ReadOnlyModelViewSet只给只读操作,不需要写接口时尽量不要让系统暴露 POST、PUT 能力。@action生成的是自定义子路径,detail=True表示这个动作是针对单个资源的。

3.3 实时位置模拟与推送:WebSocket 还是轮询

实时位置是“智慧交通”最亮眼的点,但也是最容易过度设计的点。小项目里前端用轮询就够了,不用一上来就上 WebSocket。最简单可靠的轮询方案是额外提供一个接口,比如GET /api/vehicles/current/,返回所有正在运行车辆的位置集合。

后端这个接口的典型实现:

from django.db.models import Prefetch from rest_framework.decorators import api_view from rest_framework.response import Response from bus_app.models import RealtimeLocation, Vehicle @api_view(['GET']) def current_vehicles(request): locations = RealtimeLocation.objects.select_related('vehicle', 'schedule') \ .filter(captured_at__gte='2024-01-01 00:00:00') # 实际应取最近时间 data = [{ 'vehicle_id': loc.vehicle_id, 'plate_no': loc.vehicle.plate_no, 'lng': str(loc.lng), 'lat': str(loc.lat), 'speed': loc.speed, 'direction': loc.direction, 'captured_at': loc.captured_at.strftime('%H:%M:%S') } for loc in locations[:100]] return Response(data)

实际项目中会把“最近时间”定义为datetime.now() - timedelta(minutes=5),避免返回几百条过期位置。前端每隔 5 秒调一次这个接口,地图上的车辆图标就能动起来。只有等用户量超过几百人时,再考虑用 Django Channels 和 Redis 做 WebSocket 推送。基本规律是:项目能演示、并发不高时,轮询是最稳的,还是后悔药最少的方案。

4. 数据从哪来:模拟实时公交位置与运营数据的三种落地方式

4.1 用随机游走模型模拟车辆位置

很多下载来的项目里,实时位置是一堆死数据,演示起来毫无生机。我一般会在开发环境里写一个模拟脚本,让车辆沿着线路动起来。核心思路不是简单地给经纬度加随机数,那会让车开到马路外去。正确做法是沿着线路站点之间的连线插值。

假设一辆车正在从 A 站驶向 B 站,已知两站的经纬度和车辆当前速度,可以这样推进:

import math from random import uniform from bus_app.models import RealtimeLocation, Vehicle def move_along_route(vehicle_id, speed_kmh=30): loc = RealtimeLocation.objects.filter(vehicle_id=vehicle_id).first() if not loc: return # 前端展示用高德/百度地图时,经纬度需要是 GCJ-02 坐标, # 这里直接假设使用模拟坐标 now_lng = float(loc.lng) now_lat = float(loc.lat) target_lng = float(loc.next_station_lng) target_lat = float(loc.next_station_lat) dist_left = math.sqrt((target_lng - now_lng) ** 2 + (target_lat - now_lat) ** 2) step = speed_kmh * 5 / 3600 # 5秒移动的距离,单位km,近似用十进制度替换 if dist_left <= step: # 到达下一站,切换到下一段 loc.next_station_order = (loc.next_station_order + 1) % loc.route_station_count else: ratio = step / dist_left now_lng += (target_lng - now_lng) * ratio now_lat += (target_lat - now_lat) * ratio loc.lng = now_lng loc.lat = now_lat loc.speed = uniform(15, 45) loc.captured_at = datetime.now() loc.save(update_fields=['lng', 'lat', 'speed', 'captured_at'])

update_fields是 Django 里一个容易被忽略的好习惯,它只更新指定列,不会让整行数据因为一次位置移动而重新写一遍,可以减少锁竞争和无效写日志。

4.2 定时任务把位置写入 MySQL:Celery 还是 manage.py 命令

模拟数据写进数据库不能靠人手动点击,需要有定时器持续调用。对于中小项目,我倾向不引入 Celery,而是写一个 Django management command,再用系统 cron 调度。命令文件放在bus_app/management/commands/run_simulation.py:

from django.core.management.base import BaseCommand from django.utils import timezone from bus_app.services import simulate_all_vehicles class Command(BaseCommand): help = '模拟所有在线车辆的位置更新' def add_arguments(self, parser): parser.add_argument('--interval', type=int, default=5, help='每次更新的间隔秒数') def handle(self, *args, **options): interval = options['interval'] while True: simulate_all_vehicles() time.sleep(interval)

需要说明的是,这个命令不建议用while True常驻在普通的管理命令里,它更适合在开发环境调试。生产环境建议把这个循环抽成独立进程,或者用celery beat配置定时任务。如果项目想控制服务生命周期,可以改写成--loop参数实现单次运行,由外部 cron 每 5 秒调用一次。

4.3 让系统能导入真实数据:CSV 导入脚本

模拟数据只能用来演示,真实业务需要批量导入。通常公交公司能导出的数据是 Excel,转成 CSV 后按列映射到我们的模型。我会在management/commands/import_stations.py里写一个可复用的脚本:

import csv from django.core.management.base import BaseCommand from bus_app.models import Station class Command(BaseCommand): help = '从 CSV 导入站点数据,表头: name,lng,lat' def add_arguments(self, parser): parser.add_argument('csv_file', type=str, help='CSV 文件路径') def handle(self, *args, **options): with open(options['csv_file'], 'r', encoding='utf-8-sig') as f: reader = csv.DictReader(f) station_objs = [ Station(name=row['name'], lng=row['lng'], lat=row['lat']) for row in reader ] Station.objects.bulk_create(station_objs, ignore_conflicts=True) self.stdout.write(self.style.SUCCESS(f'成功导入 {len(station_objs)} 条站点'))

utf-8-sig编码是为了兼容从 Windows 导出的带 BOM 头的 CSV,否则第一列字段会被\ufeff污染。ignore_conflicts=True也一样重要,它保证重复导入时不会因为唯一约束中断,最多跳过重复行。

5. 避坑与常见问题:Django+MySQL 公交系统最容易翻车的 5 个地方

5.1 现象:启动项目一直报 Can't connect to MySQL server

刚下载的项目在自己的电脑上跑不起来,最常见原因是 MySQL 服务没启动,或者项目里的数据库账号密码与本地环境不一致。有的项目为了让压缩包开箱即用,会在配置里写死一个root/123456,但你的 MySQL 可能是root加随机密码。另外,HOST 填localhost和127.0.0.1在某些系统上行为不同:localhost可能走 Unix socket,127.0.0.1走 TCP,如果 MySQL 没开 TCP 或者防火墙拦截,前者通后者不通。

解决步骤是先确认服务在运行,再用命令行客户端测试同样的连接串。如果确认服务没问题,就在settings.py里把 HOST 改成127.0.0.1,并检查是否给当前用户做了远程授权。注意grant all privileges on smart_bus.* to 'bus_admin'@'127.0.0.1'这类授权命令必须带上 IP,不要只授权localhost。

5.2 现象:写入中文时提示 Incorrect string value

这个坑在 MySQL 5.7 时代尤其常见,因为默认字符集是latin1。如果你建库时没有显式指定:

CREATE DATABASE smart_bus DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

那么即使 Django 的OPTIONS里写了charset: utf8mb4,表结构和连接的字符集不一致,写入中文仍然报错。

解决方式是先建库,再建表。如果已经建错了,用ALTER DATABASE smart_bus CHARACTER SET utf8mb4;改库,再ALTER TABLE CONVERT TO CHARACTER SET utf8mb4;改表。更关键的是,检查settings.py中OPTIONS是否真的写上了charset,Django 对 MySQL 的默认连接字符集是utf8,它和utf8mb4并不完全等价。

5.3 现象:实时位置比本地时间晚了 8 个小时

位置时间在界面上显示成 UTC 时间,每到中午看过去还是凌晨。原因是settings.py里保留了默认的TIME_ZONE = 'UTC',而USE_TZ = True。Django 默认把所有DateTimeField按 UTC 存入 MySQL,展示时才转成本地时间。如果项目做的是国内公交系统,我建议直接:

TIME_ZONE = 'Asia/Shanghai' USE_TZ = True

USE_TZ = True时,Django ORM 写入前会把本地时间转成 UTC,存进去的其实是 UTC;读取时会转回 Asia/Shanghai。如果不想碰时区转换这团乱麻,可以设置USE_TZ = False,让系统按 MySQL 所在时区纯本地时间处理。两种方案都能用,但务必统一,不能一半靠自动转换一半手动datetime.now()。

5.4 现象:查询接口越调越慢,最后直接超时

公交系统的接口通常只查几十条数据,不至于超时,除非踩了 N+1 查询。比如前面get_route_stations如果不用prefetch_related,每次返回 30 个站点就会产生 31 条 SQL 语句。调试方法是在 Django shell 里打开查询日志:

from django.db import connection connection.queries

观察queries数量,如果一次列表接口产生几十条 SQL,就要检查 ORM 查询链中每一层外键。还有一个容易忽略的点是给常用字段加数据库索引,比如RouteStation的(route, updown, order_no)组合索引,Django 的unique_together会自动创建,不用额外加。但如果经常按站点反查线路,RouteStation.station字段也需要db_index=True,否则查询全表扫描。

5.5 现象:多车并发更新位置时出现 deadlock

模拟脚本开启多个进程后,偶尔会看到Deadlock found when trying to get lock。原因是多个事务同时更新同一张表的不同行,或者同时执行update_or_create时相互等待。我的解决方式是收紧事务范围,让每个位置更新尽量只碰自己的唯一记录。

try: with transaction.atomic(): RealtimeLocation.objects.update_or_create( vehicle_id=vehicle_id, defaults={ 'lng': lng, 'lat': lat, 'speed': speed, 'captured_at': now, } ) except OperationalError: # 重试一次 pass

这里我故意把vehicle_id作为唯一查询条件,并且不用“查询再保存”两步,而是直接update_or_create,事务持有锁的时间尽量短。如果还频繁死锁,需要确认数据库隔离级别是否为REPEATABLE READ,可以临时改成READ COMMITTED,但这一步通常没必要,先检查代码里有没有多余的查询写在事务里面。

6. 进阶:从课程设计到可用的公交信息平台,还有三步要走

项目跑通之后,你可能会发现它还是本地演示版。我的建议是先做三件事:缓存高频查询、接入地图渲染、把 Admin 变成运营后台。

第一,缓存。线路和站点数据基本不变,完全不值得每次请求都查 MySQL。用 Django 的缓存框架配合 Redis,把get_route_stations的结果缓存 60 秒,最简单做法是:

from django.core.cache import cache def get_route_stations_cached(route_id, updown=0): key = f'route_stations_{route_id}_{updown}' data = cache.get(key) if data is None: data = get_route_stations(route_id, updown) cache.set(key, data, 60) return data

实时位置不能缓存太久,否则车辆看起来不动;但可以把计算结果缓 3 秒,足够支撑大屏演示。第二,地图渲染。前端用 Leaflet 加 OpenStreetMap 或国内厂商的地图,把captured_at最新的车辆经纬度画成点,每隔 5 秒重新请求一次,就是一个能看的“智慧交通大屏”。第三,Admin 后台。Django Admin 默认只管理你注册的模型,但公交运营需要按线路查看站点顺序,而不是一张平坦的列表。你可以重写admin.py中的TabularInline,把RouteStation内联到RouteAdmin下面,运营人员一眼看出某条线路的站序。

我最后想提醒的是:这类 zip 项目最大的问题不是代码,而是没有初始数据。解压后跑通只是第一步,真正让系统有价值的是把一条真实线路、十辆车、几十个站点填进去,让接口返回的数据可信。我以前也以为下载来的项目比自己做省事,后来才发现改别人的表结构比重新写还难受。所以如果你时间允许,还是建议把模型按自己业务重写一遍。整个流程走下来,你对 Django ORM、MySQL 设计和接口分层都会上一个台阶。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询