Notitie
Voor toegang tot deze pagina is autorisatie vereist. U kunt proberen u aan te melden of de directory te wijzigen.
Voor toegang tot deze pagina is autorisatie vereist. U kunt proberen de mappen te wijzigen.
Always On-beschikbaarheidsgroepen bieden hoge beschikbaarheid voor SQL Server-databases. Voordat je verbinding maakt, moet je databasebeheerder een availability group listener aanmaken en read-only routing op de server configureren . De mssql-python-driver ondersteunt beschikbaarheidsgroepverbindingen via deze verbindingsreeks-zoekwoorden:
-
ApplicationIntent- Verbindingen omleiden naar alleen-lezen secundaire replica's -
MultiSubnetFailover- Schakel parallelle verbindingspogingen over subnetten in voor snellere failover
Verbind met beschikbaarheidsgroep-luisteraar
Basisverbinding met luisteraars
Maak verbinding met de beschikbaarheidsgroep via de luisteraar-DNS-naam in plaats van een specifieke server:
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()}")
Met poortspecificatie
Specificeer een poortnummer wanneer de luisteraar een niet-standaard poort gebruikt:
conn = mssql_python.connect(
"Server=ag-listener.contoso.com,1433;" # Listener with port
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"Encrypt=yes;"
)
Alleen-lezenroutering
Schakel de alleen-lezenmodus in
Gebruik ApplicationIntent=ReadOnly om naar secundaire replica's te routeren:
# 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;"
)
Verifieer routering
Bevestig welke replica de verbinding heeft afgehandeld en of het primair of secundair is:
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)}")
Failover met meerdere subnetten
Schakel multi-subnet failover in
Voor beschikbaarheidsgroepen die meerdere subnetten beslaan:
conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"MultiSubnetFailover=yes;"
"Encrypt=yes;"
)
Deze instelling:
- Probeert parallel verbinding te maken met alle IP-adressen.
- Vermindert failovertijd in multi-subnet configuraties.
- Werkt het beste voor alle beschikbaarheidsgroepverbindingen.
Verbindingsarchitectuurpatronen
Afzonderlijke lees- en schrijfverbindingen
Onderhoud aparte verbindingen zodat het leesverkeer naar secundaire kanalen gaat en schrijfverkeer naar de primaire:
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()
Lees-na-schrijven patroon
Schrijf naar de primaire en verifieer vervolgens de data op een secundaire systeem, rekening houdend met replicatievertraging:
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
Failover afhandelen
Verbindingstolerantie
Herhaal automatisch query's wanneer een failover de verbinding onderbreekt:
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}
)
De retry-lus reageert alleen op tijdelijke foutcodes. Niet-tijdelijke fouten, zoals authenticatiefouten, machtigingsfouten of fouten in de querysyntaxis, worden onmiddellijk gegenereerd omdat een nieuwe poging ze niet kan verhelpen.
Verandering in primaire rol detecteren
Controleer de huidige replica-rol en verbind opnieuw als de primaire is verplaatst:
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
Rapportagewerklasten
Offload meldt zich aan de secundaire
Voer rapportagequeries uit tegen een secundaire replica om de belasting op de primaire te verminderen:
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))
Beste praktijken
Gebruik altijd MultiSubnetFailover
Neem MultiSubnetFailover=yes op in elke verbindingsreeks van een beschikbaarheidsgroep voor snellere failover:
# 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;"
)
Koppel intent aan bewerking
Route elke bewerking naar de juiste replica op basis van of deze data leest of schrijft:
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
Secundaire tijdelijk niet beschikbaar afhandelen
Probeer de secundaire opnieuw bij tijdelijke fouten, en val dan terug op de primaire replica als de secundaire onbereikbaar blijft.
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()