2026 Stripe OA 独家面经复盘与 Python 实战：负载均衡系统设计

Mar 4

Stripe 的在线笔试（OA）向来以考察“真实业务场景的系统设计与代码落地”闻名。今天我们将深度剖析一道新鲜出炉的 Stripe NG OA 考题——服务器负载均衡系统设计。

本文将为你拆解这道 60 分钟限时的高频考题，带你用资深工程师的视角构建最优数据结构，并提供满分 Python 代码实现。

一、2026 真人上岸案例：Z同学的 Stripe 突围战

就在 2026 年初，面临极度内卷的北美求职季，计算机专业的应届生 Z同学在接连挂掉两家大厂的系统设计轮后，找到了我们寻求专业的面试准备与面试培训支持。

通过我们的专家级面试辅助，Z同学深刻理解了面向对象设计（OOD）的核心模式和高级数据结构（如堆、哈希表的组合应用）。在遇到这道 Stripe 负载均衡器 OA 时，别人还在纠结如何处理 Part 3 的亲和性路由，他仅用 45 分钟就写出了结构极其优雅的完整代码，并在后续的 Virtual Onsite 中对答如流，最终在毕业前顺利拿到 Stripe 的高薪 Offer 成功上岸。

如果你也在找工作的泥潭中挣扎，不要单打独斗，专业的事交给专业的团队。

🔥 点击这里预约资深专家 1v1 面试辅导 🔥

二、面试真题深度还原

题目背景：设计一个服务器负载均衡系统，记录每个 CONNECT 请求被路由到了哪个 server。 限制时间：60分钟（需开启摄像头） 基本输入：numServer 服务总数，MaxConnection 每个服务器支持的最大连接数，requests[n] 请求列表。

题目分为 5 个循序渐进的 Parts：

Part 1：基础路由 处理 CONNECT 请求（包含 connect id, user id, obj id）。分配规则：优先选择当前连接数最小的服务器；若有平局，选择编号（下标）最小的服务器。
Part 2：断开连接 处理 DISCONNECT 请求，根据 connect id 移除相应的连接并释放服务器容量。
Part 3：路由亲和性（Affinity） 优化分配逻辑：优先把相同 obj id 的请求分配到曾经处理过该 obj id 的服务器上（同样遵循容量和最小连接数规则）。
Part 4：容量限制 引入 MaxConnection。如果最优服务器已满，请求应当被拒绝，且拒绝的连接不记录在系统 log 中。
Part 5：服务器宕机/下线 添加 SHUTDOWN 请求。关闭指定编号的服务器，并需要将该服务器上现存的所有活跃连接按照上述规则，重新分配到其他健康的服务器上。

三、核心难点与解题思路剖析

Stripe 这种循序渐进的题目，最忌讳“面向过程”写面条代码。你需要提前预见到后续需求，做好系统设置和状态封装。

高效的查找与更新：使用多个哈希表维护状态。

conn_map: connect_id -> (server_id, obj_id)，用于 O(1) 处理 Disconnect。
server_conns: server_id -> set(connect_ids)，用于管理每个 Server 上的具体连接。
obj_server_map: obj_id -> set(server_ids)，用于处理 Part 3 的亲和性匹配。

Top-K / 最小值查找：虽然可以用 Priority Queue（Min-Heap）来寻找最小连接数的服务器，但考虑到这道题有 SHUTDOWN 和状态的频繁变动，且 numServer 通常在合理范围内，使用数组/集合配合一次遍历或内置排序（在同等数量级下）可以写出更不容易出 Bug 的代码。
SHUTDOWN 的状态转移：在处理下线时，必须先将服务器标记为不可用，然后将其身上的现有连接暂时“缓存”并从旧状态中抹除，最后再重新调用 connect 逻辑进行二次分配，这样可以完美复用现有的路由逻辑，避免代码冗余。

四、Python 满分参考代码

下面是遵循上述思路、采用面向对象封装的 Python 优雅实现。注意代码的缩进和模块化拆分。

from collections import defaultdict

class LoadBalancer:
    def __init__(self, num_servers, max_connection):
        self.max_connection = max_connection
        # 记录健康的活跃服务器池
        self.active_servers = set(range(1, num_servers + 1))
        
        # server_id -> 当前挂载的连接集合
        self.server_conns = {i: set() for i in range(1, num_servers + 1)}
        
        # connect_id -> (server_id, obj_id)，用于快速断开连接
        self.conn_map = {}
        
        # obj_id -> set(server_id)，用于亲和性路由 (Part 3)
        self.obj_server_map = defaultdict(set)
        
    def _find_best_server(self, obj_id):
        # 1. 优先在处理过相同 obj_id 的健康服务器中寻找
        affinity_candidates = []
        if obj_id in self.obj_server_map:
            for s_id in self.obj_server_map[obj_id]:
                if s_id in self.active_servers and len(self.server_conns[s_id]) < self.max_connection:
                    affinity_candidates.append((len(self.server_conns[s_id]), s_id))
                    
        if affinity_candidates:
            # 同样连接数的情况下，选择 server_id 最小的
            affinity_candidates.sort()
            return affinity_candidates[0][1]
            
        # 2. 如果没有亲和性匹配，全局寻找当前连接数最小的健康服务器
        best_server = None
        min_conns = float('inf')
        
        # 保证按照 server_id 从小到大遍历，自然满足平局时选编号小的规则
        for s_id in sorted(self.active_servers):
            conns = len(self.server_conns[s_id])
            if conns < self.max_connection and conns < min_conns:
                min_conns = conns
                best_server = s_id
                
        return best_server

    def connect(self, conn_id, user_id, obj_id):
        server_id = self._find_best_server(obj_id)
        
        # Part 4: 如果没找到合适的或者全部爆满，拒绝连接
        if not server_id:
            return None 
            
        # 注册连接状态
        self.server_conns[server_id].add(conn_id)
        self.conn_map[conn_id] = (server_id, obj_id)
        self.obj_server_map[obj_id].add(server_id)
        
        return server_id

    def disconnect(self, conn_id):
        if conn_id not in self.conn_map:
            return
            
        server_id, obj_id = self.conn_map[conn_id]
        self.server_conns[server_id].remove(conn_id)
        del self.conn_map[conn_id]

    def shutdown(self, server_id):
        if server_id not in self.active_servers:
            return
            
        # 从健康池中摘除
        self.active_servers.remove(server_id)
        
        # 取出该服务器上的所有存活连接
        conns_to_reassign = list(self.server_conns[server_id])
        self.server_conns[server_id].clear()
        
        reassign_log = []
        # 遍历重新分配
        for conn_id in conns_to_reassign:
            _, obj_id = self.conn_map[conn_id]
            # 清理旧映射，避免状态污染
            del self.conn_map[conn_id] 
            
            # 复用基础路由逻辑重新分配
            new_server = self.connect(conn_id, None, obj_id)
            if new_server:
                reassign_log.append((conn_id, new_server))
                
        return reassign_log

五、面试救急：获取专属通关方案

代码写出来了，但你确定在 60 分钟高压环境、并且有摄像头监控的条件下，能一次性写出 0 Bug 的代码并完美应对刁钻的 Edge Cases 吗？

如果你正面临即将到来的 OA 或者 Onsite，如果你渴望在竞争激烈的环境里稳稳上岸，我们为你提供顶级硅谷专家团队的硬核支援：

🚀 全面系统化面试培训：精讲系统设计、代码架构与高频考点。
💡 高端定制化面试辅助：攻克薄弱环节，针对性演练。
🛡️ 专业面试代面 / 面试代考（特定情况）：化解时间冲突与高难度技术壁垒，为你的求职保驾护航。
🎯 业内顶尖面试枪手资源：全栈、后端、大数据、AI 多领域覆盖。

别让一次准备不足的 OA 毁掉你的高薪梦！机会永远留给准备最充分的人。

👉 立即点击联系专家顾问，定制你的独家高薪护航方案！ 👈

温馨提示：各大厂题库仍在极速更新，提前储备技术护城河才是王道。祝各位中国开发者都能在全球科技浪潮中斩获心仪的 Offer！

Gabby Fox