Mysql

关注公众号 jb51net

关闭
首页 > 数据库 > Mysql > MySQL Group Replication实例

MySQL Group Replication实战:多主架构的配置、监控与故障切换

作者：程序员小一

MySQL Group Replication 是 MySQL 5.7.17 版本之后引入的一个多主复制解决方案,它允许在多个服务器之间实现同步复制,从而实现高可用性和负载均衡,本文介绍MySQL Group Replication实战:多主架构的配置、监控与故障切换,感兴趣的朋友一起看看吧

一、单主架构的天花板：当写流量突破单个节点的处理极限

在高并发写入场景下，传统的 MySQL 单主多从架构面临一个不可逾越的物理瓶颈——所有写入必须经过唯一的主节点。即使从库可以通过读写分离分担读压力，主库的 CPU、内存带宽和网络 IO 终有上限。

在团队经历的一次大促压测中，单主 MySQL 在写 QPS 达到 8000 时出现了连接堆积和事务超时。横向扩展只读副本无济于事，因为瓶颈在写入端。MySQL Group Replication（MGR）的多主（Multi-Primary）模式从架构层面解决了这个痛点：任何一个节点都可以同时接收读写请求，集群通过 Paxos 变体协议保证所有节点数据一致。

但这把"双刃剑"的另一面是：MGR 对事务冲突检测、网络延迟、节点数量配置极其敏感，配置不当会导致更高的冲突率和更差的性能。本文将分享 MGR 多主架构从配置、监控到故障切换的完整实战经验。

二、MGR多主模式的数据流与冲突检测机制

sequenceDiagram
    participant C1 as Client 1
    participant N1 as Node 1 (Primary)
    participant N2 as Node 2 (Primary)
    participant N3 as Node 3 (Primary)
    C1->>N1: UPDATE balance = balance - 100
WHERE user_id = 1
    Note over N1: 本地执行
    N1->>N1: 写入 Binlog
    N1->>N2: 广播 WriteSet
(user_id=1, 行Hash)
    N1->>N3: 广播 WriteSet
    Note over N2: 冲突检测
    N2->>N2: 检查 Certifier
    Note over N3: 冲突检测
    N3->>N3: 检查 Certifier
    N2-->>N1: ACK (无冲突)
    N3-->>N1: ACK (无冲突)
    N1->>N1: 提交事务
    C2->>N2: UPDATE balance = balance - 50
WHERE user_id = 1
    Note over N2: 本地执行
    N2->>N2: WriteSet检测到冲突
(与N1事务修改同一行)
    N2->>N2: 事务回滚 (先提交者胜)
    N2-->>C2: ERROR: 事务冲突

MGR 的核心机制：

  1. WriteSet 广播：每个事务提交前，将修改行的主键哈希（WriteSet）广播给所有节点
  2. Certification（冲突认证）：每个节点维护一个"认证数据库"，检查新事务的 WriteSet 是否与正在处理的事务冲突
  3. 先提交者胜（First Committer Wins）：如果两个事务修改了同一行，先通过认证的事务成功，后者回滚
  4. 流控（Flow Control）：当某个节点的待处理事务积压过多时，限制整个集群的写入速度

三、配置与监控的工程实践

3.1 多主模式配置

-- 所有节点执行的初始化配置
-- 1. 基础配置
SET GLOBAL server_id = 1;  -- 每个节点不同
SET GLOBAL gtid_mode = ON;
SET GLOBAL enforce_gtid_consistency = ON;
-- 2. 创建复制用户
CREATE USER 'repl'@'%' IDENTIFIED BY 'SecurePassword123!';
GRANT REPLICATION SLAVE ON *.* TO 'repl'@'%';
FLUSH PRIVILEGES;
-- 3. 配置 Group Replication 通道
CHANGE MASTER TO MASTER_USER='repl', MASTER_PASSWORD='SecurePassword123!'
FOR CHANNEL 'group_replication_recovery';
-- 4. 安装插件
INSTALL PLUGIN group_replication SONAME 'group_replication.so';
-- 5. 启动 MGR (首个节点用 bootstrap)
SET GLOBAL group_replication_bootstrap_group = ON;
START GROUP_REPLICATION;
SET GLOBAL group_replication_bootstrap_group = OFF;
-- 后续节点直接加入
START GROUP_REPLICATION;
# my.cnf 关键配置
[mysqld]
# === MGR 核心配置 ===
# 事务写集提取算法 (XXHASH64 性能和碰撞率最佳平衡)
transaction_write_set_extraction = XXHASH64
# Group 名称 (所有节点必须一致)
loose-group_replication_group_name = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
# 不引导组(仅首个节点设为 ON)
loose-group_replication_start_on_boot = OFF
# 本地地址
loose-group_replication_local_address = "192.168.1.10:33061"
# 种子节点列表
loose-group_replication_group_seeds = "192.168.1.10:33061,192.168.1.11:33061,192.168.1.12:33061"
# === 多主模式专用 ===
# 多主模式开启
loose-group_replication_single_primary_mode = OFF
# 允许所有节点写入
loose-group_replication_enforce_update_everywhere_checks = ON
# === 冲突检测优化 ===
# 认证数据库大小
loose-group_replication_certifier_threshold = 25000
# 事务大小限制(字节)
loose-group_replication_transaction_size_limit = 157286400
# === 流控配置 ===
# 流控模式 (QUOTA 基于配额)
loose-group_replication_flow_control_mode = QUOTA
# 允许积压的最大事务数
loose-group_replication_flow_control_certifier_threshold = 25000
# 成员间最大延迟事务数
loose-group_replication_flow_control_applier_threshold = 25000
# === 网络优化 ===
# 消息压缩
loose-group_replication_compression_threshold = 131072
# 压缩算法
loose-group_replication_compression_algorithm = zstd

3.2 监控采集脚本

#!/usr/bin/env python3
"""MGR 集群健康监控"""
import pymysql
from typing import Dict, List
from dataclasses import dataclass
from datetime import datetime
import json
@dataclass
class MemberStatus:
    host: str
    port: int
    state: str          # ONLINE / RECOVERING / ERROR / OFFLINE
    role: str           # PRIMARY / SECONDARY
    version: str
    gtid_executed: str
    transactions_behind: int
    is_local: bool
class MGRMonitor:
    """MGR 集群监控器"""
    def __init__(self, nodes: List[Dict]):
        self.nodes = nodes
        self.connections = {}
        self._init_connections()
    def _init_connections(self):
        """建立到所有节点的连接"""
        for node in self.nodes:
            try:
                conn = pymysql.connect(
                    host=node['host'],
                    port=node['port'],
                    user=node['user'],
                    password=node['password'],
                    connect_timeout=5,
                    read_timeout=10
                )
                self.connections[f"{node['host']}:{node['port']}"] = conn
            except Exception as e:
                print(f"连接 {node['host']}:{node['port']} 失败: {e}")
    def get_cluster_status(self) -> Dict:
        """获取完整集群状态"""
        status = {
            'timestamp': datetime.now().isoformat(),
            'members': [],
            'conflicts_detected': 0,
            'transactions_checked': 0,
            'flow_control_active': False,
            'alerts': []
        }
        for key, conn in self.connections.items():
            member = self._get_member_status(conn, key)
            status['members'].append(member)
            if member.state != 'ONLINE':
                status['alerts'].append({
                    'member': key,
                    'state': member.state,
                    'message': f'节点 {key} 处于 {member.state} 状态'
                })
        # 读取冲突统计
        for conn in self.connections.values():
            stats = self._get_gr_stats(conn)
            status['conflicts_detected'] += stats.get('conflicts', 0)
            status['transactions_checked'] += stats.get('checked', 0)
        # 冲突率告警
        if status['transactions_checked'] > 0:
            conflict_rate = status['conflicts_detected'] / status['transactions_checked']
            if conflict_rate > 0.01:  # 冲突率 > 1%
                status['alerts'].append({
                    'type': 'high_conflict_rate',
                    'rate': f'{conflict_rate:.2%}',
                    'message': f'事务冲突率过高: {conflict_rate:.2%}'
                })
        return status
    def _get_member_status(self, conn, key: str) -> MemberStatus:
        """获取单个成员状态"""
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        cursor.execute("""
            SELECT 
                MEMBER_HOST,
                MEMBER_PORT,
                MEMBER_STATE,
                MEMBER_ROLE,
                MEMBER_VERSION
            FROM performance_schema.replication_group_members
            WHERE MEMBER_HOST = %s
        """, (key.split(':')[0],))
        row = cursor.fetchone()
        if not row:
            return MemberStatus(
                host=key.split(':')[0],
                port=int(key.split(':')[1]),
                state='UNKNOWN',
                role='UNKNOWN',
                version='',
                gtid_executed='',
                transactions_behind=-1,
                is_local=False
            )
        # 获取延迟信息
        cursor.execute("""
            SELECT 
                COUNT_TRANSACTIONS_IN_QUEUE as behind_count
            FROM performance_schema.replication_group_member_stats
            WHERE MEMBER_ID = (
                SELECT MEMBER_ID 
                FROM performance_schema.replication_group_members 
                WHERE MEMBER_HOST = %s
            )
        """, (row['MEMBER_HOST'],))
        stats = cursor.fetchone()
        behind = stats['behind_count'] if stats else 0
        return MemberStatus(
            host=row['MEMBER_HOST'],
            port=row['MEMBER_PORT'],
            state=row['MEMBER_STATE'],
            role=row['MEMBER_ROLE'],
            version=row['MEMBER_VERSION'],
            gtid_executed='',
            transactions_behind=behind,
            is_local=True
        )
    def _get_gr_stats(self, conn) -> Dict:
        """获取 Group Replication 统计信息"""
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        try:
            cursor.execute("""
                SELECT 
                    VARIABLE_NAME,
                    VARIABLE_VALUE
                FROM performance_schema.global_status
                WHERE VARIABLE_NAME LIKE 'group_replication%'
            """)
            stats = {}
            for row in cursor.fetchall():
                if 'conflict' in row['VARIABLE_NAME'].lower():
                    stats['conflicts'] = int(row['VARIABLE_VALUE'])
                # 解析其他指标...
            return stats
        except Exception:
            return {'conflicts': 0, 'checked': 0}

四、多主模式的现实边界

冲突率的决定性因素

MGR 多主模式最关键的指标是事务冲突率。在以下场景中冲突率可以接受（< 0.1%）：

但在以下场景中冲突率可能高达 10-30%：

节点数量的黄金法则

节点数可用性写入性能建议场景
3允许 1 节点故障最优标准配置
5允许 2 节点故障略有下降高可用场景
7+允许 3+ 节点故障显著下降不推荐
2无法容错-仅测试

奇数节点是铁律。Paxos 协议依赖多数派（N/2 + 1），偶数节点浪费了容量但不增加任何容错能力。

五、总结

MGR 多主架构的核心取舍：

  1. 写入扩展 vs 冲突处理成本：多主写入不再受单点限制，但事务冲突是必然代价
  2. 数据分布设计是前提：按用户/业务维度将写入路由到固定节点，冲突率才能可控
  3. 流控是最后防线：当节点跟不上时果断限速，总比数据不一致好

在实际的生产环境中，将核心交易表的所有写入路由到同一节点，将非核心的日志和统计表写入分散到多个节点后，整体写 QPS 提升 3.2 倍（8000 → 25600），冲突率控制在 0.05% 以下。

到此这篇关于MySQL Group Replication实战:多主架构的配置、监控与故障切换的文章就介绍到这了,更多相关MySQL Group Replication实例内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文