1. 项目概述：基于Django的洗衣服务电商平台

这个项目是一个典型的O2O（线上到线下）洗衣服务平台，采用Django作为后端框架构建。我在实际开发中发现，这类平台需要同时解决三个核心问题：用户端的便捷下单体验、商户端的高效订单管理，以及后台的智能调度系统。Django的全栈特性正好能够完美支持这种复杂业务场景的需求。

洗衣服务平台与传统电商的最大区别在于服务的非标性。用户不仅需要购买服务，还需要完成衣物的交接、状态跟踪和质量反馈。这就要求系统必须设计完善的订单状态机、物流跟踪体系和评价系统。我选择Django的原因在于其强大的ORM和Admin后台，可以快速实现这些业务模型。

2. 核心功能模块设计

2.1 用户端功能实现

用户模块采用Django内置的auth系统进行扩展，增加了手机号验证和地址管理功能。这里有个值得注意的实现细节：为了优化用户体验，我们将用户地址设计为可拖拽地图选点，通过Django REST framework提供地理坐标API。

订单系统是核心中的核心。我设计了一个七状态模型：

1. 待支付
2. 待取件
3. 清洗中
4. 待送还
5. 配送中
6. 已完成
7. 已取消

每个状态转换都通过Django信号机制触发相应操作，比如状态变更为"待取件"时自动推送通知给配送员。

2.2 商户端管理系统

商户后台采用Django Admin深度定制开发。通过重写ModelAdmin的get_queryset方法实现数据隔离，确保商户只能看到自己的订单。我们还开发了批量操作功能，支持一键导出Excel报表。

洗衣品类管理是个技术难点。不同衣物的清洗价格、周期差异很大，我们设计了一个标签系统，通过Django的ManyToManyField实现衣物类型与价格的动态关联。在模型定义时特别使用了through参数来自定义中间表：

class GarmentType(models.Model): name = models.CharField(max_length=50) base_price = models.DecimalField(max_digits=6, decimal_places=2) class ServicePrice(models.Model): garment_type = models.ForeignKey(GarmentType) service_level = models.ForeignKey(ServiceLevel) price = models.DecimalField(max_digits=6, decimal_places=2) duration = models.PositiveIntegerField() # in hours

2.3 智能调度系统

调度算法是本项目最具挑战的部分。我们基于Django的Q对象实现了复杂的查询逻辑，考虑以下因素分配订单：

• 配送员当前位置（通过Redis实时更新）
• 当前负载量
• 专业技能（如皮具护理等特殊技能）
• 历史服务质量评分

核心调度代码使用了Django的annotate和Case/When表达式：

from django.db.models import Case, When, Value, IntegerField couriers = Courier.objects.annotate( distance_score=Case( When(location__distance_lte=(user_location, 5000), then=Value(10)), default=Value(0), output_field=IntegerField() ), load_score=Case( When(active_orders__lt=3, then=Value(5)), default=Value(0), output_field=IntegerField() ) ).order_by('-distance_score', '-load_score')[:5]

3. 关键技术实现细节

3.1 订单状态机设计

使用Django FSM（有限状态机）库实现严谨的状态流转。每个状态变更都记录操作人和时间戳：

from django_fsm import FSMField, transition class Order(models.Model): state = FSMField(default='pending_payment') @transition(field=state, source='pending_payment', target='pending_pickup') def pay(self, by_user): self.payment_time = timezone.now() self.payment_user = by_user @transition(field=state, source='pending_pickup', target='in_cleaning') def pickup(self, by_courier): self.pickup_time = timezone.now() self.pickup_courier = by_courier

3.2 实时通知系统

结合Django Channels和WebSocket实现实时通知。当订单状态变化时，同时推送APP通知和短信：

# signals.py @receiver(post_save, sender=Order) def order_state_change(sender, instance, **kwargs): if 'state' in instance.get_dirty_fields(): async_to_sync(channel_layer.group_send)( f"user_{instance.user.id}", { "type": "order.update", "order_id": instance.id, "new_state": instance.state } ) send_sms.delay( instance.user.phone, f"您的订单{instance.id}状态已变更为{instance.get_state_display()}" )

3.3 支付系统集成

支付模块采用策略模式设计，支持微信支付、支付宝和余额支付三种方式。关键点在于使用Django的ContentType实现通用关联：

class Payment(models.Model): amount = models.DecimalField(max_digits=10, decimal_places=2) content_type = models.ForeignKey(ContentType, on_delete=models.PROTECT) object_id = models.PositiveIntegerField() payment_method = GenericForeignKey('content_type', 'object_id') class WechatPayment(models.Model): prepay_id = models.CharField(max_length=64) # 其他微信支付特有字段 class AlipayPayment(models.Model): trade_no = models.CharField(max_length=64) # 其他支付宝特有字段

4. 性能优化实践

4.1 数据库查询优化

针对订单列表页的N+1查询问题，我们采用select_related和prefetch_related进行优化：

# 优化前（产生N+1查询） orders = Order.objects.filter(user=request.user) for order in orders: print(order.courier.name) # 每次循环都查询数据库 # 优化后 orders = Order.objects.filter(user=request.user).select_related( 'courier', 'payment' ).prefetch_related( Prefetch('items', queryset=OrderItem.objects.select_related('garment_type')) )

4.2 缓存策略

使用Django的缓存框架实现多级缓存：

1. 全页缓存：首页、商品详情页等静态内容
2. 片段缓存：价格计算、推荐列表等
3. 对象缓存：用户资料、商户信息等

from django.core.cache import cache def get_user_profile(user_id): key = f'user_profile_{user_id}' profile = cache.get(key) if not profile: profile = UserProfile.objects.get(user_id=user_id) cache.set(key, profile, timeout=3600) return profile

4.3 异步任务处理

耗时操作如发送短信、生成报表等都通过Celery异步处理。我们特别设计了任务重试机制：

@app.task(bind=True, max_retries=3) def send_sms(self, phone, message): try: sms_client.send(phone, message) except SMSServerError as exc: raise self.retry(exc=exc, countdown=2 ** self.request.retries)

5. 部署架构与运维

5.1 服务器架构设计

采用Docker Compose部署方案，包含以下服务：

• Nginx：负载均衡和静态文件服务
• Gunicorn：Django应用服务器
• PostgreSQL：主数据库
• Redis：缓存和消息队列
• Celery：异步任务处理
• Flower：Celery监控

5.2 宝塔面板部署要点

在宝塔面板部署时需要注意几个关键配置：

1. Python项目管理器中正确设置虚拟环境路径
2. 静态文件收集后配置Nginx的location规则
3. 配置Supervisor管理Celery worker
4. 设置定时任务执行Django管理命令

# 静态文件收集 python manage.py collectstatic --noinput # 迁移数据库 python manage.py migrate # 启动Celery worker celery -A core worker -l info -P gevent

5.3 监控与日志

使用Sentry收集错误日志，配合Django的LOGGING配置实现结构化日志：

LOGGING = { 'version': 1, 'handlers': { 'sentry': { 'level': 'ERROR', 'class': 'sentry_sdk.integrations.logging.EventHandler', }, 'file': { 'level': 'INFO', 'class': 'logging.FileHandler', 'filename': '/var/log/django/app.log', 'formatter': 'verbose', }, }, 'loggers': { 'django': { 'handlers': ['file', 'sentry'], 'level': 'INFO', }, }, }

6. 开发经验与避坑指南

6.1 Django ORM使用技巧

1. 批量操作使用bulk_create和update替代循环save：

# 低效方式 for item in items: item.price = new_price item.save() # 高效方式 Item.objects.filter(id__in=[i.id for i in items]).update(price=new_price)

2. 使用only和defer控制查询字段：

# 只获取需要的字段 Order.objects.only('id', 'state').filter(user=user) # 排除大字段 Order.objects.defer('description').filter(state='completed')

6.2 测试策略

采用分层测试方案：

1. 单元测试：模型方法和工具函数
2. 集成测试：API接口和业务流程
3. E2E测试：关键用户旅程

使用pytest-django提高测试效率：

@pytest.mark.django_db def test_order_flow(): user = UserFactory() order = OrderFactory(user=user) order.pay(by_user=user) assert order.state == 'pending_pickup' order.pickup(by_courier=CourierFactory()) assert order.state == 'in_cleaning'

6.3 安全防护措施

1. CSRF防护：确保所有修改操作都使用Django的csrf_token
2. XSS防护：模板中自动转义，富文本内容使用django-bleach过滤
3. SQL注入：坚持使用ORM，原生SQL必须使用参数化查询
4. 权限控制：结合Django的permission_required和自定义装饰器

@method_decorator(login_required, name='dispatch') class OrderView(UpdateView): @method_decorator(user_passes_test(lambda u: u.is_staff)) def dispatch(self, request, *args, **kwargs): return super().dispatch(request, *args, **kwargs)

7. 项目扩展方向

7.1 小程序端开发

使用Django REST framework提供API，配合Uniapp开发微信小程序。关键是要设计适合移动端的精简数据结构：

class OrderSerializer(serializers.ModelSerializer): class Meta: model = Order fields = ['id', 'state', 'created_at', 'total_price'] extra_kwargs = { 'total_price': {'decimal_places': 2} }

7.2 智能推荐系统

基于用户历史订单数据，实现衣物护理建议和优惠券推荐：

from sklearn.neighbors import NearestNeighbors def recommend_services(user): history = OrderItem.objects.filter(order__user=user) # 构建特征向量并训练模型 nn = NearestNeighbors(n_neighbors=3) nn.fit(feature_matrix) distances, indices = nn.kneighbors([user_vector]) return Service.objects.filter(id__in=indices[0])

7.3 物联网集成

与智能洗衣设备对接，实现状态自动上报。使用Django Channels处理设备WebSocket连接：

class DeviceConsumer(AsyncWebsocketConsumer): async def connect(self): self.device_id = self.scope['url_route']['kwargs']['device_id'] await self.channel_layer.group_add( f"device_{self.device_id}", self.channel_name ) await self.accept() async def receive(self, text_data): data = json.loads(text_data) if data['type'] == 'status_update': await self.update_device_status(data)

在实际开发中，我发现Django的灵活性足以支撑洗衣服务平台从简单到复杂的各种需求。特别是在快速迭代阶段，Django的"电池全包含"理念让团队能够专注于业务逻辑而非框架选型。一个特别有用的经验是：在模型设计阶段就考虑好权限控制和状态流转，这会为后续开发节省大量时间。