SQS轮询压力管控：自适应轮询器背压决策实现问询

说明：本文围绕自适应扩展的SQS轮询器/订阅者展开，以抽象化表述探讨背压管理与中介模式。

为便于理解，代码中省略了锁、临界区、信号量及try/finally块，相关实现逻辑已充分考虑。

假设存在一个内部队列、消费者及轮询器：轮询器对内部队列而言是生产者，实际是消费者与外部SQS源之间的中介。

# smaller internal queue is better, 
# as all those items in queue are in kind of limbo
queue = Queue(32) 

def consume():
    while True:
        # this may block if queue is empty
        item = queue.dequeue()
        consume(item)

def poll():
    running += 1
    while True:
        if running > expected:
            break
        items = poll_batch()
        for item in items:
            # this may block if queue is full
            queue.enqueue(item) 
    running -= 1

poll_batch存在天然限制：每次最多返回10条数据，且存在网络开销。若请求往返耗时100ms，则天然上限为每秒100条数据，无法突破。为此可通过新增或停止轮询循环来适配速率。

示例supervisor逻辑如下：

def supervisor():
    while True:
        sleep(1)
        decision = do_i_need_more_or_less() # -1, 0, 1
        expected = running + decision
        if expected > running:
            new_thread(poll)

核心技术问询 #

如何实现do_i_need_more_or_less()函数？该函数依赖哪些数据，需采集哪些数据以做出扩缩容决策？

理论上需平衡生产与消费速率，但实际存在诸多难点：小内部队列下难以通过队列长度判断速率趋势；需测量「潜在消费速率」，但提升生产速率会导致轮询器阻塞，且需精准识别阻塞状态及差异。

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