Stripe 2026 New Grad OA 是典型的"系统状态模拟题"——不考算法 trick，但考你能否同时维护 4-5 个互相依赖的状态：服务器连接数、连接 ID 与服务器的映射、objId 与服务器的亲和绑定、容量上限、SHUTDOWN 后连接的重路由。题目分 5 个递进 Part，每一 Part 都建立在前一 Part 的代码之上。本文给出 5 个 Part 的完整 Python 解法，并标注 HackerRank 隐藏测试常踩点。
Stripe NG OA 概览
| 维度 | 详情 |
|---|---|
| 平台 | HackerRank（全程录屏 + 摄像头） |
| 时长 | 60 分钟 |
| 题量 | 1 道大题，5 个 Part |
| 难度 | LeetCode Easy ~ Medium，重逻辑 |
| 通过线 | 5/5 隐藏 case 全绿 |
| 语言 | Python / Java / C++ 自选 |
题目核心：Server Load Balancing System
输入：
numServer：服务器总数maxConnection：每台服务器最大连接数requests：请求列表，每条请求是["CONNECT", connectId, userId, objId]/["DISCONNECT", connectId, userId, objId]/["SHUTDOWN", serverId]
输出：所有被接受的连接日志（按发生顺序）
题目要求模拟服务器在以下规则下的负载均衡：
Part 1：纯 CONNECT 路由
规则：每个 CONNECT 请求路由到当前连接数最少的服务器；连接数相同则选 index 较小的。
class LoadBalancer:
def __init__(self, num_server: int, max_connection: int):
self.num_server = num_server
self.max_connection = max_connection
self.counts = [0] * num_server
self.conn_to_server = {}
self.log = []
def _pick_server(self):
min_idx = 0
for i in range(1, self.num_server):
if self.counts[i] < self.counts[min_idx]:
min_idx = i
return min_idx
def connect(self, connect_id, user_id, obj_id):
s = self._pick_server()
self.counts[s] += 1
self.conn_to_server[connect_id] = s
self.log.append(("CONNECT", connect_id, s))
Part 1 核心：用 list 维护 counts，避免堆——后续 Part 还要按 objId / serverId 索引，堆不灵活。
Part 2：加上 DISCONNECT
规则：DISCONNECT 释放对应 connectId 占用的服务器槽位。
def disconnect(self, connect_id, user_id, obj_id):
if connect_id not in self.conn_to_server:
return
s = self.conn_to_server.pop(connect_id)
self.counts[s] -= 1
self.log.append(("DISCONNECT", connect_id, s))
隐藏 case：DISCONNECT 不存在的 connectId（之前没 CONNECT 过 / 已被 SHUTDOWN 清掉）时，静默忽略而不是报错。
Part 3：按 objId 亲和路由
规则：相同 objId 的连接应尽量路由到同一台服务器。
def __init__(self, ...):
...
self.obj_to_server = {} # objId -> serverId
self.obj_count = defaultdict(int) # objId 当前活动连接数
def connect(self, connect_id, user_id, obj_id):
if obj_id in self.obj_to_server:
s = self.obj_to_server[obj_id]
else:
s = self._pick_server()
self.obj_to_server[obj_id] = s
self.counts[s] += 1
self.obj_count[obj_id] += 1
self.conn_to_server[connect_id] = s
self.conn_to_obj[connect_id] = obj_id
self.log.append(("CONNECT", connect_id, s))
def disconnect(self, connect_id, user_id, obj_id):
if connect_id not in self.conn_to_server:
return
s = self.conn_to_server.pop(connect_id)
o = self.conn_to_obj.pop(connect_id)
self.counts[s] -= 1
self.obj_count[o] -= 1
if self.obj_count[o] == 0:
del self.obj_to_server[o]
self.log.append(("DISCONNECT", connect_id, s))
Part 3 关键：obj_count == 0 时移除 obj_to_server 映射——否则同一 objId 后续重连会卡在已被 SHUTDOWN 的服务器上（Part 5 会爆）。
Part 4：加上容量上限
规则：服务器达到 maxConnection 后拒绝新 CONNECT。
def connect(self, connect_id, user_id, obj_id):
if obj_id in self.obj_to_server:
s = self.obj_to_server[obj_id]
if self.counts[s] >= self.max_connection:
self.log.append(("REJECT", connect_id))
return
else:
s = self._pick_available_server()
if s is None:
self.log.append(("REJECT", connect_id))
return
self.obj_to_server[obj_id] = s
...
def _pick_available_server(self):
candidates = [i for i in range(self.num_server)
if self.counts[i] < self.max_connection]
if not candidates:
return None
return min(candidates, key=lambda i: (self.counts[i], i))
Part 4 决策点：当绑定的 objId server 满了时，是否回退到其他 server？题目期望不回退，直接 REJECT——这是 Stripe 的"affinity 优先级 > 完成率"业务直觉。
Part 5：SHUTDOWN 重路由
规则：["SHUTDOWN", serverId] 时：
- 该 server 上所有连接作废（不算 DISCONNECT）
- 该 server 上的 objId 绑定释放
- 后续 CONNECT 请求绑定到其他 server
def shutdown(self, server_id):
self.counts[server_id] = 0
objs_to_release = [
obj for obj, srv in self.obj_to_server.items() if srv == server_id
]
for obj in objs_to_release:
del self.obj_to_server[obj]
self.obj_count[obj] = 0
conns_to_drop = [
cid for cid, srv in self.conn_to_server.items() if srv == server_id
]
for cid in conns_to_drop:
del self.conn_to_server[cid]
if cid in self.conn_to_obj:
del self.conn_to_obj[cid]
# 标记为下线（防止后续 _pick_available_server 选中）
self.shut_down.add(server_id)
self.log.append(("SHUTDOWN", server_id))
def _pick_available_server(self):
candidates = [
i for i in range(self.num_server)
if i not in self.shut_down and self.counts[i] < self.max_connection
]
if not candidates:
return None
return min(candidates, key=lambda i: (self.counts[i], i))
Part 5 三个隐藏 case：
- SHUTDOWN 后该 server 的 objId 绑定要全部清掉（否则下次该 obj 再 CONNECT 会路由到下线 server）
- SHUTDOWN 不能影响 log 里的历史 CONNECT 记录
- SHUTDOWN 一个已经下线的 server 应当幂等（不报错）
完整代码（5 个 Part 整合）
from collections import defaultdict
from typing import List
class LoadBalancer:
def __init__(self, num_server: int, max_connection: int):
self.num_server = num_server
self.max_connection = max_connection
self.counts = [0] * num_server
self.shut_down = set()
self.conn_to_server = {}
self.conn_to_obj = {}
self.obj_to_server = {}
self.obj_count = defaultdict(int)
self.log = []
def _pick_available_server(self):
candidates = [
i for i in range(self.num_server)
if i not in self.shut_down and self.counts[i] < self.max_connection
]
if not candidates:
return None
return min(candidates, key=lambda i: (self.counts[i], i))
def connect(self, connect_id, user_id, obj_id):
if obj_id in self.obj_to_server:
s = self.obj_to_server[obj_id]
if s in self.shut_down or self.counts[s] >= self.max_connection:
self.log.append(("REJECT", connect_id))
return
else:
s = self._pick_available_server()
if s is None:
self.log.append(("REJECT", connect_id))
return
self.obj_to_server[obj_id] = s
self.counts[s] += 1
self.obj_count[obj_id] += 1
self.conn_to_server[connect_id] = s
self.conn_to_obj[connect_id] = obj_id
self.log.append(("CONNECT", connect_id, s))
def disconnect(self, connect_id, user_id, obj_id):
if connect_id not in self.conn_to_server:
return
s = self.conn_to_server.pop(connect_id)
o = self.conn_to_obj.pop(connect_id)
self.counts[s] -= 1
self.obj_count[o] -= 1
if self.obj_count[o] == 0 and o in self.obj_to_server:
del self.obj_to_server[o]
self.log.append(("DISCONNECT", connect_id, s))
def shutdown(self, server_id):
if server_id in self.shut_down:
return
self.shut_down.add(server_id)
self.counts[server_id] = 0
for obj in [o for o, srv in self.obj_to_server.items() if srv == server_id]:
del self.obj_to_server[obj]
self.obj_count[obj] = 0
for cid in [c for c, srv in self.conn_to_server.items() if srv == server_id]:
del self.conn_to_server[cid]
if cid in self.conn_to_obj:
del self.conn_to_obj[cid]
self.log.append(("SHUTDOWN", server_id))
def run(self, requests: List[List]):
for req in requests:
op = req[0]
if op == "CONNECT":
self.connect(req[1], req[2], req[3])
elif op == "DISCONNECT":
self.disconnect(req[1], req[2], req[3])
elif op == "SHUTDOWN":
self.shutdown(req[1])
return self.log
时间复杂度：每次操作 O(n_server)，整体 O(R · n_server)，n_server ≤ 50
空间复杂度：O(R + obj 数量)
Stripe NG OA 拿满分的 4 个习惯
1）每加一 Part，先写 1-2 个手动测试
Part 5 的隐藏 case 多到无法只靠自查。每完成一个 Part，造一个 5-10 行的请求序列手动验证状态变化。
2）状态变量集中初始化
Stripe 评分会扫描"状态分散"的反模式。所有状态在 __init__ 初始化，不要在运行中动态加属性。
3）拒绝写日志保留
REJECT 不写入主 log，但 Stripe 部分版本要求返回拒绝列表。提交前确认题面要求。
4）连接幂等性
Part 5 SHUTDOWN 后的 server 应当能被再次 SHUTDOWN（幂等）；同一 connectId 连续 DISCONNECT 也应静默处理。
FAQ
Stripe NG OA 题目每年都换吗？
题面会变，但结构高度稳定——5 个递进 Part、维护多个映射、最后一个 Part 是"重路由 / 撤销"。2024-2026 都遵循这个模板：从 brace expansion → shipping cost → server load balancing。
60 分钟够吗？
够，但要前 15 分钟做规划：把 Part 1-5 的状态变量列出来，规划数据结构。直接开写很容易在 Part 4-5 因为状态分散需要重构。
一定要用 Class 吗？
强烈推荐用 Class——5 个 Part 共享状态，函数式写法变量传参会膨胀到几十行。Stripe 评分对 OOP 风格有偏好。
Part 5 时间不够能拿到通过吗？
可以。完成 Part 1-4（约 80% 通过率）就达到 Stripe NG OA 通过线。Part 5 是 stretch goal。
这道题在 1point3acres 还能搜到讨论吗？
可以。关键词搜 "Stripe NG OA 5 part"、"Stripe server load balancing"、"Stripe 2026 OA"，每周都有候选人 fresh 反馈。
正在准备 Stripe NG OA？
oavoservice 提供 Stripe NG OA 全流程实时辅助：拆 Part、状态变量规划、HackerRank 全过程协助。我们对 Stripe 历年的"5 Part 状态模拟"题型有完整复盘——server load balancing、subscription notifications、shipping cost、brace expansion 都能给出最优代码模板。
立即添加微信：Coding0201，获取 Stripe NG OA 实战辅导。
#Stripe #StripeOA #NewGrad #负载均衡 #状态模拟 #OA真题
联系方式
Email: [email protected]
Telegram: @OAVOProxy