第三章练习答案
预算结算
放在正文 Budget 类之后,单独运行:
python
b = Budget(30)
b.reserve('x', 20)
try:
b.settle('x', 25)
except ValueError as error:
assert str(error) == 'invalid usage'
else:
raise AssertionError('overrun accepted')
assert b.remaining == 10 and b.reserved == {'x': 20}
b.settle('x', 10)
assert b.remaining == 20 and b.reserved == {}
print('failed settlement preserved')结果 failed settlement preserved。检验在删除预留之前,因此错误结算不破坏状态。若用量未知时全额退回,会把可能已经发生的花费重新分配给其他请求,突破任务预算。
旧事件读取
以下是可独立保存运行的完整程序:
python
from copy import deepcopy
def read_answer(event):
version = event.get('version')
if type(version) is not int or version not in (1, 2):
raise ValueError('unsupported version')
field = 'text' if version == 1 else 'answer'
if not isinstance(event.get(field), str):
raise ValueError('invalid answer')
return event[field]
old = {'version': 1, 'text': ''}
original = deepcopy(old)
assert read_answer(old) == '' and old == original
assert read_answer({'version': 2, 'answer': '完成'}) == '完成'
for item, reason in [
({'version': 3, 'answer': '完成'}, 'unsupported version'),
({'version': 2, 'answer': 7}, 'invalid answer'),
({'version': True, 'text': '完成'}, 'unsupported version'),
]:
try:
read_answer(item)
except ValueError as error:
assert str(error) == reason
else:
raise AssertionError('bad event accepted')
print('old and new readable')结果 old and new readable。Python 的 bool 是 int 子类,精确检查类型避免把 True 当版本1。兼容读取器让未迁移历史事件和在途旧任务仍能恢复;等到这些消费者的生命周期结束,才评估移除旧分支。
总截止时间变体
保存 answer_deadline.py,独立Python3.11+运行;返回的0.5秒是退避之后的剩余时间,实际等待完应再读取时钟。
python
def remaining_timeout(total,queued,backoff,minimum):
remaining=total-queued-backoff
return remaining if remaining>=minimum else None
assert remaining_timeout(3,1,1.5,.6) is None
assert remaining_timeout(3,1,1.5,.4)==.5
assert remaining_timeout(3,3,0,.4) is None
print('deadline variants passed')3−1−1.5=0.5,无法容纳0.6秒的最小调用窗口,因此第一种情况不要启动请求。队列等待和退避都属于用户看到的总时延。