Hinweis
Für den Zugriff auf diese Seite ist eine Autorisierung erforderlich. Sie können versuchen, sich anzumelden oder das Verzeichnis zu wechseln.
Für den Zugriff auf diese Seite ist eine Autorisierung erforderlich. Sie können versuchen, das Verzeichnis zu wechseln.
Always On-Verfügbarkeitsgruppen bieten hohe Verfügbarkeit für SQL Server-Datenbanken. Vor der Verbindung muss Ihr Datenbankadministrator einen Verfügbarkeitsgruppen-Listener erstellen und das Nur-Lese-Routing auf dem Server konfigurieren. Der mssql-python-Treiber unterstützt Verfügbarkeitsgruppenverbindungen über diese Verbindungszeichenfolge-Schlüsselwörter:
-
ApplicationIntent- Verbindungen zu schreibgeschützten sekundären Replikaten weiterleiten -
MultiSubnetFailover- Ermöglichen Sie parallele Verbindungsversuche über Subnetze hinweg für schnellere Failover
Verbinden Sie sich mit Verfügbarkeitsgruppen-Listener
Grundlegende Hörerverbindung
Verbinden Sie sich mit der Verfügbarkeitsgruppe über den DNS-Namen des Listeners statt über einen bestimmten 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()}")
Mit Portspezifikation
Geben Sie eine Portnummer an, wenn der Hörer einen nicht-standardmäßigen Port verwendet:
conn = mssql_python.connect(
"Server=ag-listener.contoso.com,1433;" # Listener with port
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"Encrypt=yes;"
)
Schreibgeschütztes Routing
Schreib-nur-Absicht aktivieren
Verwenden Sie ApplicationIntent=ReadOnly, um an sekundäre Replikate weiterzuleiten:
# 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;"
)
Routing prüfen
Bestätigen Sie, welche Replik die Verbindung bearbeitet hat und ob sie primär oder sekundär ist:
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 für mehrere Subnetze
Multi-Subnet-Failover aktivieren
Für Verfügbarkeitsgruppen, die sich über mehrere Subnetze erstrecken:
conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"MultiSubnetFailover=yes;"
"Encrypt=yes;"
)
Diese Einstellung:
- Versucht, parallel Verbindungen zu allen IP-Adressen herzustellen.
- Verringert die Failover-Zeit in Multi-Subnet-Konfigurationen.
- Funktioniert am besten für alle Verfügbarkeitsgruppenverbindungen.
Muster der Verbindungsarchitektur
Getrennte Lese- und Schreibverbindungen
Verwenden Sie separate Verbindungen, sodass Lesezugriffe an die Sekundärinstanzen und Schreibzugriffe an die Primärinstanz gehen:
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()
Lese-nach-Schreib-Muster
Schreibe auf das primäre System und verifiziere dann die Daten auf einer sekundären Station, wobei die Replikationsverzögerung berücksichtigt wird:
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-Behandlung
Verbindungsstabilität
Abfragen automatisch wiederholen, wenn ein Failover die Verbindung unterbricht:
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}
)
Die Retry-Schleife reagiert nur auf vorübergehende Fehlercodes. Nicht-vorübergehende Fehler wie Authentifizierungsfehler, Berechtigungsfehler oder Abfragesyntaxfehler werden sofort angezeigt, da ein erneutes Versuch sie nicht lösen kann.
Identifikation einer Änderung der primären Rolle
Überprüfen Sie die aktuelle Rolle des Replikats und stellen Sie die Verbindung erneut her, falls das primäre Replikat gewechselt hat:
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
Berichterstattung von Arbeitslasten
Berichte auf das Sekundärsystem auslagern
Führen Sie Reporting-Abfragen gegen eine sekundäre Replik aus, um die Belastung der primären zu reduzieren:
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))
Bewährte Methoden
Verwenden Sie immer MultiSubnetFailover
Fügen Sie MultiSubnetFailover=yes in jede Verbindungszeichenfolge der Verfügbarkeitsgruppe ein, um ein schnelleres Failover zu ermöglichen:
# 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;"
)
Intent dem Vorgang zuordnen
Routen Sie jede Operation an die richtige Replik basierend darauf, ob sie Daten liest oder schreibt:
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
Sekundäre Kontrolle vorübergehend nicht verfügbar
Versuchen Sie die Sekundäre bei vorübergehenden Fehlern erneut und greifen Sie dann auf die primäre Replik zurück, falls die Sekundärkopie nicht erreichbar bleibt.
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()