3. 服务实例主备切换
任务目标
实现SOME/IP服务实例的主备冗余：主实例正常时处理客户端请求；当主实例故障（进程退出或心跳超时）后，备用实例自动接管服务。客户端通过相同的Service ID/Instance ID访问服务，切换过程对客户端无感知。
核心设计
采用共享状态 + 故障检测的最小可复现方案：主备实例共享同一服务状态存储（本案例用文件模拟），通过租约（Lease）机制决定当前主实例。主实例周期性续租；备用实例检测到租约过期后立即抢占成为新主实例。客户端始终向同一服务地址发送请求，由当前持有租约的实例响应。
实现代码
import time
import os
import threading
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
LEASE_FILE = "/tmp/someip_lease.json"
LEASE_TIMEOUT = 3.0 # 租约超时时间（秒）
RENEW_INTERVAL = 1.0 # 主实例续租周期（秒）
CHECK_INTERVAL = 0.5 # 备用实例检测周期（秒）
STATE_FILE = "/tmp/someip_state.json"
class ServiceState:
"""共享服务状态（主备通过文件共享）"""
@staticmethod
def read():
try:
with open(STATE_FILE) as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return {"counter": 0}
@staticmethod
def write(data):
with open(STATE_FILE, "w") as f:
json.dump(data, f)
def acquire_lease(instance_id):
"""尝试获取租约，成功返回True"""
now = time.time()
try:
with open(LEASE_FILE) as f:
lease = json.load(f)
if lease["expires_at"] > now and lease["owner"] != instance_id:
return False # 其他实例仍持有有效租约
except (FileNotFoundError, json.JSONDecodeError):
pass
# 写入自己的租约
with open(LEASE_FILE, "w") as f:
json.dump({"owner": instance_id, "expires_at": now + LEASE_TIMEOUT}, f)
return True
def renew_lease(instance_id):
"""主实例续租"""
with open(LEASE_FILE, "w") as f:
json.dump({"owner": instance_id, "expires_at": time.time() + LEASE_TIMEOUT}, f)
def is_primary(instance_id):
"""检查当前实例是否为主实例"""
try:
with open(LEASE_FILE) as f:
lease = json.load(f)
return lease["owner"] == instance_id and lease["expires_at"] > time.time()
except (FileNotFoundError, json.JSONDecodeError):
return False
class RequestHandler(BaseHTTPRequestHandler):
"""模拟SOME/IP服务请求处理（HTTP封装便于验证）"""
instance_id = None
def do_GET(self):
if not is_primary(self.instance_id):
self.send_response(503)
self.end_headers()
self.wfile.write(b"NOT_PRIMARY")
return
# 处理业务：计数器自增
state = ServiceState.read()
state["counter"] += 1
ServiceState.write(state)
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(state).encode())
def log_message(self, format, *args):
print(f"[{self.instance_id}] {format % args}")
def run_server(instance_id, port):
"""启动服务实例"""
RequestHandler.instance_id = instance_id
server = HTTPServer(("127.0.0.1", port), RequestHandler)
print(f"[{instance_id}] listening on port {port}")
server.serve_forever()
def main(instance_id, port):
# 启动HTTP服务线程
server_thread = threading.Thread(target=run_server, args=(instance_id, port), daemon=True)
server_thread.start()
# 主备逻辑
while True:
if acquire_lease(instance_id):
# 当前是主实例：周期性续租
print(f"[{instance_id}] became PRIMARY")
while is_primary(instance_id):
renew_lease(instance_id)
time.sleep(RENEW_INTERVAL)
print(f"[{instance_id}] lost primary role")
else:
# 备用实例：等待租约过期后抢占
time.sleep(CHECK_INTERVAL)
if __name__ == "__main__":
import sys
inst = sys.argv[1] # 实例ID，如 "A" 或 "B"
port = int(sys.argv[2])
main(inst, port)
运行与验证
依赖：Python 3.6+，无第三方库。
启动两个实例：
# 终端1：启动实例A（端口8001）
python3 someip_redundancy.py A 8001
# 终端2：启动实例B（端口8002）
python3 someip_redundancy.py B 8002
验证步骤：
- 观察两个终端日志，确认只有一个实例输出
became PRIMARY（通常先启动的A成为主实例）。 - 向主实例发送请求，确认返回200且计数器递增：
curl http://127.0.0.1:8001/ # 预期返回 {"counter": 1}
curl http://127.0.0.1:8001/ # 预期返回 {"counter": 2}
- 向备用实例发送请求，预期返回503（
NOT_PRIMARY）：
curl http://127.0.0.1:8002/ # 预期返回 503 NOT_PRIMARY
- 模拟主实例故障：在终端1按
Ctrl+C终止实例A。 - 等待约3秒（租约超时），观察终端2日志出现
became PRIMARY。 - 再次向原备用实例（现主实例B）发送请求，确认返回200且计数器从上次值继续递增：
curl http://127.0.0.1:8002/ # 预期返回 {"counter": 3}
通过条件：实例A故障后，实例B在租约超时（≤3秒）内自动接管；客户端改用端口8002后，服务状态（计数器值）连续无丢失，业务无中断感知。
关键说明
- 租约文件与状态文件均存储在
/tmp，实际部署应替换为共享存储或分布式协调服务（如etcd）。 - 客户端感知切换的关键是服务发现：本案例通过切换请求端口模拟，真实SOME/IP场景中客户端通过SD（Service Discovery）重新发现新主实例的端点。
- 本案例使用HTTP封装SOME/IP请求仅为便于验证；核心主备切换逻辑（租约抢占）与具体通信协议无关，可直接迁移至SOME/IP实现。