← 返回部落格列表 Stripe 2026 NG OA 真題:伺服器負載均衡 5 個 Part 全解析|狀態模擬系統
Stripe

Stripe 2026 NG OA 真題:伺服器負載均衡 5 個 Part 全解析|狀態模擬系統

2026-05-14

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

輸入

輸出:所有被接受的連接日誌(按發生順序)

題目要求模擬伺服器在以下規則下的負載均衡:

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] 時:

  1. 該 server 上所有連接作廢(不算 DISCONNECT)
  2. 該 server 上的 objId 繫結釋放
  3. 後續 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

  1. SHUTDOWN 後該 server 的 objId 繫結要全部清掉
  2. SHUTDOWN 不能影響 log 裡的歷史 CONNECT 記錄
  3. 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