Używanie routingu tylko do odczytu i grup dostępności

Grupy dostępności Always On zapewniają wysoką dostępność baz danych SQL Server. Przed nawiązaniem połączenia administrator bazy danych musi utworzyć detektor grupy dostępności i skonfigurować routing połączeń tylko do odczytu na serwerze. Sterownik mssql-python obsługuje połączenia z grupami dostępności za pomocą następujących słów kluczowych w ciągu połączenia:

• ApplicationIntent - Trasowanie połączeń do replik wtórnych tylko do odczytu
• MultiSubnetFailover — Włącz równoległe próby połączenia w różnych podsieciach, aby przyspieszyć przełączanie awaryjne

Połącz się z grupowym słuchaczem dostępności

Podstawowe połączenie ze słuchaczem

Połącz się z grupą dostępności przez nazwę DNS słuchacza, zamiast konkretnego serwera:

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()}")

Ze specyfikacją portu

Określ numer portu, jeśli nasłuch używa portu innego niż domyślny:

conn = mssql_python.connect(
    "Server=ag-listener.contoso.com,1433;"  # Listener with port
    "Database=<database>;"
    "Authentication=ActiveDirectoryDefault;"
    "Encrypt=yes;"
)

Rutowanie tylko odczytu

Włącz intencję tylko do odczytu

Użyj ApplicationIntent=ReadOnly, aby kierować do replik pomocniczych:

# 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;"
)

Zweryfikuj trasowanie

Potwierdź, która replika obsługiwała połączenie i czy jest to połączenie pierwotne czy wtórne:

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 wielu podsieci

Włącz wielopodsieciowe przełączanie awaryjne

Dla grup dostępności obejmujących wiele podsieci:

conn = mssql_python.connect(
    "Server=ag-listener.contoso.com;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryDefault;"
    "MultiSubnetFailover=yes;"
    "Encrypt=yes;"
)

To ustawienie:

• Próbuje łączyć się ze wszystkimi adresami IP równolegle.
• Skraca czas przełączania się w konfiguracjach wielopodsieciowych.
• Działa najlepiej w przypadku wszystkich połączeń grupy dostępności.

Wzorce architektury połączeń

Oddzielne połączenia odczytu i zapisu

Utrzymuj oddzielne połączenia, aby ruch odczytu trafiał do replik pomocniczych, a operacje zapisu do repliki podstawowej:

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()

Wzorzec odczytu po zapisie

Zapisz na serwerze podstawowym, a następnie zweryfikuj dane na serwerze wtórnym, uwzględniając opóźnienie replikacji:

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

Obsługa trybu failover

Odporność połączenia

Automatycznie ponawiaj zapytania, gdy przełączenie awaryjne przerwie połączenie:

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}
)

Pętla powtórek reaguje tylko na przejściowe kody błędów. Nieprzejściowe błędy, takie jak błędy uwierzytelniania, uprawnienia czy błędy składni zapytań, są zgłaszane natychmiast, ponieważ ponowne próby nie są w stanie ich rozwiązać.

Wykryć zmianę roli głównej

Sprawdź bieżącą rolę repliki i połącz się ponownie, jeśli serwer podstawowy został przeniesiony:

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

Raportowanie obciążeń

Przenieś raporty do systemu pomocniczego

Uruchamiaj zapytania raportowe na replice wtórnej, aby zmniejszyć obciążenie repliki podstawowej:

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))

Najlepsze rozwiązania

Zawsze używaj MultiSubnetFailover

Dołącz MultiSubnetFailover=yes do każdego parametrów połączenia grupy dostępności, aby przyspieszyć przełączanie awaryjne:

# 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;"
)

Dopasuj intencję do operacji

Kieruj każdą operację do właściwej repliki w zależności od tego, czy odczytuje lub zapisuje dane:

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

Pomocniczy uchwyt jest tymczasowo niedostępny

Ponów próbę użycia repliki pomocniczej w przypadku błędów przejściowych, a jeśli replika pomocnicza pozostaje nieosiągalna, przełącz się z powrotem na replikę pierwotną.

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()

Treści powiązane

• Parametry połączenia
• Logika ponowień
• Buforowanie połączeń
• grupy dostępności Always On