如何在Twisted Web客户端中使用ThrottlingFactory模拟高延迟

ThrottlingFactory是Twisted内置的传输层带宽节流组件，原示例代码不生效的核心原因：默认初始化的Agent会使用内置HTTP连接池自动创建TCP连接，整个流程没有接入节流工厂的逻辑，限速、延迟规则自然不会作用到请求连接上。

接入的核心逻辑是自定义HTTP连接池，重写连接创建流程，将HTTP客户端的原生协议工厂用ThrottlingFactory包装后再建立TCP连接；如果需要模拟固定链路高延迟，可以在节流工厂基础上叠加协议层的延迟转发逻辑。

可直接运行的修改后代码 #

from sys import argv
from pprint import pformat

from twisted.internet.task import react
from twisted.internet.endpoints import HostnameEndpoint
from twisted.protocols.policies import ThrottlingFactory, ProtocolWrapper
from twisted.web.client import Agent, readBody, HTTPConnectionPool
from twisted.web.http_headers import Headers


# 固定链路延迟模拟包装，不需要可直接删除
class DelayedProtocolWrapper(ProtocolWrapper):
    def __init__(self, factory, wrappedProtocol, delay=0.5):
        super().__init__(factory, wrappedProtocol)
        self.delay = delay

    def dataReceived(self, data):
        self.reactor.callLater(self.delay, super().dataReceived, data)

    def write(self, data):
        self.reactor.callLater(self.delay, super().write, data)


class DelayedThrottlingFactory(ThrottlingFactory):
    def __init__(self, wrappedFactory, readLimit, writeLimit, delay=0):
        super().__init__(wrappedFactory, readLimit=readLimit, writeLimit=writeLimit)
        self.delay = delay

    def buildProtocol(self, addr):
        wrappedProto = super().buildProtocol(addr)
        if self.delay > 0:
            return DelayedProtocolWrapper(self, wrappedProto, self.delay)
        return wrappedProto


class ThrottledHTTPConnectionPool(HTTPConnectionPool):
    def __init__(self, reactor, readLimit=0, writeLimit=0, delay=0, persistent=True):
        super().__init__(reactor, persistent=persistent)
        self.readLimit = readLimit
        self.writeLimit = writeLimit
        self.delay = delay

    def _getEndpoint(self, uri):
        endpoint = HostnameEndpoint(
            self._reactor,
            uri.host.decode("ascii"),
            uri.port
        )
        original_connect = endpoint.connect
        def throttled_connect(protocolFactory):
            throttled_factory = DelayedThrottlingFactory(
                protocolFactory,
                readLimit=self.readLimit,
                writeLimit=self.writeLimit,
                delay=self.delay
            )
            return original_connect(throttled_factory)
        endpoint.connect = throttled_connect
        return endpoint


def cbRequest(response):
    print("Response version:", response.version)
    print("Response code:", response.code)
    print("Response phrase:", response.phrase)
    print("Response headers:")
    print(pformat(list(response.headers.getAllRawHeaders())))
    d = readBody(response)
    d.addCallback(cbBody)
    return d


def cbBody(body):
    print("Response body:")
    print(body)


def main(reactor, url=b"http://httpbin.org/get"):
    # 自定义弱网参数：读限速10KB/s、写限速10KB/s、双向延迟500ms
    pool = ThrottledHTTPConnectionPool(
        reactor,
        readLimit=1024 * 10,
        writeLimit=1024 * 10,
        delay=0.5
    )
    agent = Agent(reactor, pool=pool)
    d = agent.request(
        b"GET", url, Headers({"User-Agent": ["Twisted Web Client Example"]}), None
    )
    d.addCallback(cbRequest)
    return d


react(main, argv[1:])

参数说明 #

• readLimit/writeLimit：单位为字节/秒，设为0时代表不限速，可根据测试场景调整带宽上限，例如传入1024*50即对应50KB/s的低带宽网络。
• delay：单位为秒，对应双向链路的固定传播延迟，设为0时不添加额外延迟，仅保留带宽节流效果。
• 若仅需要带宽限制不需要模拟固定延迟，直接删除DelayedProtocolWrapper、DelayedThrottlingFactory相关代码，在连接池的throttled_connect方法中直接使用原生ThrottlingFactory包装协议工厂即可。

内容的提问来源于stack exchange，提问作者Joe