第 50 章 / 共 57 章 · 约 9 分钟
LLD 案例:自提快递柜
自提快递柜(Amazon Locker 这类)是一道小而完整的题:格口分配是一个简单的匹配问题,取件码涉及安全,过期处理涉及定时任务。题目不大,正好用来展示你对细节的把握。
需求澄清
功能需求
- 一个站点有若干格口,分小、中、大三种尺寸。
- 快递员投递:扫描包裹,系统分配一个放得下的格口,门打开,放入后关门。
- 投递成功后,系统把 6 位取件码发给收件人。
- 收件人在 3 天内到站点输入取件码,对应格口打开,取件后格口释放。
- 超过 3 天没取的包裹,由退件快递员取走,格口释放。
非功能需求
- 一个格口同一时间只放一个包裹。
- 取件码不能被轻易猜中,要能防住在键盘上反复试码。
明确不做:下单时预留格口(放到扩展讨论)、硬件通信细节。
核心类
| 类 | 职责 |
|---|---|
LockerStation | 一个站点:分配格口、校验取件码、处理过期 |
Compartment | 格口:编号、尺寸、当前放的包裹 |
Size | 尺寸枚举,可以比较大小 |
Parcel | 一件待取的包裹:所在格口、过期时间 |
Notifier | 把取件码发给收件人(短信、App 推送) |
图表可横向滚动,展开后可缩放查看。
查看图表源码
classDiagram
accTitle: 自提快递柜类图:站点、格口、包裹与通知
class LockerStation {
-String id
-List compartments
-Map byCode
-int failures
-DateTime lockedUntil
+deliver(String pkg, Size size) String
+pickup(String code) String
+expired() List
}
class Compartment {
-String id
-Size size
-String pkg
}
class Size {
<<enumeration>>
S
M
L
}
class Parcel {
-Compartment compartment
-String pkg
-DateTime expiresAt
}
class Notifier {
<<interface>>
+sendCode(String customer, String code)
}
LockerStation "1" *-- "many" Compartment
LockerStation "1" o-- "many" Parcel : by code
Compartment --> Size
Parcel --> Compartment
LockerStation ..> Notifier
用到的设计模式
- 策略模式:格口分配规则可以替换。默认是“能放下的最小格口”,也可以换成“离地面最近的格口”方便老人取件。
- 观察者模式:投递完成、即将过期、已退回这些事件通知收件人。
- 整数可比较的枚举:
Size继承IntEnum,“格口尺寸 ≥ 包裹尺寸”可以直接比较。
核心代码
import secrets
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import IntEnum
class Size(IntEnum):
S = 1
M = 2
L = 3
@dataclass
class Compartment:
id: str
size: Size
pkg: str | None = None
@dataclass
class Parcel:
compartment: Compartment
pkg: str
expires_at: datetime
class LockerStation:
KEEP = timedelta(days=3)
def __init__(self, compartments: list[Compartment], clock=datetime.now):
self.compartments = sorted(compartments, key=lambda c: c.size) # smallest first
self.by_code: dict[str, Parcel] = {}
self.failures = 0
self.locked_until = datetime.min
self.clock = clock
def deliver(self, pkg: str, size: Size) -> str:
box = next((c for c in self.compartments if c.pkg is None and c.size >= size), None)
if box is None:
raise RuntimeError("station full") # the courier picks another station
box.pkg = pkg
code = self._new_code()
self.by_code[code] = Parcel(box, pkg, self.clock() + self.KEEP)
return code # texted to the customer, never shown to the courier
def pickup(self, code: str) -> str:
now = self.clock()
if now < self.locked_until:
raise RuntimeError("keypad locked, try again later")
parcel = self.by_code.get(code)
if parcel is None or now > parcel.expires_at:
self.failures += 1
if self.failures % 5 == 0: # every 5th miss: lock 2, 4, 8 ... minutes
self.locked_until = now + timedelta(minutes=2 ** (self.failures // 5))
raise PermissionError("wrong or expired code")
self.failures = 0
return self._release(code) # the door opens
def expired(self) -> list[str]:
"""Run daily: parcels past their date go back with the returns courier."""
now = self.clock()
return [self._release(c) for c, p in list(self.by_code.items()) if now > p.expires_at]
def _release(self, code: str) -> str:
parcel = self.by_code.pop(code)
parcel.compartment.pkg = None
return parcel.pkg
def _new_code(self) -> str:
while True: # unique among the codes live at this station
code = f"{secrets.randbelow(10**6):06d}"
if code not in self.by_code:
return code
分配:能放下的最小格口
格口按尺寸从小到大排好,取第一个空闲且放得下的。小包裹优先放小格口,可以把大格口留给只能放进大格口的包裹;反过来,小包裹占了大格口,后面来一个大包裹就没地方放了。站点只有几十到上百个格口,线性扫描足够。格口多了以后可以按尺寸分成三个空闲队列,和停车场的空闲车位池一样。
取件码
- 唯一性只需要在站点内。收件人是在某个站点的键盘上输码,所以“站点 + 取件码”能定位包裹就够了;取走以后,这个码可以再次使用。
- 用安全的随机数。
secrets生成的码不可预测;用自增序号或时间戳,别人很容易推算出下一个码。 - 限制试码。6 位数字共 100 万种组合,站点里同时有效的码可能有上百个,如果不限次数,随机猜中的概率并不低。代码里每错 5 次就锁定键盘,锁定时间按 2、4、8 分钟翻倍,同时可以给运维发告警。
- 码只发给收件人,快递员看不到,避免投递环节泄露。
过期处理
expired 每天跑一次,把超过 3 天的包裹交给退件流程并释放格口。过期的码在 pickup 里也会被当作无效码,所以即使定时任务还没跑,过期包裹也取不走。
扩展讨论
- 下单时预留格口:用户在结账时选了某个站点,可以先预留一个格口,预计送达前保留,送达后转为占用;没送到就释放。预留要有截止时间,否则格口会被一直占着。
- 站点已满:投递时没有空格口,系统推荐附近站点,并通知收件人取件地点变了。
- 退货:用户在柜子上发起退货,系统分配格口并生成投递码,快递员按码取走。
- 离线:站点和服务器断网时,柜子本地保存取件码的哈希,仍能开门;恢复联网后同步取件记录。