Always On 可用性群組為 SQL Server 資料庫提供高可用性。 在連接之前，你的資料庫管理員必須 建立可用性群組監聽器 ，並在伺服器 上設定唯讀路由 。 mssql-python 驅動程式透過以下 連接字串 關鍵字支援可用性群組連線：
-
ApplicationIntent- 將連線路由至唯讀次級副本 -
MultiSubnetFailover- 啟用跨子網的平行連線嘗試，以加快備援速度
連線到可用性群組接聽程式
基本監聽器連線
透過監聽者 DNS 名稱連接可用性群組，而非特定伺服器：
import mssql_python
# Connect via availability group listener
conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"Encrypt=yes;"
)
cursor = conn.cursor()
cursor.execute("SELECT @@SERVERNAME AS ServerName")
print(f"Connected to: {cursor.fetchval()}")
附端口規格
當竊聽者使用非預設埠時，請指定埠號：
conn = mssql_python.connect(
"Server=ag-listener.contoso.com,1433;" # Listener with port
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"Encrypt=yes;"
)
唯讀路由
啟用唯讀意圖
使用 ApplicationIntent=ReadOnly 路由至次要複本：
# Connect for read operations - routes to secondary
read_conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"ApplicationIntent=ReadOnly;"
"Encrypt=yes;"
)
# Connect for read-write operations - routes to primary
write_conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"ApplicationIntent=ReadWrite;" # Default
"Encrypt=yes;"
)
驗證路由
確認是由哪個副本處理該連線，以及其為主要副本還是次要副本：
def check_replica_role(conn) -> str:
"""Check if connected to primary or secondary."""
cursor = conn.cursor()
cursor.execute("""
SELECT
@@SERVERNAME AS ServerName,
CASE
WHEN DATABASEPROPERTYEX(DB_NAME(), 'Updateability') = 'READ_WRITE'
THEN 'Primary'
ELSE 'Secondary'
END AS Role
""")
row = cursor.fetchone()
return f"{row.ServerName} ({row.Role})"
read_conn = mssql_python.connect(connection_string + "ApplicationIntent=ReadOnly;")
print(f"Read connection: {check_replica_role(read_conn)}")
write_conn = mssql_python.connect(connection_string + "ApplicationIntent=ReadWrite;")
print(f"Write connection: {check_replica_role(write_conn)}")
多重子網路容錯移轉
啟用多子網路故障轉移
對於跨多個子網路的可用性群組：
conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"MultiSubnetFailover=yes;"
"Encrypt=yes;"
)
這項設定：
- 嘗試並行連接所有 IP 位址。
- 可縮短多子網路組態中的故障轉移時間。
- 這方法對所有可用性群組連線都效果最佳。
連線架構模式
獨立的讀寫連線
保持獨立連線，讓讀取流量送到次要節點，寫入流量則送到主要節點：
class DatabaseConnections:
"""Manage separate connections for read and write operations."""
def __init__(self, listener: str, database: str):
self.base_conn_str = f"Server={listener};Database={database};Authentication=ActiveDirectoryDefault;Encrypt=yes;"
self._read_conn = None
self._write_conn = None
@property
def read_connection(self):
"""Get or create read-only connection (secondary replica)."""
if self._read_conn is None:
self._read_conn = mssql_python.connect(
self.base_conn_str + "ApplicationIntent=ReadOnly;MultiSubnetFailover=yes;"
)
return self._read_conn
@property
def write_connection(self):
"""Get or create read-write connection (primary replica)."""
if self._write_conn is None:
self._write_conn = mssql_python.connect(
self.base_conn_str + "ApplicationIntent=ReadWrite;MultiSubnetFailover=yes;"
)
return self._write_conn
def close(self):
if self._read_conn:
self._read_conn.close()
if self._write_conn:
self._write_conn.close()
# Usage
db = DatabaseConnections("ag-listener.contoso.com", "<database>")
# Queries go to secondary
cursor = db.read_connection.cursor()
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product")
products = cursor.fetchall()
# Writes go to primary
cursor = db.write_connection.cursor()
cursor.execute("INSERT INTO #Products (Name) VALUES (%(name)s)", {"name": "New Product"})
db.write_connection.commit()
db.close()
寫入後讀取模式
先寫入主要節點，然後再於次要節點驗證資料，並將複寫延遲納入考量：
def create_order_and_verify(conn_manager, order_data: dict):
"""Create order on primary, verify on secondary with eventual consistency."""
# Write to primary
write_cursor = conn_manager.write_connection.cursor()
write_cursor.execute("""
INSERT INTO #Orders (CustomerID, Total) VALUES (%(cust)s, %(total)s);
SELECT SCOPE_IDENTITY();
""", order_data)
order_id = write_cursor.fetchval()
conn_manager.write_connection.commit()
# Wait for replication (in production, use more sophisticated approach)
import time
time.sleep(1)
# Verify on secondary
read_cursor = conn_manager.read_connection.cursor()
read_cursor.execute("SELECT * FROM #Orders WHERE OrderID = %(id)s", {"id": order_id})
if read_cursor.fetchone():
print(f"Order {order_id} replicated to secondary")
else:
print(f"Order {order_id} not yet replicated")
return order_id
處理故障轉移
連線韌性
當故障轉移中斷連線時，自動重試查詢：
import time
def execute_with_failover_retry(conn_str: str, query: str, params: dict,
max_retries: int = 3) -> list:
"""Execute query with automatic reconnection on failover."""
for attempt in range(max_retries + 1):
try:
conn = mssql_python.connect(conn_str)
cursor = conn.cursor()
cursor.execute(query, params)
results = cursor.fetchall()
conn.close()
return results
except mssql_python.OperationalError as e:
error_str = str(e)
# Check for failover-related errors
# Azure SQL transient error codes
# See: /azure/azure-sql/database/troubleshoot-common-errors-issues
if any(code in error_str for code in ["40613", "40197", "40501"]):
if attempt < max_retries:
print(f"Failover detected, retrying ({attempt + 1}/{max_retries})...")
time.sleep(5 * (attempt + 1)) # Exponential backoff
continue
raise
# Usage
results = execute_with_failover_retry(
"Server=<ag-listener>.contoso.com;Database=<database>;Authentication=ActiveDirectoryDefault;MultiSubnetFailover=yes;Encrypt=yes;",
"SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductSubcategoryID = %(cat)s",
{"cat": 5}
)
重試迴圈僅對暫態錯誤碼做出反應。 非暫時性失敗，如認證錯誤、權限錯誤或查詢語法錯誤，會立即被提出，因為重試無法解決。
偵測主要角色變更
檢查目前的副本角色，如果主節點移動了就重新連線：
def is_primary(conn) -> bool:
"""Check if current connection is to primary replica."""
cursor = conn.cursor()
cursor.execute("""
SELECT DATABASEPROPERTYEX(DB_NAME(), 'Updateability') AS Updateability
""")
return cursor.fetchval() == 'READ_WRITE'
def ensure_primary(conn_str: str) -> mssql_python.Connection:
"""Ensure connection is to primary, reconnect if needed."""
conn = mssql_python.connect(conn_str + "ApplicationIntent=ReadWrite;")
if not is_primary(conn):
# Might happen during failover
conn.close()
time.sleep(2)
conn = mssql_python.connect(conn_str + "ApplicationIntent=ReadWrite;")
return conn
工作負載報告
將報表移轉至次要系統
對次要副本執行報告查詢以減輕主副本的負載：
class ReportingService:
"""Service that runs reports on secondary replicas."""
def __init__(self, listener: str, database: str):
self.conn_str = (
f"Server={listener};"
f"Database={database};"
"Authentication=ActiveDirectoryDefault;"
"ApplicationIntent=ReadOnly;"
"MultiSubnetFailover=yes;"
"Encrypt=yes;"
)
def run_report(self, report_query: str, params: dict = None) -> list:
"""Execute report query on secondary replica."""
conn = mssql_python.connect(self.conn_str)
cursor = conn.cursor()
try:
cursor.execute(report_query, params or {})
return cursor.fetchall()
finally:
cursor.close()
conn.close()
def get_sales_summary(self, start_date, end_date) -> dict:
"""Run sales summary report."""
results = self.run_report("""
SELECT
YEAR(OrderDate) AS Year,
MONTH(OrderDate) AS Month,
COUNT(*) AS OrderCount,
SUM(TotalDue) AS TotalSales
FROM Sales.SalesOrderHeader
WHERE OrderDate BETWEEN %(start)s AND %(end)s
GROUP BY YEAR(OrderDate), MONTH(OrderDate)
ORDER BY Year, Month
""", {"start": start_date, "end": end_date})
return [
{"year": r.Year, "month": r.Month,
"orders": r.OrderCount, "sales": r.TotalSales}
for r in results
]
# Usage
reports = ReportingService("ag-listener.contoso.com", "sales_db")
summary = reports.get_sales_summary(date(2024, 1, 1), date(2024, 12, 31))
最佳做法
一定要使用 MultiSubnetFailover
在每個可用性群組連線字串中包含 MultiSubnetFailover=yes，以加快故障轉移：
# Recommended for all AG connections
conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"MultiSubnetFailover=yes;" # Always include this
"Encrypt=yes;"
)
匹配意圖與運作
根據每個操作是讀取還是寫入資料，將其路由至正確的複本：
def get_appropriate_connection(operation: str, connections: DatabaseConnections):
"""Get connection appropriate for the operation type."""
read_operations = {"SELECT", "REPORT", "EXPORT", "ANALYTICS"}
if operation.upper() in read_operations:
return connections.read_connection
else:
return connections.write_connection
處理次要系統暫時無法使用的情況
遇到暫態錯誤時再試一次次要副本，如果次要副本無法連通，再回退到主副本。
import time
def query_with_fallback(conn_manager, query: str, params: dict, max_retries: int = 2):
"""Query the secondary, retrying transient errors before falling back to the primary."""
transient_codes = ["40613", "40197", "40501"]
for attempt in range(max_retries + 1):
try:
cursor = conn_manager.read_connection.cursor()
cursor.execute(query, params)
return cursor.fetchall()
except mssql_python.OperationalError as e:
# Re-raise failures that retrying can't fix, such as authentication,
# permission, or query syntax errors.
if not any(code in str(e) for code in transient_codes):
raise
# Retry the secondary for transient errors before giving up.
if attempt < max_retries:
time.sleep(2 * (attempt + 1))
continue
# Secondary still unavailable after retries: fall back to the primary.
print("Secondary unavailable, using primary")
cursor = conn_manager.write_connection.cursor()
cursor.execute(query, params)
return cursor.fetchall()