跳转到内容

第 9 章 · 适配器(Adapter)

一句话:在调用方和一个接口对不上的类之间加一层翻译,让调用方只看到自己定义的接口。 解决的痛:第三方 SDK 的接口各不相同,转换代码散在每个调用处,每接一家就要把每一处都改一遍,漏改一处就出错。

你在一家电商公司写后台。订单发货以后,用户和客服都要查物流:用户在订单详情页看包裹到哪了,客服在客服工具里查订单,回答“我的快递怎么还没到”。

公司接了两家快递:顺丰和中通。两家都提供 SDK,但长得完全不一样:

  • 顺丰:SfClient.query_route(waybill) 返回一个 dict,轨迹在 routes 列表里,最新的一条在最后;时间是字符串 "2025-01-11 16:40:00";状态是操作码 opcode,比如 "30" 表示运输中。
  • 中通:ZtoApi.get_trace(bill_no, lang) 直接返回一个列表,最新的一条在最前;时间是秒级时间戳(Unix timestamp);状态是数字码,比如 3 表示派送中。

发货时我们把自己的订单号作为客户单号交给了快递公司,所以两家都能直接按订单号查。真实项目里这些 SDK 背后是 HTTP 请求,本章用返回固定数据的几行类代替。

页面要的东西两家是一样的:一个状态(统一成“已揽收、运输中、派送中、已签收”四种中文)、最后一条轨迹的文字、更新时间。更新时间统一用秒级时间戳传,页面自己格式化成北京时间。你要写的就是订单详情页和客服工具这两个调用处:按订单查出物流,整理成这三样,显示出来。

最直接的写法是在每个调用处判断这单走的是哪家快递,调用那家的 SDK,再把返回值整理成页面要的字段:

from datetime import datetime, timedelta, timezone
CST = timezone(timedelta(hours=8)) # 北京时间
class SfClient: # 顺丰 SDK(模拟):最新的轨迹在 routes 最后,时间是字符串
def query_route(self, waybill: str) -> dict:
return {"routes": [
{"accept_time": "2025-01-11 09:12:00", "remark": "顺丰速运 已收取快件", "opcode": "50"},
{"accept_time": "2025-01-11 16:40:00", "remark": "快件到达 深圳转运中心", "opcode": "30"},
]}
class ZtoApi: # 中通 SDK(模拟):最新的轨迹在最前,时间是秒级时间戳,状态是数字码
def get_trace(self, bill_no: str, lang: str) -> list[dict]:
return [
{"scan_time": 1736589600, "desc": "【杭州市】派件员 王师傅 正在派件", "status": 3},
{"scan_time": 1736560800, "desc": "【杭州市】快件已到达 杭州中转部", "status": 2},
]
SF_STATUS = {"50": "已揽收", "30": "运输中", "44": "派送中", "80": "已签收"}
ZTO_STATUS = {1: "已揽收", 2: "运输中", 3: "派送中", 4: "已签收"}
ORDERS = {"A1001": "sf", "A1002": "zto"} # 订单号 → 快递公司
def parse_time(text: str) -> int:
return int(datetime.strptime(text, "%Y-%m-%d %H:%M:%S").replace(tzinfo=CST).timestamp())
def fmt_time(ts: int) -> str: # 页面显示用:秒级时间戳 → 北京时间
return datetime.fromtimestamp(ts, CST).strftime("%Y-%m-%d %H:%M")
# 调用点一:订单详情页
def order_detail_page(order_no: str) -> None:
carrier = ORDERS[order_no]
if carrier == "sf":
last = SfClient().query_route(order_no)["routes"][-1] # 顺丰:最新的在最后
status, event = SF_STATUS[last["opcode"]], last["remark"]
updated_at = parse_time(last["accept_time"])
elif carrier == "zto":
last = ZtoApi().get_trace(order_no, "zh-CN")[0] # 中通:最新的在最前
status, event = ZTO_STATUS[last["status"]], last["desc"]
updated_at = last["scan_time"]
else:
raise ValueError(f"未知快递:{carrier}")
print(f"订单 {order_no}:{status} · {event} · {fmt_time(updated_at)}")
# 调用点二:客服工具
def support_lookup(order_no: str) -> None:
carrier = ORDERS[order_no]
if carrier == "sf":
last = SfClient().query_route(order_no)["routes"][-1]
status, event = SF_STATUS[last["opcode"]], last["remark"]
updated_at = parse_time(last["accept_time"])
elif carrier == "zto":
last = ZtoApi().get_trace(order_no, "zh-CN")[0]
status, event = ZTO_STATUS[last["status"]], last["desc"]
updated_at = last["scan_time"]
else:
raise ValueError(f"未知快递:{carrier}")
print(f"[客服] {order_no} {status},最后轨迹:{event}({fmt_time(updated_at)})")
if __name__ == "__main__":
order_detail_page("A1001")
support_lookup("A1002")
# 输出:
# 订单 A1001:运输中 · 快件到达 深圳转运中心 · 2025-01-11 16:40
# [客服] A1002 派送中,最后轨迹:【杭州市】派件员 王师傅 正在派件(2025-01-11 18:00)

TypeScript 版结构一样,SDK 的方法名按 TS 习惯写成驼峰,返回数据的字段名保持原样:

interface SfRoute { accept_time: string; remark: string; opcode: string }
interface ZtoTrace { scan_time: number; desc: string; status: number }
class SfClient { // 顺丰 SDK(模拟):最新的轨迹在 routes 最后,时间是字符串
queryRoute(waybill: string): { routes: SfRoute[] } {
return { routes: [
{ accept_time: "2025-01-11 09:12:00", remark: "顺丰速运 已收取快件", opcode: "50" },
{ accept_time: "2025-01-11 16:40:00", remark: "快件到达 深圳转运中心", opcode: "30" },
] };
}
}
class ZtoApi { // 中通 SDK(模拟):最新的轨迹在最前,时间是秒级时间戳,状态是数字码
getTrace(billNo: string, lang: string): ZtoTrace[] {
return [
{ scan_time: 1736589600, desc: "【杭州市】派件员 王师傅 正在派件", status: 3 },
{ scan_time: 1736560800, desc: "【杭州市】快件已到达 杭州中转部", status: 2 },
];
}
}
const SF_STATUS: Record<string, string> = { "50": "已揽收", "30": "运输中", "44": "派送中", "80": "已签收" };
const ZTO_STATUS: Record<number, string> = { 1: "已揽收", 2: "运输中", 3: "派送中", 4: "已签收" };
const ORDERS: Record<string, string> = { A1001: "sf", A1002: "zto" }; // 订单号 → 快递公司
function parseTime(text: string): number {
return Date.parse(text.replace(" ", "T") + "+08:00") / 1000;
}
function fmtTime(ts: number): string { // 页面显示用:秒级时间戳 → 北京时间
return new Date((ts + 8 * 3600) * 1000).toISOString().slice(0, 16).replace("T", " ");
}
// 调用点一:订单详情页
function orderDetailPage(orderNo: string): void {
const carrier = ORDERS[orderNo];
let status: string, event: string, updatedAt: number;
if (carrier === "sf") {
const routes = new SfClient().queryRoute(orderNo).routes;
const last = routes[routes.length - 1]; // 顺丰:最新的在最后
[status, event, updatedAt] = [SF_STATUS[last.opcode], last.remark, parseTime(last.accept_time)];
} else if (carrier === "zto") {
const last = new ZtoApi().getTrace(orderNo, "zh-CN")[0]; // 中通:最新的在最前
[status, event, updatedAt] = [ZTO_STATUS[last.status], last.desc, last.scan_time];
} else {
throw new Error(`未知快递:${carrier}`);
}
console.log(`订单 ${orderNo}:${status} · ${event} · ${fmtTime(updatedAt)}`);
}
// 调用点二:客服工具
function supportLookup(orderNo: string): void {
const carrier = ORDERS[orderNo];
let status: string, event: string, updatedAt: number;
if (carrier === "sf") {
const routes = new SfClient().queryRoute(orderNo).routes;
const last = routes[routes.length - 1];
[status, event, updatedAt] = [SF_STATUS[last.opcode], last.remark, parseTime(last.accept_time)];
} else if (carrier === "zto") {
const last = new ZtoApi().getTrace(orderNo, "zh-CN")[0];
[status, event, updatedAt] = [ZTO_STATUS[last.status], last.desc, last.scan_time];
} else {
throw new Error(`未知快递:${carrier}`);
}
console.log(`[客服] ${orderNo} ${status},最后轨迹:${event}(${fmtTime(updatedAt)})`);
}
orderDetailPage("A1001");
supportLookup("A1002");
// 输出:
// 订单 A1001:运输中 · 快件到达 深圳转运中心 · 2025-01-11 16:40
// [客服] A1002 派送中,最后轨迹:【杭州市】派件员 王师傅 正在派件(2025-01-11 18:00)

两个调用处各自完整:一单走哪家、调了哪个 SDK、哪个字段变成了页面上的哪一项,从上往下读一遍就清楚。只接两家快递、只有两个地方要查物流时,这样写完全够用,大多数人接第一个 SDK 时都这么写。

公司开始用京东物流发大件,同时客服主管提了一条:接入京东物流;客服工具加一个“按订单批量查询”,一次贴进几十个订单号,列出每单的状态和更新时间。

京东的 SDK 又是另一副样子:JdClient.query_trace(order_id) 返回 {"traces": [...]},最新的在最后;状态是英文枚举,比如 "SIGNED" 表示已签收;时间是毫秒级时间戳,比秒多三位。

要改的地方有四处:

  1. 新增京东的状态对照表 JD_STATUS,弄清楚京东的字段怎么变成页面要的三样(枚举转中文、毫秒除以 1000)。这一处任何写法都躲不掉。
  2. order_detail_page:加一个 elif carrier == "jd" 分支,在分支里整理京东的字段。
  3. support_lookup:把同样的分支和整理再写一遍。
  4. 新增 support_batch_lookup:support_lookup 是直接打印的,批量查询要的是一张短表,用不上它,于是三家的 if 连同整理逻辑整串复制进新函数。

改完之后,京东的分支在三个函数里各有一份:

# 第 1 处:新增的对照表,和 SF_STATUS、ZTO_STATUS 放在一起
JD_STATUS = {"PICKED_UP": "已揽收", "IN_TRANSIT": "运输中", "DELIVERING": "派送中", "SIGNED": "已签收"}
# order_detail_page 里新增
elif carrier == "jd":
last = JdClient().query_trace(order_no)["traces"][-1]
status, event = JD_STATUS[last["state"]], last["opDesc"]
updated_at = last["opTime"] // 1000 # 毫秒 → 秒
# support_lookup 里新增
elif carrier == "jd":
last = JdClient().query_trace(order_no)["traces"][-1]
status, event = JD_STATUS[last["state"]], last["opDesc"]
updated_at = last["opTime"] # 抄的时候少了 // 1000
# support_batch_lookup:顺丰、中通、京东三个分支整串再抄一遍

TS 版要改的是同样的四处,京东分支在 orderDetailPage、supportLookup 和新的 supportBatchLookup 里各写一份。

事故就出在第 3 处。订单详情页测过了,京东大件的更新时间显示正常;客服工具那一份少写了 // 1000。京东返回的最后一条轨迹时间是 1736591400000(北京时间 2025-01-11 18:30),被原样当成秒交给了客服页面。真实的客服页面在浏览器里格式化时间,相当于执行 new Date(1736591400000 * 1000).toLocaleString("zh-CN", { timeZone: "Asia/Shanghai" }),客服看到的更新时间是 57000/5/23 20:00:00,把这一句贴进浏览器控制台就能复现。本章代码里模拟页面的格式化函数会是另外两种样子:Python 的 fmt_time 换算不出这个年份,直接抛异常;TS 的 fmtTime 打印出 +057000-05-23 20。客服把截图发到群里,大家笑了一阵。更麻烦的是不显眼的抄错:第 4 处整串复制时,京东分支是照着中通分支改的,取的还是 ["traces"][0],可京东最新的一条在最后,于是批量查询里已经签收的京东包裹全显示“已揽收”。状态、时间看上去都像真的,没人会立刻发现。

测试也跟着麻烦。想给客服工具写集成测试,又不想真的去调三家快递的接口,就得换一家“假物流”。可是调哪个 SDK 写死在每个 if 里,只能在三个函数里各加一个 elif carrier == "fake",让测试专用的分支进入生产代码。

问题的形状是:每家快递的怪癖(最新一条在哪头、状态怎么编码、时间用什么单位)在每个调用处都要整理一遍;调用处有几个,同一段翻译就抄几份,每加一家快递或一个调用处,都要再抄一轮。

朴素写法:四处要改的位置

图中 ①–④ 就是上面列出的四处,三个调用处各自连着三家 SDK,每条线上都有一段自己的字段整理。

调用处真正想要的只有一句话:“给我这个订单的物流”。那就先把这句话写成我们自己的接口 Tracker:只有一个方法 track(order_no),返回一个 TrackingInfo,里面正好是页面要的三样。三家 SDK 我们改不了,就给每家写一个适配器:SfAdapter、ZtoAdapter、JdAdapter 各自持有那家的 SDK 对象,track() 被调用时,转去调那家的方法,再把返回值翻译成 TrackingInfo。调用处只认 Tracker:

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Protocol
CST = timezone(timedelta(hours=8)) # 北京时间
@dataclass(frozen=True)
class TrackingInfo: # 页面要的三样
status: str
last_event: str
updated_at: int # 秒级时间戳
class Tracker(Protocol): # 我们自己定义的接口,调用处只认它
def track(self, order_no: str) -> TrackingInfo: ...
class SfClient: # 顺丰 SDK(模拟),和 9.2 相同
def query_route(self, waybill: str) -> dict:
return {"routes": [
{"accept_time": "2025-01-11 09:12:00", "remark": "顺丰速运 已收取快件", "opcode": "50"},
{"accept_time": "2025-01-11 16:40:00", "remark": "快件到达 深圳转运中心", "opcode": "30"},
]}
class ZtoApi: # 中通 SDK(模拟),和 9.2 相同
def get_trace(self, bill_no: str, lang: str) -> list[dict]:
return [
{"scan_time": 1736589600, "desc": "【杭州市】派件员 王师傅 正在派件", "status": 3},
{"scan_time": 1736560800, "desc": "【杭州市】快件已到达 杭州中转部", "status": 2},
]
class JdClient: # 京东物流 SDK(模拟):最新的在最后,状态是英文枚举,时间是毫秒
def query_trace(self, order_id: str) -> dict:
return {"traces": [
{"opTime": 1736553600000, "opDesc": "京东物流 已揽件", "state": "PICKED_UP"},
{"opTime": 1736591400000, "opDesc": "您的订单已由本人签收", "state": "SIGNED"},
]}
class SfAdapter: # 顺丰的怪癖只在这个类里
STATUS = {"50": "已揽收", "30": "运输中", "44": "派送中", "80": "已签收"}
def __init__(self, client: SfClient) -> None:
self.client = client
def track(self, order_no: str) -> TrackingInfo:
last = self.client.query_route(order_no)["routes"][-1] # 最新的在最后
updated_at = datetime.strptime(last["accept_time"], "%Y-%m-%d %H:%M:%S").replace(tzinfo=CST)
return TrackingInfo(self.STATUS[last["opcode"]], last["remark"], int(updated_at.timestamp()))
class ZtoAdapter: # 中通的怪癖只在这个类里
STATUS = {1: "已揽收", 2: "运输中", 3: "派送中", 4: "已签收"}
def __init__(self, api: ZtoApi) -> None:
self.api = api
def track(self, order_no: str) -> TrackingInfo:
last = self.api.get_trace(order_no, "zh-CN")[0] # 最新的在最前
return TrackingInfo(self.STATUS[last["status"]], last["desc"], last["scan_time"])
class JdAdapter: # 新接的京东:加这一个类
STATUS = {"PICKED_UP": "已揽收", "IN_TRANSIT": "运输中", "DELIVERING": "派送中", "SIGNED": "已签收"}
def __init__(self, client: JdClient) -> None:
self.client = client
def track(self, order_no: str) -> TrackingInfo:
last = self.client.query_trace(order_no)["traces"][-1]
return TrackingInfo(self.STATUS[last["state"]], last["opDesc"], last["opTime"] // 1000) # 毫秒 → 秒,只写这一次
TRACKERS: dict[str, Tracker] = {"sf": SfAdapter(SfClient()), "zto": ZtoAdapter(ZtoApi()), "jd": JdAdapter(JdClient())}
ORDERS = {"A1001": "sf", "A1002": "zto", "A1003": "jd"} # 订单号 → 快递公司
def get_tracker(carrier: str) -> Tracker:
if carrier not in TRACKERS:
raise ValueError(f"未知快递:{carrier}")
return TRACKERS[carrier]
def fmt_time(ts: int) -> str:
return datetime.fromtimestamp(ts, CST).strftime("%Y-%m-%d %H:%M")
# 调用点一:订单详情页
def order_detail_page(tracker: Tracker, order_no: str) -> None:
info = tracker.track(order_no)
print(f"订单 {order_no}:{info.status} · {info.last_event} · {fmt_time(info.updated_at)}")
# 调用点二:客服工具
def support_lookup(tracker: Tracker, order_no: str) -> None:
info = tracker.track(order_no)
print(f"[客服] {order_no} {info.status},最后轨迹:{info.last_event}({fmt_time(info.updated_at)})")
# 调用点三:客服工具的批量查询,就是对 Tracker 循环
def support_batch_lookup(items: list[tuple[str, Tracker]]) -> None:
print(f"[客服] 批量查询 {len(items)} 单")
for order_no, tracker in items:
info = tracker.track(order_no)
print(f"- {order_no} {info.status} {fmt_time(info.updated_at)}")
if __name__ == "__main__":
order_detail_page(get_tracker(ORDERS["A1003"]), "A1003")
support_lookup(get_tracker(ORDERS["A1001"]), "A1001")
support_batch_lookup([(no, get_tracker(carrier)) for no, carrier in ORDERS.items()])
# 输出:
# 订单 A1003:已签收 · 您的订单已由本人签收 · 2025-01-11 18:30
# [客服] A1001 运输中,最后轨迹:快件到达 深圳转运中心(2025-01-11 16:40)
# [客服] 批量查询 3 单
# - A1001 运输中 2025-01-11 16:40
# - A1002 派送中 2025-01-11 18:00
# - A1003 已签收 2025-01-11 18:30

TS 版的 Tracker 是 interface,适配器用构造参数属性持有 SDK 对象:

interface TrackingInfo { // 页面要的三样
readonly status: string;
readonly lastEvent: string;
readonly updatedAt: number; // 秒级时间戳
}
interface Tracker { // 我们自己定义的接口,调用处只认它
track(orderNo: string): TrackingInfo;
}
interface SfRoute { accept_time: string; remark: string; opcode: string }
interface ZtoTrace { scan_time: number; desc: string; status: number }
interface JdTrace { opTime: number; opDesc: string; state: string }
class SfClient { // 顺丰 SDK(模拟),和 9.2 相同
queryRoute(waybill: string): { routes: SfRoute[] } {
return { routes: [
{ accept_time: "2025-01-11 09:12:00", remark: "顺丰速运 已收取快件", opcode: "50" },
{ accept_time: "2025-01-11 16:40:00", remark: "快件到达 深圳转运中心", opcode: "30" },
] };
}
}
class ZtoApi { // 中通 SDK(模拟),和 9.2 相同
getTrace(billNo: string, lang: string): ZtoTrace[] {
return [
{ scan_time: 1736589600, desc: "【杭州市】派件员 王师傅 正在派件", status: 3 },
{ scan_time: 1736560800, desc: "【杭州市】快件已到达 杭州中转部", status: 2 },
];
}
}
class JdClient { // 京东物流 SDK(模拟):最新的在最后,状态是英文枚举,时间是毫秒
queryTrace(orderId: string): { traces: JdTrace[] } {
return { traces: [
{ opTime: 1736553600000, opDesc: "京东物流 已揽件", state: "PICKED_UP" },
{ opTime: 1736591400000, opDesc: "您的订单已由本人签收", state: "SIGNED" },
] };
}
}
class SfAdapter implements Tracker { // 顺丰的怪癖只在这个类里
static readonly STATUS: Record<string, string> = { "50": "已揽收", "30": "运输中", "44": "派送中", "80": "已签收" };
constructor(private readonly client: SfClient) {}
track(orderNo: string): TrackingInfo {
const routes = this.client.queryRoute(orderNo).routes;
const last = routes[routes.length - 1]; // 最新的在最后
const updatedAt = Date.parse(last.accept_time.replace(" ", "T") + "+08:00") / 1000;
return { status: SfAdapter.STATUS[last.opcode], lastEvent: last.remark, updatedAt };
}
}
class ZtoAdapter implements Tracker { // 中通的怪癖只在这个类里
static readonly STATUS: Record<number, string> = { 1: "已揽收", 2: "运输中", 3: "派送中", 4: "已签收" };
constructor(private readonly api: ZtoApi) {}
track(orderNo: string): TrackingInfo {
const last = this.api.getTrace(orderNo, "zh-CN")[0]; // 最新的在最前
return { status: ZtoAdapter.STATUS[last.status], lastEvent: last.desc, updatedAt: last.scan_time };
}
}
class JdAdapter implements Tracker { // 新接的京东:加这一个类
static readonly STATUS: Record<string, string> = {
PICKED_UP: "已揽收", IN_TRANSIT: "运输中", DELIVERING: "派送中", SIGNED: "已签收",
};
constructor(private readonly client: JdClient) {}
track(orderNo: string): TrackingInfo {
const traces = this.client.queryTrace(orderNo).traces;
const last = traces[traces.length - 1];
const updatedAt = Math.floor(last.opTime / 1000); // 毫秒 → 秒,只写这一次
return { status: JdAdapter.STATUS[last.state], lastEvent: last.opDesc, updatedAt };
}
}
const TRACKERS: Record<string, Tracker> = {
sf: new SfAdapter(new SfClient()), zto: new ZtoAdapter(new ZtoApi()), jd: new JdAdapter(new JdClient()),
};
const ORDERS: Record<string, string> = { A1001: "sf", A1002: "zto", A1003: "jd" }; // 订单号 → 快递公司
function getTracker(carrier: string): Tracker {
const tracker = TRACKERS[carrier];
if (tracker === undefined) throw new Error(`未知快递:${carrier}`);
return tracker;
}
function fmtTime(ts: number): string {
return new Date((ts + 8 * 3600) * 1000).toISOString().slice(0, 16).replace("T", " ");
}
// 调用点一:订单详情页
function orderDetailPage(tracker: Tracker, orderNo: string): void {
const info = tracker.track(orderNo);
console.log(`订单 ${orderNo}:${info.status} · ${info.lastEvent} · ${fmtTime(info.updatedAt)}`);
}
// 调用点二:客服工具
function supportLookup(tracker: Tracker, orderNo: string): void {
const info = tracker.track(orderNo);
console.log(`[客服] ${orderNo} ${info.status},最后轨迹:${info.lastEvent}(${fmtTime(info.updatedAt)})`);
}
// 调用点三:客服工具的批量查询,就是对 Tracker 循环
function supportBatchLookup(items: [string, Tracker][]): void {
console.log(`[客服] 批量查询 ${items.length} 单`);
for (const [orderNo, tracker] of items) {
const info = tracker.track(orderNo);
console.log(`- ${orderNo} ${info.status} ${fmtTime(info.updatedAt)}`);
}
}
orderDetailPage(getTracker(ORDERS["A1003"]), "A1003");
supportLookup(getTracker(ORDERS["A1001"]), "A1001");
supportBatchLookup(Object.entries(ORDERS).map(([no, carrier]): [string, Tracker] => [no, getTracker(carrier)]));
// 输出:
// 订单 A1003:已签收 · 您的订单已由本人签收 · 2025-01-11 18:30
// [客服] A1001 运输中,最后轨迹:快件到达 深圳转运中心(2025-01-11 16:40)
// [客服] 批量查询 3 单
// - A1001 运输中 2025-01-11 16:40
// - A1002 派送中 2025-01-11 18:00
// - A1003 已签收 2025-01-11 18:30

适配器版:三个适配器实现同一个 Tracker 接口

看标红的 JdAdapter:接京东就是这一个类,它持有 JdClient,对外只露出 track。

一次 tracker.track 调用穿过 ZtoAdapter 的顺序

看顺序图:调用处只发出 track("A1002"),“取第 0 条”“数字码 3 换成派送中”这些中通专有的步骤全发生在 ZtoAdapter 里面,调用处拿回来的已经是 TrackingInfo。

逐条对照 9.3 的四处改动:

  • 第 1 处,京东的状态对照和字段整理:还是要写,这是京东本身的知识,现在写在 JdAdapter 里。除了这个类,TRACKERS 里还要加一行 "jd": JdAdapter(JdClient())。任何写法都得在某个地方登记一家新快递,适配器省不掉这一行。
  • 第 2、3 处,两个调用处里的京东分支:不用加了。调用处只调 tracker.track(order_no),里面没有任何按快递公司的判断,接京东时一行没动。
  • 第 4 处,批量查询:就是一个对 Tracker 的循环,三行,不用再抄三家的整理逻辑。以后再加“导出物流报表”“物流超时提醒”,也都是拿 Tracker 来用。
  • 事故:// 1000 和“取最后一条”都只写在 JdAdapter.track 一处。写错了当然还会错,但错在一处,订单详情页、客服工具和批量查询会一起显示同一个错的年份或状态,第一个测京东订单的人就能看见,改一行全好;不会再出现“详情页对、客服工具错”这种只有一部分人看得到的错。

测试时换假物流,也不用碰生产代码,在测试文件里写一个实现了 Tracker 的类就行:

class FakeTracker:
def track(self, order_no: str) -> TrackingInfo:
return TrackingInfo("派送中", "测试轨迹", 1736589600)
# support_lookup(FakeTracker(), "T001") 不会去调任何快递公司的接口

要说清楚适配器的收益在哪:加京东要写的东西(状态对照、时间换算)一样不少,映射表也照样加一行。它换来的是每家快递的怪癖只住在一个类里,调用处只看见一个接口,所以调用处再多也不用再抄。代价是多了一个接口和三个类,读代码时想知道“派送中”是从哪个字段来的,要跳进对应的适配器去看。

GoF 对适配器的定义是:将一个类的接口转换成客户希望的另外一个接口。适配器模式使得原本由于接口不兼容而不能一起工作的那些类可以一起工作。

换成本章的话说:调用处希望的接口是 Tracker,三家 SDK 的接口都不是它,我们又改不了 SDK,就在中间放一个适配器,一头实现 Tracker,一头调 SDK。所谓“转换”,既包括方法名和参数(track 变成 get_trace(bill_no, "zh-CN")),也包括返回值(列表的第 0 条、数字码、毫秒变成 TrackingInfo)。

本章的适配器是持有一个 SDK 对象,再把调用转过去,叫对象适配器(object adapter),用的是组合。还有一种类适配器(class adapter):适配器同时继承目标接口和被适配的类,用多重继承把两边接起来。Python 支持多重继承,可以这样写;TS 的类只能继承一个类,写不了。对象适配器更常用:它持有的对象可以换成 SDK 的任何子类或测试替身,也不把 SDK 自己的方法暴露给调用方。本书只用这种。

适配器的结构:Client 只认 Target,Adapter 实现 Target 并转调 Adaptee

对照 9.4 的案例图:Tracker 是 Target,三家快递的 SDK 是 Adaptee,SfAdapter 等三个类是 Adapter。

角色 在本章案例里是谁 职责
目标(Target) Tracker 调用方期望的接口,这里是 track(order_no)
被适配者(Adaptee) SfClient、ZtoApi、JdClient 已经存在、接口对不上的类,通常改不了
适配器(Adapter) SfAdapter、ZtoAdapter、JdAdapter 实现目标接口,持有被适配者,把调用和返回值翻译过去、翻译回来
客户(Client) order_detail_page、support_lookup、support_batch_lookup 只通过目标接口干活

常见场景:

  • 接第三方 SDK。支付、短信、物流、对象存储,每家的接口都不一样,先定一个自己的接口,每家一个适配器。
  • 老系统接口迁移。新代码按新接口写,旧模块暂时不动,用一个适配器把旧接口包成新接口,等旧模块重写完再拿掉。
  • 让一个类适配框架要求的接口。框架要一个有 read() 的对象、一个按某种格式调用的函数,你手里的类不是这样,写一层适配。9.6 的两个例子都是这一类。
  • 测试替身。像 9.4 的 FakeTracker,假实现也是 Tracker 的一个实现,调用方分不出来。

例一:Python 的 io.TextIOWrapper(Python)

Section titled “例一:Python 的 io.TextIOWrapper(Python)”

open("orders.csv") 返回的对象可以 readline()、可以 for line in f,拿到的是字符串。但操作系统给的只有字节。Python 的 io 模块分了层:底下是只会读字节的二进制流(open 默认用的是 BufferedReader),上面套一个 io.TextIOWrapper,把“读字节”适配成“按行读字符串”。你可以自己验证:open(path).buffer 就是底下那个 BufferedReader。适配时它要做两件事:按 encoding 把字节解码成字符;在默认的通用换行模式下,把 \r\n 和 \r 都统一成 \n。简化版:

class BytesStream: # 被适配者:大致相当于 BufferedReader,只会按字节读
def __init__(self, data: bytes) -> None:
self.data = data
self.pos = 0
def read_bytes(self, n: int) -> bytes:
chunk = self.data[self.pos:self.pos + n]
self.pos += len(chunk)
return chunk
class TextWrapper: # 适配器:大致相当于 io.TextIOWrapper,对外是按行读字符串
def __init__(self, raw: BytesStream, encoding: str = "utf-8") -> None:
self.raw = raw
self.encoding = encoding
self.pending = b"" # 读进来还没交出去的字节
def readline(self) -> str:
while b"\n" not in self.pending:
chunk = self.raw.read_bytes(8) # 真实实现一次读一大块,这里读 8 字节好看出拼接
if not chunk:
break
self.pending += chunk
line, sep, self.pending = self.pending.partition(b"\n")
text = (line + sep).decode(self.encoding) # 编码:字节 → 字符串
return text.replace("\r\n", "\n") # 换行:只演示 \r\n 统一成 \n
if __name__ == "__main__":
raw = BytesStream("订单号,状态\r\nA1001,运输中\n".encode("utf-8"))
f = TextWrapper(raw)
print(repr(f.readline()))
print(repr(f.readline()))
print(repr(f.readline()))
# 输出:
# '订单号,状态\n'
# 'A1001,运输中\n'
# ''

BytesStream 对应本章的 ZtoApi,TextWrapper 对应 ZtoAdapter,readline 就是 track:调用方只想要“下一行字符串”,至于底下一次读几个字节、一个汉字的三个字节被切在两次读取之间怎么拼回来、换行符是哪种,都由适配器处理。真实的 TextIOWrapper 用的是增量解码器,还支持写、seek 和更多换行模式,这里都省略了。

例二:Node 的 util.promisify(TypeScript)

Section titled “例二:Node 的 util.promisify(TypeScript)”

Node 早期的 API 是回调风格:最后一个参数是一个 (err, result) => void 函数,比如 fs.readFile(path, callback)。后来大家都写 async/await,要的是返回 Promise 的函数。Node 在 util 模块里提供了 promisify:传进去一个回调风格的函数,返回一个参数相同、但返回 Promise 的新函数,常见写法是 const readFile = promisify(fs.readFile)。简化版只处理一个参数:

type Callback<T> = (err: Error | null, value?: T) => void;
const legacyConfig = { // 被适配者:老的回调风格 API
readConfig(path: string, callback: Callback<string>): void {
setTimeout(() => {
if (path.endsWith(".json")) callback(null, `{"port":8080}`);
else callback(new Error(`不支持的格式:${path}`));
}, 0);
},
};
// 适配器:大致相当于 util.promisify(真实版本支持任意个参数,还认 util.promisify.custom)
function promisify<A, T>(fn: (arg: A, callback: Callback<T>) => void): (arg: A) => Promise<T> {
return (arg: A) =>
new Promise<T>((resolve, reject) => {
fn(arg, (err, value) => {
if (err) reject(err); // 回调的 err → Promise 的 reject
else resolve(value as T); // 回调的结果 → Promise 的 resolve
});
});
}
const readConfig = promisify(legacyConfig.readConfig);
async function main(): Promise<void> {
console.log(`读到配置:${await readConfig("app.json")}`);
try {
await readConfig("app.yaml");
} catch (error) {
console.log(`读取失败:${(error as Error).message}`);
}
}
void main();
// 输出:
// 读到配置:{"port":8080}
// 读取失败:不支持的格式:app.yaml

这是一个函数级的适配器:被适配的不是一个类,而是一个函数的调用约定。目标接口是“返回 Promise”,被适配者是“最后一个参数是回调”,promisify 在中间把 err 翻译成 reject、把结果翻译成 resolve,和 ZtoAdapter 把数字码翻译成中文状态是同一件事。它还说明适配器不一定是一个类:在 Python 和 TS 里,函数可以当值传来传去,一个返回函数的函数就能做适配。

  • 接口只差一个方法名,两边又都是自己的代码。比如你写的 OrderRepo.fetch 和同事写的接口要求的 get 对不上,直接改名,或者在类里加一个 get 方法转调就好。为此单独写一个适配器类,读代码的人要多跳一层,才发现它只是换了个名字。
  • 适配器开始塞业务逻辑。比如有人在 JdAdapter.track 里加上“大件订单超过三天未签收就发优惠券”。适配器只该做翻译;业务规则塞进来以后,换一家快递时这条规则会跟着旧适配器一起消失,顺丰、中通的大件也没有这条规则,没人会想到去适配器里找它。
  • SDK 只在一个地方用一次。比如只有一个后台脚本调了一次某家的对账接口,在那里直接调 SDK、就地整理字段就行。为它定一个接口、写一个适配器,多出两个类,却没有第二个调用处能从中受益。

适配器、装饰器、外观、桥接在代码上都是“一个对象包着或连着另一个对象”,区别在为什么包、接口变不变:

模式 意图 对外的接口变不变 什么时候引入
适配器(Adapter) 让一个接口对不上的类按调用方要的接口工作 变:对外换成目标接口 事后补救:两边都已经写好,接不上
装饰器(Decorator) 不改原对象,给它叠加功能 不变:和被包的对象是同一个接口 想给已有对象加日志、缓存、重试时,见第 10 章
外观(Facade) 给一组子系统一个简单入口 新定义一个更简单的接口,背后是好几个对象 调用方每次都要按顺序调好几个子系统时,见第 11 章
桥接(Bridge) 把两个独立变化的维度拆开再组合 不为对接而转换:抽象和实现两侧的接口都是事先设计的 设计之初就知道有两个维度要各自扩展,见第 14 章

一句话区分:适配器换接口,装饰器加功能,外观简化入口,桥接是事先拆开、不是事后补救。

公司有一个老日志组件 LegacyLogger,只有一个方法 write(level, msg),level 是整数:10 表示普通信息,20 表示警告,30 表示错误。新写的代码统一用一个 Logger 接口,它有三个方法:info(msg)、warn(msg)、error(msg)。老组件被好几个系统用着,不能改。

请用适配器实现:

  • 写 Logger 接口(Python 用 Protocol,TS 用 interface)和适配器 LegacyLoggerAdapter,持有一个 LegacyLogger,把三个方法翻译成 write 加对应的级别。
  • 写一个业务函数 place_order(logger, order_no, stock):开始时记一条 info;库存为 0 记一条 error 并返回;库存少于 5 记一条 warn;最后记一条“已下单”的 info。
  • 再写一个 MemoryLogger,同样实现 Logger,只把日志存进一个列表,用它当测试替身,调用一次 place_order 后打印列表里的内容。Python 和 TS 各写一版。

提示:place_order 里不应该出现 10、20、30 这些数字。做完想一想:要是以后换成另一个级别是字符串 "INFO" 的日志库,你要动哪几个地方?

  • 适配器在调用方和接口对不上的类之间加一层翻译:一头实现调用方要的接口,一头持有并调用被适配的对象。
  • 加一家新的第三方,要写的翻译一样不少,映射表也要加一行;收益是这家的怪癖只写在一个适配器里,调用处只认一个接口,调用处再多也不用抄。
  • 本书用对象适配器(组合);类适配器靠多重继承,Python 能写,TS 写不了。
  • 适配器只做翻译,不放业务规则;接口只差一个名字的自家代码直接改,只用一次的 SDK 直接调。
  • 和装饰器、外观、桥接的区别在意图:适配器换接口,装饰器不换接口加功能,外观给一组对象一个简单入口,桥接是事先设计的拆分。