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,避免 heap——後續 Part 還要按 objId / serverId 索引,heap 不靈活。
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):
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))
Part 5 三個隱藏 case:
- SHUTDOWN 後該 server 的 objId 繫結要全部清掉
- 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