引言

在某个自动化测试项目中，需要缓存查询设备返回的响应以备后续查询，吞吐量为每秒50个左右，并且有多个不同的请求需要对其响应加以区分。

方案

缓存响应的话，肯定得考虑缓存的方式，一种是远端缓存，比如redis，一种是本地缓存，而本地缓存也可以细分为内存与本地持久化两种。 因为对写入和查询的时延要求较高，且只需要在程序运行时可以写入与查询即可，无需持久化，故我选择直接在本地内存中缓存响应信息。（redis属于外部依赖，会增加程序复杂性，并且读写效率不及本地内存，而本地持久化，读写相对耗时，且不具备必要性）

具体设计

缓存的数据容器

缓存的容器，我选择dict类型，而key值为请求的头部信息，value为一个列表或者队列，用于存储过往的响应

缓存内容的类型选择

value的类型，我比较了一下，在元素个数固定，并且需要频繁对头部和尾部进行删除和写入的情况下，deque明显比list性能更优，所以我最终决定使用deque。

缓存的个数限制

需要对value存储的响应个数上限进行限制，达到上限即清除时间最久的那个响应，否则占用内存会不断膨胀，并且影响程序读写性能。

记录响应时间

又因为需要考虑响应的即时性，需要对响应的写入时间进行记录，否则无法判断，响应是否最新写入的，在自动化测试查询响应并加以判断时，会产生误判，所以，deque的元素类型设定为一个tuple，tuple的元素为响应的值与响应的写入时间。

多线程安全性

考虑到高并发场景，需要对读写进行加锁。

是否封装为响应缓存类

本身功能不是那么复杂的情况下，封装为响应缓存类会存在过度封装的问题，并且性能上也可能会有影响，所以暂不进行封装。

代码实现

from collections import deque
from datetime import datetime
from decimal import Decimal
from threading import Lock
from typing import Optional, Deque, Tuple
from settings import test_logger

# 全局锁，用于保护共享资源rsp_dict的并发访问
lock = Lock()
# 缓存响应的个数
rsp_cache_size = 50

# 共享响应字典，用于存储键值对及其时间戳
rsp_dict: dict[str, Deque[Tuple[str, str]]] = {}

time_fmt = "%Y-%m-%d %H:%M:%S.%f"


# @timeit
def set_rsp(k: str, rsp: str) -> None:
    """
    设置响应值到缓存中，如果超过最大缓存大小则移除最早的数据。

    :param k: 键
    :param rsp: 响应字符串
    :return:
    """
    # 减少锁的竞争: 确保在锁内部的操作尽可能快，考虑是否可以将一些不直接影响数据一致性的逻辑移出锁块。
    ts = get_current_time()
    with lock:
        current = rsp_dict.setdefault(k, deque(maxlen=rsp_cache_size))
        current.append((rsp, ts))
        test_logger.info(F"set rsp, K: {k}, v: {rsp}, timestamp: {ts}, rsp cache size: {len(current)}")


def get_current_time() -> str:
    """
    获取当前时间
    :return:
    """
    return datetime.now()


def get_current_time_str() -> str:
    """
    获取当前时间的格式化时间戳字符串
    :return:
    """
    return datetime.now().strftime(time_fmt)


# @timeit
def get_rsp(k: str):
    """
    获取指定键的最新响应及时间戳。

    :param k: 键
    :return: 最新响应与时间戳元组，如果键不存在则返回(None, None)
    """
    with lock:
        record = rsp_dict.get(k, deque())
        rsp, ts = record[-1] if record else (None, None)
        test_logger.info(F"get rsp,K: {k}, v: {rsp}, timestamp: {ts}")
        return rsp, ts


# @timeit
def clacc_time_diff(t1: datetime, t2: datetime) -> Decimal:
    """
    计算两个时间戳之间的时间差，单位为秒，精确到小数点后3位。
    """
    ts_cost = (t2 - t1).total_seconds()
    ts_cost = Decimal(ts_cost).quantize(Decimal("0.000"))
    return ts_cost


def get_rsp_with_timeout(k: str, timeout: float = 0.1) -> Optional[str]:
    """
    获取指定键的最新响应，如果响应时间超过指定超时时间则返回None。
    :param k:
    :param timeout:
    :return:
    """
    rsp, ts = get_rsp(k)
    if ts:
        if clacc_time_diff(get_current_time(), ts) <= timeout:
            test_logger.info(F"get rsp,K: {k}, v: {rsp}, timestamp: {ts}")
            return rsp
    else:
        return None

后续扩展

这只是简单实现了一个本地的响应缓存模块，可内嵌在测试框架内。后续如果随着测试项目的复杂化与分布式执行，也可以考虑将该模块加以扩展，响应缓存在远端数据库，并封装为工具类，支持不同的缓存策略。