MySQL高可用与架构设计

在现代互联网应用中，数据库的高可用性和可扩展性至关重要。单点故障可能导致整个系统瘫痪，性能瓶颈可能影响用户体验。今天，我们将深入探讨MySQL的高可用架构设计，从主从复制到分布式集群，帮助你构建稳定可靠的数据库系统。

1. 主从复制架构

复制原理与三种复制模式

复制基本原理：

-- 复制过程涉及的关键线程-- 主库：Binlog Dump Thread-- 从库：I/O Thread, SQL Thread-- 查看主库状态SHOW MASTER STATUS;/*+------------------+----------+--------------+------------------+-------------------+| File             | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set |+------------------+----------+--------------+------------------+-------------------+| mysql-bin.000003 |      194 |              |                  |                   |+------------------+----------+--------------+------------------+-------------------+*/-- 查看从库状态SHOW SLAVE STATUS\G/*             Slave_IO_State: Waiting for master to send event                Master_Host: 192.168.1.100                Master_User: repl                Master_Port: 3306              Connect_Retry: 60            Master_Log_File: mysql-bin.000003        Read_Master_Log_Pos: 194             Relay_Log_File: relay-bin.000002              Relay_Log_Pos: 320      Relay_Master_Log_File: mysql-bin.000003           Slave_IO_Running: Yes          Slave_SQL_Running: Yes            Replicate_Do_DB:         Replicate_Ignore_DB:          Replicate_Do_Table:      Replicate_Ignore_Table:     Replicate_Wild_Do_Table: Replicate_Wild_Ignore_Table:                  Last_Errno: 0                 Last_Error:                Skip_Counter: 0        Exec_Master_Log_Pos: 194            Relay_Log_Space: 526            Until_Condition: None             Until_Log_File:               Until_Log_Pos: 0         Master_SSL_Allowed: No         Master_SSL_CA_File:          Master_SSL_CA_Path:             Master_SSL_Cert:           Master_SSL_Cipher:              Master_SSL_Key:       Seconds_Behind_Master: 0Master_SSL_Verify_Server_Cert: No             Last_IO_Errno: 0             Last_IO_Error:             Last_SQL_Errno: 0            Last_SQL_Error:   Replicate_Ignore_Server_Ids:              Master_Server_Id: 1                  Master_UUID: 6b0f1c1a-5d5e-11eb-ae93-000c29a3a3a3             Master_Info_File: mysql.slave_master_info                    SQL_Delay: 0          SQL_Remaining_Delay: NULL      Slave_SQL_Running_State: Slave has read all relay log; waiting for more updates           Master_Retry_Count: 86400                  Master_Bind:       Last_IO_Error_Timestamp:      Last_SQL_Error_Timestamp:                Master_SSL_Crl:            Master_SSL_Crlpath:            Retrieved_Gtid_Set:             Executed_Gtid_Set:                 Auto_Position: 0         Replicate_Rewrite_DB:                  Channel_Name:            Master_TLS_Version: */

三种复制模式对比：

-- 1. 基于语句的复制（Statement-Based Replication）-- 配置SET GLOBAL binlog_format = 'STATEMENT';-- 优点：二进制日志较小，网络传输量少-- 缺点：非确定性函数可能导致数据不一致-- 2. 基于行的复制（Row-Based Replication）SET GLOBAL binlog_format = 'ROW';-- 优点：数据一致性更好-- 缺点：二进制日志较大，网络传输量大-- 3. 混合模式复制（Mixed）SET GLOBAL binlog_format = 'MIXED';-- 优点：结合两者优势，自动选择最优方式-- 缺点：配置相对复杂-- 生产环境推荐使用ROW或MIXED模式

基于二进制日志的复制机制

二进制日志配置：

-- 查看二进制日志配置SHOW VARIABLES LIKE 'log_bin%';SHOW VARIABLES LIKE 'binlog_format%';SHOW VARIABLES LIKE 'sync_binlog%';SHOW VARIABLES LIKE 'expire_logs_days%';-- 二进制日志配置示例（my.cnf）/*[mysqld]# 启用二进制日志log_bin = /var/lib/mysql/mysql-bin# 日志格式binlog_format = ROW# 每次事务提交都同步到磁盘sync_binlog = 1# 日志保留7天expire_logs_days = 7# 每个日志文件大小max_binlog_size = 100M# 自动清理日志binlog_expire_logs_seconds = 604800*/

复制过滤规则：

-- 主库过滤规则-- 在my.cnf中配置/*# 忽略系统库的复制binlog_ignore_db = mysqlbinlog_ignore_db = information_schemabinlog_ignore_db = performance_schemabinlog_ignore_db = sys*/-- 从库过滤规则CHANGE MASTER TO     REPLICATE_DO_DB = (app_db),    REPLICATE_IGNORE_DB = (test,temp_db),    REPLICATE_DO_TABLE = (app_db.important_table),    REPLICATE_IGNORE_TABLE = (app_db.log_table);-- 通配符过滤CHANGE MASTER TO    REPLICATE_WILD_DO_TABLE = ('app_db.shard_%'),    REPLICATE_WILD_IGNORE_TABLE = ('app_db.temp_%');

半同步复制配置实战

半同步复制原理：

-- 安装半同步插件（主从库都需要）INSTALL PLUGIN rpl_semi_sync_master SONAME 'semisync_master.so';INSTALL PLUGIN rpl_semi_sync_slave SONAME 'semisync_slave.so';-- 查看插件状态SELECT PLUGIN_NAME, PLUGIN_STATUS FROM INFORMATION_SCHEMA.PLUGINS WHERE PLUGIN_NAME LIKE '%semi%';-- 配置主库半同步SET GLOBAL rpl_semi_sync_master_enabled = 1;SET GLOBAL rpl_semi_sync_master_timeout = 1000;  -- 1秒超时-- 配置从库半同步SET GLOBAL rpl_semi_sync_slave_enabled = 1;-- 查看半同步状态SHOW STATUS LIKE 'Rpl_semi_sync%';/*Rpl_semi_sync_master_status          | ONRpl_semi_sync_master_clients         | 2      -- 连接的半同步从库数量Rpl_semi_sync_master_yes_tx          | 1000   -- 成功通过半同步的事务数Rpl_semi_sync_master_no_tx           | 5      -- 超时后转为异步的事务数*/

半同步复制配置优化：

-- 持久化配置（在my.cnf中）/*[mysqld]# 主库配置plugin_load = "rpl_semi_sync_master=semisync_master.so;rpl_semi_sync_slave=semisync_slave.so"rpl_semi_sync_master_enabled = 1rpl_semi_sync_slave_enabled = 1rpl_semi_sync_master_timeout = 1000rpl_semi_sync_master_wait_point = AFTER_SYNC  -- MySQL 5.7+ 推荐*/-- 监控半同步复制SELECT     VARIABLE_NAME,    VARIABLE_VALUEFROM performance_schema.global_statusWHERE VARIABLE_NAME LIKE 'RPL_SEMI_SYNC%';-- 半同步复制降级监控-- 当从库响应超时或故障时，主库会自动降级为异步复制-- 需要监控降级事件并及时处理

多源复制与链式复制

多源复制配置：

-- MySQL 5.7+ 支持多源复制-- 从多个主库复制数据到单个从库-- 配置多源复制通道-- 主库1配置CHANGE MASTER TO     MASTER_HOST = 'master1_host',    MASTER_USER = 'repl',    MASTER_PASSWORD = 'password',    MASTER_PORT = 3306,    MASTER_AUTO_POSITION = 1FOR CHANNEL 'master1';-- 主库2配置  CHANGE MASTER TO    MASTER_HOST = 'master2_host',    MASTER_USER = 'repl',    MASTER_PASSWORD = 'password',    MASTER_PORT = 3306,    MASTER_AUTO_POSITION = 1FOR CHANNEL 'master2';-- 启动多源复制START SLAVE FOR CHANNEL 'master1';START SLAVE FOR CHANNEL 'master2';-- 查看多源复制状态SHOW SLAVE STATUS FOR CHANNEL 'master1'\GSHOW SLAVE STATUS FOR CHANNEL 'master2'\G-- 按通道过滤操作STOP SLAVE SQL_THREAD FOR CHANNEL 'master1';START SLAVE SQL_THREAD FOR CHANNEL 'master1';

链式复制架构：

-- 三级复制链：Master -> Relay Slave -> Leaf Slave-- 配置中继从库/*Master配置:log_bin = onlog_slave_updates = off  -- 默认，中继库不需要记录从库更新Relay Slave配置:log_bin = onlog_slave_updates = on   -- 关键：记录从主库接收的更新Leaf Slave配置:log_bin = off  -- 或者 on，根据需求log_slave_updates = off*/-- 中继从库的特殊配置/*[mysqld]# 中继从库配置server_id = 2log_bin = mysql-binlog_slave_updates = 1relay_log = relay-binread_only = 1# 过滤规则（可选）replicate_do_db = app_dbreplicate_ignore_db = mysql*/

复制故障排查与修复

常见复制错误处理：

-- 1. 主键冲突错误-- 错误信息：Duplicate entry 'X' for key 'PRIMARY'-- 解决方案：STOP SLAVE;SET GLOBAL sql_slave_skip_counter = 1;START SLAVE;-- 或者手动处理冲突数据STOP SLAVE;-- 查看冲突数据SELECT * FROM table_name WHERE primary_key = 'X';-- 删除冲突数据或更新主键DELETE FROM table_name WHERE primary_key = 'X';START SLAVE;-- 2. 数据不存在错误-- 错误信息：Can't find record in 'table_name'-- 解决方案：STOP SLAVE;-- 在从库插入缺失的数据INSERT IGNORE INTO table_name VALUES (...);START SLAVE;-- 3. 网络中断导致的复制延迟-- 监控复制延迟SHOW SLAVE STATUS\G-- 查看Seconds_Behind_Master-- 自动重连配置CHANGE MASTER TO     MASTER_CONNECT_RETRY = 60,    MASTER_RETRY_COUNT = 86400;

GTID复制故障处理：

-- 启用GTID复制-- 在my.cnf中配置/*[mysqld]gtid_mode = ONenforce_gtid_consistency = ON*/-- GTID复制错误处理-- 查看错误的GTIDSHOW SLAVE STATUS\G-- Last_SQL_Error: Coordinator stopped because there were error(s) in the worker(s)...-- Retrieved_Gtid_Set: 6b0f1c1a-5d5e-11eb-ae93-000c29a3a3a3:1-100-- Executed_Gtid_Set: 6b0f1c1a-5d5e-11eb-ae93-000c29a3a3a3:1-95-- 跳过特定GTID事务STOP SLAVE;SET GTID_NEXT = '6b0f1c1a-5d5e-11eb-ae93-000c29a3a3a3:96';BEGIN; COMMIT;SET GTID_NEXT = 'AUTOMATIC';START SLAVE;-- 重置GTID复制-- 注意：这会清除所有复制信息，需要重新配置STOP SLAVE;RESET SLAVE ALL;CHANGE MASTER TO ...;START SLAVE;

2. 高可用集群方案

MySQL Router读写分离

MySQL Router部署配置：

# MySQL Router配置文件 (mysqlrouter.conf)[DEFAULT]logging_folder = /var/log/mysqlrouterruntime_folder = /var/run/mysqlrouterconfig_folder = /etc/mysqlrouter[routing:read_write]bind_address = 0.0.0.0bind_port = 6446destinations = metadata-cache://mycluster/?role=PRIMARYrouting_strategy = first-available[routing:read_only]bind_address = 0.0.0.0bind_port = 6447destinations = metadata-cache://mycluster/?role=SECONDARYrouting_strategy = round-robin# 启动MySQL Router# mysqlrouter --config=/etc/mysqlrouter/mysqlrouter.conf &

应用程序连接配置：

# Python应用程序连接示例import mysql.connector# 写操作连接（主库）write_config = {    'host': 'router_host',    'port': 6446,  # 读写端口    'user': 'app_user',    'password': 'password',    'database': 'app_db'}# 读操作连接（从库）read_config = {    'host': 'router_host',     'port': 6447,  # 只读端口    'user': 'app_user',    'password': 'password',    'database': 'app_db'}# 写操作def update_user_profile(user_id, data):    conn = mysql.connector.connect(**write_config)    # 执行更新操作    conn.close()# 读操作  def get_user_profile(user_id):    conn = mysql.connector.connect(**read_config)    # 执行查询操作    conn.close()

MHA自动故障转移

MHA架构组成：

# MHA组件# 1. MHA Manager - 管理节点# 2. MHA Node - 数据节点代理# MHA Manager配置 (app1.cnf)[server default]manager_log=/var/log/masterha/app1.logmanager_workdir=/var/log/masterha/app1master_binlog_dir=/var/lib/mysqluser=mha_userpassword=mha_passwordping_interval=3remote_workdir=/tmprepl_user=repl_userrepl_password=repl_passwordssh_user=root[server1]hostname=master_hostport=3306[server2] hostname=slave1_hostport=3306candidate_master=1[server3]hostname=slave2_hostport=3306no_master=1# 启动MHA监控masterha_manager --conf=/etc/masterha/app1.cnf

MHA故障转移过程：

# 1. 检测主库故障# 2. 选择新主库（优先candidate_master=1的从库）# 3. 应用差异的二进制日志# 4. 提升新主库# 5. 其他从库指向新主库# 6. 虚拟IP切换（可选）# 手动触发故障转移masterha_master_switch --conf=/etc/masterha/app1.cnf --master_state=dead# 检查MHA状态masterha_check_status --conf=/etc/masterha/app1.cnf# MHA监控脚本示例#!/bin/bash# mha_monitor.shCONFIG_FILE="/etc/masterha/app1.cnf"LOG_FILE="/var/log/masterha/monitor.log"while true; do    status=$(masterha_check_status --conf=$CONFIG_FILE 2>&1)    if [[ $status != *"alive"* ]]; then        echo "$(date): MHA manager is not running, restarting..." >> $LOG_FILE        nohup masterha_manager --conf=$CONFIG_FILE >> $LOG_FILE 2>&1 &    fi    sleep 30done

Orchestrator管理工具

Orchestrator部署配置：

// orchestrator.conf.json{  "Debug": false,  "EnableSyslog": false,    "MySQLTopologyUser": "orchestrator",  "MySQLTopologyPassword": "orchestrator_password",  "MySQLTopologyCredentialsConfigFile": "",  "MySQLTopologySSLPrivateKeyFile": "",  "MySQLTopologySSLCertFile": "",  "MySQLTopologySSLCAFile": "",  "MySQLTopologySSLSkipVerify": true,  "MySQLTopologyUseMutualTLS": false,    "MySQLOrchestratorHost": "127.0.0.1",  "MySQLOrchestratorPort": 3306,  "MySQLOrchestratorDatabase": "orchestrator",  "MySQLOrchestratorUser": "orchestrator",  "MySQLOrchestratorPassword": "orchestrator_password",    "RaftEnabled": true,  "RaftDataDir": "/var/lib/orchestrator",  "RaftBind": "192.168.1.100",  "DefaultRaftPort": 10008,    "AutoPseudoGTID": false,  "DetectClusterAliasQuery": "SELECT SUBSTRING_INDEX(@@hostname, '.', 1)",  "DetectInstanceAliasQuery": "SELECT @@hostname",    "RecoveryPeriodBlockSeconds": 3600,  "RecoveryIgnoreHostnameFilters": [],    "PromotionIgnoreHostnameFilters": [],    "ApplyMySQLPromotionAfterMasterFailover": true,  "PreFailoverProcesses": [    "echo 'Will recover from {failureType} on {failureCluster}' >> /tmp/recovery.log"  ],  "PostFailoverProcesses": [    "echo 'Recovered from {failureType} on {failureCluster}. Failed: {failedHost}:{failedPort}; Successor: {successorHost}:{successorPort}' >> /tmp/recovery.log"  ]}

Orchestrator API使用：

# 通过REST API管理集群# 发现并注册实例curl "http://orchestrator:3000/api/discover/192.168.1.101/3306"# 查看集群拓扑curl "http://orchestrator:3000/api/cluster/myapp"# 手动故障转移curl "http://orchestrator:3000/api/force-master-failover/myapp"# 查看恢复信息curl "http://orchestrator:3000/api/audit-recovery"# 维护模式curl "http://orchestrator:3000/api/maintenance/myapp/begin"curl "http://orchestrator:3000/api/maintenance/myapp/end"

基于Keepalived的VIP方案

Keepalived配置：

# keepalived.confglobal_defs {    router_id MYSQL_HA}vrrp_script chk_mysql {    script "/usr/bin/mysqlchk"    interval 2    weight 2    fall 2    rise 2}vrrp_instance VI_1 {    state BACKUP    interface eth0    virtual_router_id 51    priority 100    advert_int 1        authentication {        auth_type PASS        auth_pass 1111    }        virtual_ipaddress {        192.168.1.200    }        track_script {        chk_mysql    }        notify_master "/etc/keepalived/notify.sh master"    notify_backup "/etc/keepalived/notify.sh backup"    notify_fault "/etc/keepalived/notify.sh fault"}

MySQL健康检查脚本：

#!/bin/bash# mysqlchk - MySQL健康检查脚本MYSQL_HOST="localhost"MYSQL_PORT="3306"MYSQL_USER="health_check"MYSQL_PASS="health_check_password"MYSQL_CMD="/usr/bin/mysql"# 检查MySQL是否可连接$MYSQL_CMD -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASS -e "SELECT 1;" > /dev/null 2>&1if [ $? -eq 0 ]; then    # 检查复制状态（如果是从库）    SLAVE_STATUS=$($MYSQL_CMD -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASS -e "SHOW SLAVE STATUS\G" 2>/dev/null)        if [ -n "$SLAVE_STATUS" ]; then        # 是从库，检查复制状态        IO_RUNNING=$(echo "$SLAVE_STATUS" | grep "Slave_IO_Running:" | awk '{print $2}')        SQL_RUNNING=$(echo "$SLAVE_STATUS" | grep "Slave_SQL_Running:" | awk '{print $2}')        SECONDS_BEHIND=$(echo "$SLAVE_STATUS" | grep "Seconds_Behind_Master:" | awk '{print $2}')                if [ "$IO_RUNNING" = "Yes" ] && [ "$SQL_RUNNING" = "Yes" ] && [ "$SECONDS_BEHIND" -lt 60 ]; then            exit 0  # 健康        else            exit 1  # 不健康        fi    else        # 是主库，直接健康        exit 0    fielse    exit 1  # MySQL不可连接fi

状态切换通知脚本：

#!/bin/bash# notify.sh - 状态切换通知TYPE=$1VIP="192.168.1.200"LOG_FILE="/var/log/keepalived.log"log() {    echo "$(date): $1" >> $LOG_FILE}case $TYPE in    master)        log "切换为MASTER状态，绑定VIP: $VIP"        # 这里可以添加提升为主库的逻辑        # 比如设置read_only=OFF，通知应用等        mysql -e "SET GLOBAL read_only=OFF;"        ;;    backup)        log "切换为BACKUP状态，释放VIP"        # 设置只读模式        mysql -e "SET GLOBAL read_only=ON;"        ;;    fault)        log "进入FAULT状态"        ;;    *)        log "未知状态: $TYPE"        ;;esac

高可用架构选型指南

架构选型矩阵：

选型考虑因素：

-- 业务需求评估-- 1. RTO（恢复时间目标）SELECT     CASE         WHEN rto_requirement <= 30 THEN '需要自动故障转移'        WHEN rto_requirement <= 300 THEN '半自动故障转移'        ELSE '手动故障转移可接受'    END as ha_levelFROM business_requirements;-- 2. RPO（数据恢复点目标）SELECT     CASE        WHEN rpo_requirement = 0 THEN '需要同步复制'        WHEN rpo_requirement <= 1 THEN '需要半同步复制'         WHEN rpo_requirement <= 60 THEN '异步复制可接受'        ELSE '数据丢失可接受'    END as data_protection_levelFROM business_requirements;-- 3. 读写分离需求SELECT     CASE        WHEN read_ratio > 0.8 THEN '需要强大的读写分离'        WHEN read_ratio > 0.5 THEN '需要基础读写分离'        ELSE '读写分离非必需'    END as read_write_separationFROM workload_analysis;

3. 数据库架构设计

读写分离架构设计

应用层读写分离：

// Java应用层读写分离示例@Componentpublic class DataSourceRouter {        @Value("${datasource.master.url}")    private String masterUrl;        @Value("${datasource.slave.url}")     private String slaveUrl;        private ThreadLocal<Boolean> readOnly = new ThreadLocal<>();        public void setReadOnly(boolean readOnly) {        this.readOnly.set(readOnly);    }        public DataSource getDataSource() {        if (Boolean.TRUE.equals(readOnly.get())) {            return createDataSource(slaveUrl);        } else {            return createDataSource(masterUrl);        }    }        // AOP切面自动设置读写分离    @Aspect    @Component    public class ReadWriteSeparationAspect {                @Around("@annotation(org.springframework.transaction.annotation.Transactional)")        public Object handleTransaction(ProceedingJoinPoint joinPoint) throws Throwable {            Transactional transactional = ((MethodSignature) joinPoint.getSignature())                .getMethod().getAnnotation(Transactional.class);                        if (transactional.readOnly()) {                DataSourceContextHolder.setReadOnly(true);            }                        try {                return joinPoint.proceed();            } finally {                DataSourceContextHolder.clear();            }        }    }}

中间件读写分离：

# ShardingSphere配置示例# config-sharding.yamldataSources:  master_ds:    url: jdbc:mysql://master:3306/db?serverTimezone=UTC&useSSL=false    username: root    password: password    connectionTimeoutMilliseconds: 30000    idleTimeoutMilliseconds: 60000    maxLifetimeMilliseconds: 1800000    maxPoolSize: 50  slave_ds_0:    url: jdbc:mysql://slave0:3306/db?serverTimezone=UTC&useSSL=false    username: root    password: password    connectionTimeoutMilliseconds: 30000    idleTimeoutMilliseconds: 60000    maxLifetimeMilliseconds: 1800000    maxPoolSize: 50  slave_ds_1:    url: jdbc:mysql://slave1:3306/db?serverTimezone=UTC&useSSL=false    username: root    password: password    connectionTimeoutMilliseconds: 30000    idleTimeoutMilliseconds: 60000    maxLifetimeMilliseconds: 1800000    maxPoolSize: 50rules:- !LOAD_BALANCE  loadBalancers:    round_robin:      type: ROUND_ROBIN  dataSources:    read_ds:      dataSourceNames:        - slave_ds_0        - slave_ds_1      loadBalancerName: round_robin      - !SINGLE  defaultDataSource: master_ds  loadBalancers:    round_robin:      type: ROUND_ROBIN

分库分表策略与实现

水平分表策略：

-- 用户表按ID范围分表-- 创建分表CREATE TABLE users_0000 LIKE users_template;CREATE TABLE users_0001 LIKE users_template;CREATE TABLE users_0002 LIKE users_template;-- ... 创建更多分表-- 分表路由函数DELIMITER //CREATE FUNCTION get_user_table_name(user_id BIGINT)RETURNS VARCHAR(64)DETERMINISTICBEGIN    DECLARE table_suffix VARCHAR(4);    SET table_suffix = LPAD(MOD(user_id, 16), 4, '0');    RETURN CONCAT('users_', table_suffix);END //DELIMITER ;-- 分表查询示例SET @user_id = 123456;SET @table_name = get_user_table_name(@user_id);SET @sql = CONCAT('SELECT * FROM ', @table_name, ' WHERE user_id = ?');PREPARE stmt FROM @sql;EXECUTE stmt USING @user_id;DEALLOCATE PREPARE stmt;

垂直分库设计：

-- 业务垂直拆分-- 用户库CREATE DATABASE user_center;USE user_center;CREATE TABLE users (    user_id BIGINT PRIMARY KEY,    username VARCHAR(50),    email VARCHAR(100),    password_hash VARCHAR(255),    created_at TIMESTAMP);CREATE TABLE user_profiles (    user_id BIGINT PRIMARY KEY,    real_name VARCHAR(100),    avatar_url VARCHAR(500),    bio TEXT);-- 订单库CREATE DATABASE order_center;USE order_center;CREATE TABLE orders (    order_id BIGINT PRIMARY KEY,    user_id BIGINT,  -- 跨库关联    total_amount DECIMAL(12,2),    status VARCHAR(20),    created_at TIMESTAMP);CREATE TABLE order_items (    item_id BIGINT PRIMARY KEY,    order_id BIGINT,    product_id BIGINT,    quantity INT,    price DECIMAL(10,2));-- 商品库CREATE DATABASE product_center;USE product_center;CREATE TABLE products (    product_id BIGINT PRIMARY KEY,    product_name VARCHAR(200),    category_id INT,    price DECIMAL(10,2),    stock_quantity INT);

数据拆分：垂直拆分与水平拆分

垂直拆分实施：

-- 原始大表CREATE TABLE user_comprehensive (    user_id BIGINT PRIMARY KEY,    -- 基础信息    username VARCHAR(50),    email VARCHAR(100),    password_hash VARCHAR(255),    -- 个人信息    real_name VARCHAR(100),    id_card VARCHAR(20),    phone VARCHAR(20),    -- 扩展信息    education VARCHAR(50),    occupation VARCHAR(50),    income_level INT,    -- 行为信息    last_login_time TIMESTAMP,    login_count INT,    -- 其他字段...    created_at TIMESTAMP,    updated_at TIMESTAMP);-- 垂直拆分后-- 用户基础表（高频访问）CREATE TABLE users_basic (    user_id BIGINT PRIMARY KEY,    username VARCHAR(50),    email VARCHAR(100),    password_hash VARCHAR(255),    last_login_time TIMESTAMP,    login_count INT,    created_at TIMESTAMP);-- 用户详情表（低频访问）CREATE TABLE users_detail (    user_id BIGINT PRIMARY KEY,    real_name VARCHAR(100),    id_card VARCHAR(20),    phone VARCHAR(20),    education VARCHAR(50),    occupation VARCHAR(50),    income_level INT,    updated_at TIMESTAMP);-- 创建索引优化查询ALTER TABLE users_basic ADD INDEX idx_username (username);ALTER TABLE users_basic ADD INDEX idx_email (email);ALTER TABLE users_detail ADD INDEX idx_phone (phone);

水平拆分策略：

-- 时间范围分表（适用于时间序列数据）-- 按月分表CREATE TABLE logs_2023_01 LIKE logs_template;CREATE TABLE logs_2023_02 LIKE logs_template;CREATE TABLE logs_2023_03 LIKE logs_template;-- 时间分表管理存储过程DELIMITER //CREATE PROCEDURE create_next_month_table()BEGIN    DECLARE next_month VARCHAR(7);    DECLARE table_name VARCHAR(64);    DECLARE create_sql TEXT;        SET next_month = DATE_FORMAT(DATE_ADD(NOW(), INTERVAL 1 MONTH), '%Y_%m');    SET table_name = CONCAT('logs_', next_month);    SET create_sql = CONCAT('CREATE TABLE IF NOT EXISTS ', table_name, ' LIKE logs_template');        PREPARE stmt FROM create_sql;    EXECUTE stmt;    DEALLOCATE PREPARE stmt;        -- 记录创建日志    INSERT INTO table_creation_log (table_name, created_at)     VALUES (table_name, NOW());END //DELIMITER ;-- 地理分表（适用于地域性数据）CREATE TABLE users_north LIKE users_template;  -- 北方用户CREATE TABLE users_south LIKE users_template;  -- 南方用户CREATE TABLE users_east LIKE users_template;   -- 东方用户  CREATE TABLE users_west LIKE users_template;   -- 西方用户-- 基于业务特征分表CREATE TABLE users_vip LIKE users_template;    -- VIP用户CREATE TABLE users_normal LIKE users_template; -- 普通用户CREATE TABLE users_trial LIKE users_template;  -- 试用用户

分布式ID生成方案

数据库序列方案：

-- 基于数据库的ID生成器CREATE TABLE sequence_generator (    sequence_name VARCHAR(50) PRIMARY KEY,    current_value BIGINT NOT NULL DEFAULT 0,    step INT NOT NULL DEFAULT 1,    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP);-- 获取下一个ID的存储过程DELIMITER //CREATE FUNCTION next_id(seq_name VARCHAR(50))RETURNS BIGINTBEGIN    DECLARE current_val BIGINT;    DECLARE retry_count INT DEFAULT 0;    DECLARE max_retries INT DEFAULT 3;        retry_loop: WHILE retry_count < max_retries DO        -- 获取当前值        SELECT current_value INTO current_val        FROM sequence_generator         WHERE sequence_name = seq_name;                IF current_val IS NULL THEN            -- 初始化序列            INSERT INTO sequence_generator (sequence_name, current_value)             VALUES (seq_name, 1)            ON DUPLICATE KEY UPDATE current_value = 1;            SET current_val = 1;        END IF;                -- 尝试更新        UPDATE sequence_generator         SET current_value = current_value + step,            updated_at = CURRENT_TIMESTAMP        WHERE sequence_name = seq_name           AND current_value = current_val;                IF ROW_COUNT() = 1 THEN            RETURN current_val + 1;        END IF;                SET retry_count = retry_count + 1;        DO SLEEP(0.01);  -- 短暂等待后重试    END WHILE;        -- 重试失败，抛出异常    SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'Failed to generate sequence ID';END //DELIMITER ;

Snowflake算法实现：

-- Snowflake ID生成器表CREATE TABLE snowflake_worker (    worker_id INT PRIMARY KEY,    datacenter_id INT NOT NULL,    worker_name VARCHAR(100),    last_timestamp BIGINT,    sequence BIGINT DEFAULT 0,    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP);-- Snowflake ID生成函数DELIMITER //CREATE FUNCTION snowflake_next_id(worker_id INT)RETURNS BIGINTBEGIN    DECLARE epoch BIGINT DEFAULT 1609459200000; -- 2021-01-01    DECLARE current_ms BIGINT;    DECLARE last_ms BIGINT;    DECLARE sequence_val BIGINT;    DECLARE datacenter_id_val INT;        -- 获取worker信息    SELECT last_timestamp, sequence, datacenter_id     INTO last_ms, sequence_val, datacenter_id_val    FROM snowflake_worker     WHERE worker_id = worker_id    FOR UPDATE;  -- 加锁防止并发        -- 计算当前时间戳    SET current_ms = (UNIX_TIMESTAMP(NOW(3)) * 1000);        IF current_ms < last_ms THEN        SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'Clock moved backwards';    END IF;        IF current_ms = last_ms THEN        SET sequence_val = (sequence_val + 1) & 4095;  -- 12位序列号，最大4095        IF sequence_val = 0 THEN            -- 序列号耗尽，等待下一毫秒            SET current_ms = wait_next_ms(last_ms);        END IF;    ELSE        SET sequence_val = 0;    END IF;        -- 更新worker状态    UPDATE snowflake_worker     SET last_timestamp = current_ms,        sequence = sequence_val    WHERE worker_id = worker_id;        -- 生成ID: 时间戳(41位) + 数据中心ID(5位) + 工作节点ID(5位) + 序列号(12位)    RETURN ((current_ms - epoch) << 22)          | (datacenter_id_val << 17)          | (worker_id << 12)          | sequence_val;END //CREATE FUNCTION wait_next_ms(last_ms BIGINT)RETURNS BIGINTBEGIN    DECLARE current_ms BIGINT;    SET current_ms = (UNIX_TIMESTAMP(NOW(3)) * 1000);    WHILE current_ms <= last_ms DO        SET current_ms = (UNIX_TIMESTAMP(NOW(3)) * 1000);    END WHILE;    RETURN current_ms;END //DELIMITER ;

数据迁移与同步方案

在线数据迁移：

-- 双写迁移方案-- 1. 准备阶段：创建新表，建立双写机制CREATE TABLE users_new LIKE users_old;-- 2. 数据同步阶段：存量数据迁移INSERT INTO users_new SELECT * FROM users_old WHERE id > ? AND id <= ?;  -- 分批迁移-- 3. 增量数据双写-- 应用程序同时写入users_old和users_new-- 4. 数据验证SELECT     COUNT(*) as old_count,    (SELECT COUNT(*) FROM users_new) as new_count,    COUNT(*) - (SELECT COUNT(*) FROM users_new) as diffFROM users_old;-- 5. 切换阶段：停止写入旧表，全面使用新表-- 6. 清理阶段：删除旧表-- 使用pt-online-schema-change工具-- pt-online-schema-change --alter="ADD COLUMN new_column INT" D=database,t=table --execute

数据同步监控：

-- 创建数据同步监控表CREATE TABLE data_sync_monitor (    id BIGINT AUTO_INCREMENT PRIMARY KEY,    sync_job VARCHAR(100) NOT NULL,    source_count BIGINT,    target_count BIGINT,    diff_count BIGINT,    sync_status ENUM('running', 'completed', 'failed'),    started_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,    completed_at TIMESTAMP NULL,    error_message TEXT);-- 数据一致性检查存储过程DELIMITER //CREATE PROCEDURE check_data_consistency(    IN source_table VARCHAR(64),    IN target_table VARCHAR(64),    IN primary_key VARCHAR(64))BEGIN    DECLARE source_total BIGINT;    DECLARE target_total BIGINT;    DECLARE diff_count BIGINT;        -- 检查记录总数    SET @source_sql = CONCAT('SELECT COUNT(*) INTO @source_count FROM ', source_table);    PREPARE stmt1 FROM @source_sql;    EXECUTE stmt1;    DEALLOCATE PREPARE stmt1;        SET @target_sql = CONCAT('SELECT COUNT(*) INTO @target_count FROM ', target_table);    PREPARE stmt2 FROM @target_sql;    EXECUTE stmt2;    DEALLOCATE PREPARE stmt2;        SET source_total = @source_count;    SET target_total = @target_count;    SET diff_count = ABS(source_total - target_total);        -- 记录检查结果    INSERT INTO data_sync_monitor (sync_job, source_count, target_count, diff_count, sync_status)    VALUES (CONCAT(source_table, '_to_', target_table), source_total, target_total, diff_count,            CASE WHEN diff_count = 0 THEN 'completed' ELSE 'failed' END);        -- 如果有差异，记录具体差异数据    IF diff_count > 0 THEN        -- 这里可以添加更详细的差异分析        INSERT INTO data_diff_log (sync_job, diff_type, diff_details)        VALUES (CONCAT(source_table, '_to_', target_table), 'count_mismatch',                CONCAT('Source: ', source_total, ', Target: ', target_total));    END IF;    END //DELIMITER ;

总结

通过本篇的深入学习，我们掌握了MySQL高可用架构设计的核心知识：

1. 主从复制：理解了复制原理、配置方法和故障处理
2. 高可用方案：掌握了MHA、Orchestrator、Keepalived等工具的使用
3. 架构设计：学会了读写分离、分库分表、分布式ID生成等高级技术
4. 数据迁移：了解了在线数据迁移和同步的最佳实践

关键架构原则：

• 冗余设计：确保没有单点故障
• 自动故障转移：减少人工干预，提高可用性
• 监控告警：及时发现问题并处理
• 容量规划：提前规划系统扩展能力
• 数据安全：保证数据的一致性和完整性

架构演进路径：

1. 单机架构 → 主从复制
2. 主从复制 → 读写分离
3. 读写分离 → 分库分表
4. 分库分表 → 分布式数据库

动手练习：

1. 搭建MySQL主从复制环境，并测试故障转移
2. 配置MHA或Orchestrator实现自动故障转移
3. 设计并实施读写分离架构
4. 实践分库分表方案，解决单表数据量过大的问题
5. 实现分布式ID生成方案

欢迎在评论区分享你的高可用架构实践经验和遇到的问题！

Redis完全指南：配置文件详解与常用命令大全

本文深入解析Redis核心配置，并提供全面的命令参考手册，助你彻底掌握Redis使用技巧。

📖 概述

Redis作为高性能的键值数据库，在缓存、消息队列、会话存储等场景中广泛应用。掌握其配置文件和常用命令是每个开发者必备的技能。

⚙️ Redis配置文件深度解析

配置文件位置与加载

# 默认配置文件路径/etc/redis/redis.conf# 指定配置文件启动redis-server /path/to/your/redis.conf# 检查当前配置redis-cli config get *

核心配置项详解

🔒 网络与安全配置

# 绑定IP地址（生产环境建议指定）bind 127.0.0.1 192.168.1.100# 端口配置port 6379# 保护模式（外网访问需关闭）protected-mode no# 连接密码requirepass "your_strong_password_here"# 最大连接数maxclients 10000

💡 生产环境建议：

• 务必设置强密码
• 限制绑定IP，避免暴露到公网
• 适当调整最大连接数

💾 持久化配置

RDB持久化配置：

# 自动保存条件save 900 1      # 15分钟内至少1个变更save 300 10     # 5分钟内至少10个变更  save 60 10000   # 1分钟内至少10000个变更# RDB文件配置dbfilename dump.rdbdir /var/lib/redis# 压缩配置rdbcompression yesrdbchecksum yes

AOF持久化配置：

# 开启AOFappendonly yesappendfilename "appendonly.aof"# 同步策略appendfsync everysec    # 推荐配置# AOF重写配置auto-aof-rewrite-percentage 100auto-aof-rewrite-min-size 64mb

🎯 持久化策略选择：

• 缓存场景：仅使用RDB
• 数据安全要求高：RDB+AOF
• 性能优先：调整AOF同步策略为everysec

🧠 内存管理配置

# 内存限制maxmemory 2gb# 内存淘汰策略maxmemory-policy volatile-lru# 淘汰策略说明：# volatile-lru    -> 从过期键中淘汰最近最少使用# allkeys-lru     -> 从所有键中淘汰最近最少使用  # volatile-ttl    -> 从过期键中淘汰存活时间最短# volatile-random -> 从过期键中随机淘汰# noeviction      -> 不淘汰，返回错误

性能优化配置

# 内核参数优化vm.overcommit_memory = 1# 禁用透明大页echo never > /sys/kernel/mm/transparent_hugepage/enabled# 网络优化tcp-backlog 511timeout 0tcp-keepalive 300

⌨️ Redis常用命令大全

1. 🔑 键(Key)操作命令

基础操作

# 设置键值（带过期时间）SET user:1001 "John Doe" EX 3600# 批量操作MSET user:1001 "John" user:1002 "Jane" user:1003 "Bob"# 获取值GET user:1001# 删除键DEL user:1001 user:1002

高级特性

# 设置带过期时间的键（原子操作）SETEX session:token 1800 "encrypted_data"# 仅当键不存在时设置（分布式锁基础）SETNX lock:resource_1 "owner_id"# 获取并设置（原子操作）GETSET counter:clicks "100"

2. 📝 字符串(String)操作

# 数值操作INCR article:1001:views    # 阅读量+1INCRBY user:1001:points 10 # 积分+10DECR inventory:item_001    # 库存-1# 字符串操作APPEND user:1001:bio " Additional info"STRLEN user:1001:name      # 字符串长度GETRANGE user:1001:bio 0 4 # 获取子串

3. 🗂️ 哈希(Hash)操作

用户信息存储示例：

# 设置用户信息HSET user:1001 name "John" age 30 email "john@example.com"# 批量设置HMSET product:1001 name "Laptop" price 999.99 stock 50 category "Electronics"# 获取信息HGET user:1001 nameHMGET user:1001 name age emailHGETALL user:1001# 数值操作HINCRBY user:1001:stats login_count 1HINCRBYFLOAT product:1001 price -50.5

4. 📋 列表(List)操作

消息队列实现：

# 生产者：推送消息LPUSH message:queue "task_1"LPUSH message:queue "task_2"# 消费者：获取消息RPOP message:queue# 阻塞式获取（推荐）BRPOP message:queue 30# 查看队列LRANGE message:queue 0 -1LLEN message:queue

5. 🔄 集合(Set)操作

标签系统实现：

# 添加标签SADD article:1001:tags "tech" "programming" "redis"SADD user:1001:interests "coding" "gaming"# 查找共同兴趣SINTER user:1001:interests user:1002:interests# 推荐相关文章SUNION article:1001:tags article:1002:tags# 随机推荐SRANDMEMBER article:1001:tags 3

6. 📊 有序集合(Sorted Set)操作

排行榜实现：

# 添加分数ZADD leaderboard 1500 "player_1"ZADD leaderboard 3200 "player_2" 2800 "player_3"# 获取排名ZREVRANGE leaderboard 0 9 WITHSCORES  # 前10名ZRANK leaderboard "player_1"          # 升序排名ZREVRANK leaderboard "player_1"       # 降序排名# 范围查询ZRANGEBYSCORE leaderboard 2000 3000 WITHSCORES

🛠️ 实战应用场景

缓存策略实现

# 缓存查询结果SETEX cache:user:1001:profile 300 "{user_data}"# 缓存穿透防护SETNX cache_mutex:user:9999 1 EX 5

分布式会话

# 存储会话HSET session:abc123 user_id 1001 last_active 1635789000EXPIRE session:abc123 1800# 更新活跃时间EXPIRE session:abc123 1800

限流器实现

# 简单限流INCR rate_limit:api:1001EXPIRE rate_limit:api:1001 60# 复杂限流（使用Lua脚本）EVAL "local current = redis.call('incr', KEYS[1]) if current == 1 then redis.call('expire', KEYS[1], ARGV[1]) end return current" 1 rate_limit:complex 60

📈 监控与维护命令

系统状态检查

# 基础信息redis-cli info# 内存分析redis-cli info memory# 持久化状态redis-cli info persistence# 查看慢查询redis-cli slowlog get 10

性能监控

# 实时监控redis-cli monitor# 客户端连接管理redis-cli client listredis-cli client kill 127.0.0.1:53422# 内存分析redis-cli --bigkeysredis-cli --memkeys

备份与恢复

# 手动RDB备份redis-cli bgsave# AOF重写redis-cli bgrewriteaof# 数据迁移redis-cli --rdb dump.rdb

🚀 性能优化技巧

1. 连接池配置

# Python示例import redispool = redis.ConnectionPool(    max_connections=50,    host='localhost',     port=6379,    decode_responses=True)r = redis.Redis(connection_pool=pool)

2. 管道(Pipeline)优化

# 批量操作，减少网络往返pipe = r.pipeline()for user_id in user_ids:    pipe.hgetall(f"user:{user_id}")results = pipe.execute()

3. Lua脚本使用

# 原子性操作示例EVAL "local current = redis.call('get', KEYS[1]) if current then return redis.call('incr', KEYS[1]) else return nil end" 1 counter:test

⚠️ 常见问题排查

内存问题

# 查看内存使用详情redis-cli info memory | grep used_memory_human# 查找大Keyredis-cli --bigkeys# 内存碎片率redis-cli info memory | grep mem_fragmentation_ratio

连接问题

# 查看连接数redis-cli info clients# 客户端列表redis-cli client list# 网络统计redis-cli info stats | grep -E "(total_connections_received|rejected_connections)"

📚 总结

通过本文的学习，你应该已经掌握了：

• ✅ Redis配置文件的各项参数含义及调优方法
• ✅ 各类数据结构的适用场景及操作命令
• ✅ 常见业务场景的Redis实现方案
• ✅ 性能监控与问题排查技巧

Redis的强大之处在于其丰富的数据结构和原子操作，合理运用可以极大提升系统性能。建议在实际项目中多实践，逐步深入理解各个特性的使用场景。

欢迎在评论区留言交流，如果你觉得这篇文章有帮助，请点赞收藏支持！

Redis总结与展望：从入门到生产实践的完整指南

经过前面五篇深入的学习，我们已经完成了从Redis小白到生产级应用开发者的蜕变。在这最后一篇中，让我们回顾整个学习旅程，总结关键知识点，并展望Redis未来的发展方向和生态体系。

一、系列回顾：我们的Redis学习之旅

让我们简要回顾这个系列涵盖的核心内容：

第一篇：Redis核心概念与快速入门

• 理解了Redis为什么快（内存存储、单线程模型、I/O多路复用）
• 学会了使用Docker快速搭建Redis环境
• 掌握了基本的键值操作和通用命令
• 认识了Redis的典型应用场景

第二篇：玩转Redis五大核心数据结构

• String：不仅仅是文本，支持计数器、位图等高级用法
• Hash：存储对象的最佳选择，内存效率高
• List：实现消息队列和最新列表的利器
• Set：无序唯一集合，强大的集合运算能力
• Sorted Set：有序集合，排行榜和时间轴的核心

第三篇：Redis的持久化与高可用

• RDB：快照式持久化，适合备份和快速恢复
• AOF：日志式持久化，保证数据安全
• 主从复制：数据冗余和读写分离的基础
• 哨兵模式：实现自动故障转移的高可用方案

第四篇：Redis在Asp.Net Core项目中的实战应用

• 集成StackExchange.Redis客户端
• 实现商品信息缓存和缓存策略
• 解决缓存穿透、击穿、雪崩问题
• 使用分布式锁控制并发访问
• 配置分布式Session和消息队列

第五篇：进阶知识与运维管理

• 内存优化和淘汰策略配置
• Redis Cluster集群搭建和管理
• 性能监控、慢查询分析和调优
• 备份恢复、安全配置和故障诊断

二、Redis最佳实践总结

1. 键名设计规范

// 好的键名设计"user:1001:profile"          // 用户信息"product:2024:hotlist"       // 商品热榜"order:20240101:123456"      // 订单信息"session:abc123def456"       // 会话数据// 避免的键名设计"user_info_1001"             // 不一致的分隔符"data"                       // 过于简单，容易冲突"very_long_key_name_that_is_hard_to_read_and_remember" // 过长

键名设计原则：

• 使用统一的命名空间和分隔符（推荐冒号）
• 保持简洁但具有描述性
• 避免特殊字符和过长的键名

2. 避免大Key和热Key

大Key问题解决方案：

// 拆分大Hashpublic async Task SetLargeUserDataAsync(int userId, UserLargeData data){    // 拆分为多个Hash    await _database.HashSetAsync($"user:{userId}:basic", new[] {        new HashEntry("name", data.Name),        new HashEntry("email", data.Email)    });        await _database.HashSetAsync($"user:{userId}:profile", new[] {        new HashEntry("bio", data.Bio),        new HashEntry("avatar", data.AvatarUrl)    });}// 使用SCAN替代KEYSpublic async Task<List<string>> ScanKeysAsync(string pattern, int pageSize = 1000){    var keys = new List<string>();    var cursor = 0;        do    {        var result = await _database.ExecuteAsync("SCAN", cursor.ToString(), "MATCH", pattern, "COUNT", pageSize.ToString());        var innerResult = (RedisResult[])result;                cursor = int.Parse((string)innerResult[0]);        var pageKeys = (string[])innerResult[1];        keys.AddRange(pageKeys);            } while (cursor != 0);        return keys;}

热Key解决方案：

// 本地缓存 + Redis多级缓存public class MultiLevelCacheService{    private readonly IMemoryCache _memoryCache;    private readonly IRedisService _redisService;    private readonly TimeSpan _localCacheDuration = TimeSpan.FromMinutes(1);        public async Task<T> GetWithLocalCacheAsync<T>(string key)    {        // 先查本地缓存        if (_memoryCache.TryGetValue(key, out T localValue))            return localValue;                    // 本地缓存未命中，查询Redis        var redisValue = await _redisService.GetAsync<T>(key);        if (redisValue != null)        {            // 写入本地缓存            _memoryCache.Set(key, redisValue, _localCacheDuration);        }                return redisValue;    }}

3. 连接池与资源管理

public static class RedisConnectionManager{    private static Lazy<ConnectionMultiplexer> _lazyConnection;        static RedisConnectionManager()    {        _lazyConnection = new Lazy<ConnectionMultiplexer>(() =>        {            var configuration = new ConfigurationOptions            {                EndPoints = { "localhost:6379" },                AbortOnConnectFail = false,                ConnectRetry = 3,                ConnectTimeout = 5000,                KeepAlive = 180,                SyncTimeout = 5000,                // 连接池配置                AllowAdmin = false,                ClientName = $"{Environment.MachineName}:{Guid.NewGuid()}"            };                        return ConnectionMultiplexer.Connect(configuration);        });    }        public static ConnectionMultiplexer Connection => _lazyConnection.Value;        public static IDatabase GetDatabase()    {        return Connection.GetDatabase();    }}

4. 监控与告警配置

在Asp.Net Core中实现完整的监控：

public class RedisMetricsCollector : BackgroundService{    private readonly IConnectionMultiplexer _redis;    private readonly ILogger<RedisMetricsCollector> _logger;    private readonly IMetricsPublisher _metricsPublisher;        protected override async Task ExecuteAsync(CancellationToken stoppingToken)    {        while (!stoppingToken.IsCancellationRequested)        {            try            {                var server = _redis.GetServer(_redis.GetEndPoints().First());                var info = await server.InfoAsync("all");                                // 收集关键指标                var metrics = new RedisMetrics                {                    Timestamp = DateTime.UtcNow,                    ConnectedClients = long.Parse(info.First(x => x.Key == "Clients")                        .First(x => x.Key == "connected_clients").Value),                    UsedMemory = long.Parse(info.First(x => x.Key == "Memory")                        .First(x => x.Key == "used_memory").Value),                    OpsPerSecond = long.Parse(info.First(x => x.Key == "Stats")                        .First(x => x.Key == "instantaneous_ops_per_sec").Value),                    HitRate = CalculateHitRate(info),                    NetworkInput = long.Parse(info.First(x => x.Key == "Stats")                        .First(x => x.Key == "total_net_input_bytes").Value),                    NetworkOutput = long.Parse(info.First(x => x.Key == "Stats")                        .First(x => x.Key == "total_net_output_bytes").Value)                };                                await _metricsPublisher.PublishAsync(metrics);                                // 检查告警条件                await CheckAlerts(metrics);            }            catch (Exception ex)            {                _logger.LogError(ex, "收集Redis指标时发生错误");            }                        await Task.Delay(TimeSpan.FromSeconds(30), stoppingToken);        }    }        private double CalculateHitRate(ILookup<string, KeyValuePair<string, string>> info)    {        var hits = long.Parse(info.First(x => x.Key == "Stats")            .First(x => x.Key == "keyspace_hits").Value);        var misses = long.Parse(info.First(x => x.Key == "Stats")            .First(x => x.Key == "keyspace_misses").Value);                    return hits + misses == 0 ? 0 : (double)hits / (hits + misses);    }        private async Task CheckAlerts(RedisMetrics metrics)    {        // 内存使用率超过80%        if (metrics.UsedMemory > 0.8 * 1024 * 1024 * 1024) // 假设1GB内存        {            _logger.LogWarning("Redis内存使用率过高: {UsedMemory} bytes", metrics.UsedMemory);        }                // 命中率低于90%        if (metrics.HitRate < 0.9)        {            _logger.LogWarning("Redis缓存命中率过低: {HitRate:P2}", metrics.HitRate);        }    }}

三、Redis生态与工具介绍

1. 常用Redis可视化工具

RedisInsight（官方推荐）

• 功能全面的GUI工具
• 支持数据浏览、CLI操作、性能监控
• 免费使用，跨平台支持

Another Redis Desktop Manager

• 开源免费的桌面管理器
• 直观的界面，支持多种数据类型展示
• 跨平台支持

Redis Commander

• 基于Web的管理界面
• 适合部署在服务器环境
• 轻量级，功能齐全

2. 监控与运维平台

Prometheus + Grafana

# Redis Exporter配置scrape_configs:  - job_name: 'redis'    static_configs:      - targets: ['redis-exporter:9121']    metrics_path: /scrape    params:      target: ['redis-server:6379']

DataDog / New Relic

• 商业APM工具
• 提供深度性能分析和告警
• 企业级功能支持

3. Redis模块系统简介

Redis 4.0引入了模块系统，允许开发者扩展Redis功能：

RedisJSON

• 原生支持JSON文档
• 提供JSONPath查询语法

# 存储和查询JSON文档127.0.0.1:6379> JSON.SET user:1001 $ '{"name":"Alice","age":30}'127.0.0.1:6379> JSON.GET user:1001 $.name

RedisSearch

• 全文搜索功能
• 二级索引支持

# 创建全文搜索索引127.0.0.1:6379> FT.CREATE productIdx ON HASH PREFIX 1 product: SCHEMA name TEXT WEIGHT 5.0 description TEXT

RedisBloom

• 概率数据结构
• 布隆过滤器、基数估算等

# 使用布隆过滤器127.0.0.1:6379> BF.ADD visited:users user123127.0.0.1:6379> BF.EXISTS visited:users user123

RedisTimeSeries

• 时间序列数据处理
• 支持聚合和降采样

# 存储时间序列数据127.0.0.1:6379> TS.ADD temperature:room1 1620000000 25.5127.0.0.1:6379> TS.RANGE temperature:room1 1620000000 1620003600

四、Redis未来发展趋势

1. Redis 7.0+ 新特性

Functions（替代Lua脚本）

#!lua name=mylibredis.register_function('my_hset', function(keys, args)    return redis.call('HSET', keys[1], args[1], args[2])end)

ACL增强

• 更细粒度的权限控制
• 键模式权限管理
• 用户角色管理

性能优化

• 多线程I/O（非数据操作）
• 更高效的内存管理
• 改进的集群性能

2. 云原生与Kubernetes集成

Redis Operator

• 自动化Redis集群部署
• 故障自愈和弹性伸缩
• 备份和恢复管理

服务网格集成

• 与Istio、Linkerd的深度集成
• 智能流量路由和负载均衡
• 可观测性增强

3. AI与机器学习集成

向量搜索

# 使用Redis作为向量数据库127.0.0.1:6379> FT.CREATE vec_idx ON HASH PREFIX 1 vec: SCHEMA vector VECTOR127.0.0.1:6379> HSET vec:1 vector "0.1,0.2,0.3"

实时特征存储

• 机器学习特征工程
• 在线推理数据准备
• 实时推荐系统

五、Redis的局限性及替代方案

虽然Redis功能强大，但也有其局限性：

不适合使用Redis的场景

大量数据存储

• Redis主要依赖内存，成本较高
• 替代方案：Cassandra、HBase

复杂查询和分析

• Redis查询能力相对有限
• 替代方案：Elasticsearch、ClickHouse

强一致性事务

• Redis事务非ACID兼容
• 替代方案：关系型数据库

新兴竞品分析

KeyDB

• Redis的多线程版本
• 更好的多核CPU利用率
• 完全兼容Redis协议

Dragonfly

• 新型高性能内存数据库
• 声称比Redis快25倍
• 创新的数据结构设计

AWS ElastiCache for Redis

• 托管Redis服务
• 自动备份、故障转移
• 企业级功能支持

六、结语：持续学习的建议

通过这个系列的学习，你已经建立了坚实的Redis知识体系。但技术的道路永无止境，以下是一些持续学习的建议：

1. 实践是最好的老师

• 在自己的项目中积极应用Redis
• 尝试解决真实世界的性能问题
• 参与开源项目，阅读优秀的Redis使用案例

2. 关注社区动态

• 关注Redis官方博客和GitHub仓库
• 参与Redis Conf等技术大会
• 加入相关的技术社区和论坛

3. 深入原理研究

• 阅读《Redis设计与实现》
• 分析Redis源码，理解内部机制
• 尝试自己实现简单的内存数据库

4. 拓展技术视野

• 学习其他类型的数据库（关系型、文档型、图数据库等）
• 了解分布式系统理论
• 掌握云原生技术栈

最后的思考

Redis不仅仅是一个缓存工具，它已经发展成为现代应用架构中的多功能数据平台。从简单的键值存储到复杂的数据结构服务，从单机部署到全球分布式集群，Redis一直在演进。

记住这个核心理念：

“选择合适的工具解决正确的问题，并深入理解你所使用的工具。”

希望这个Redis系列教程能够成为你技术成长道路上有价值的参考资料。无论你是初学者还是经验丰富的开发者，对Redis的深入理解都将为你的职业生涯带来显著的提升。

感谢你坚持学完这个系列！如果在学习过程中有任何疑问或心得，欢迎在评论区分享交流。技术的道路需要同行者，让我们共同进步！

“学无止境，实践出真知。愿你在技术的道路上越走越远，不断突破自我！”

这个完整的Redis系列教程到这里就全部结束了。从基础概念到生产实践，从简单使用到深度优化，希望这个系列能够成为你在Redis学习道路上的得力助手。祝你编程愉快，技术精进！

Redis在Asp.Net Core项目中的实战应用：从缓存到分布式锁

通过前几篇的学习，我们已经掌握了Redis的核心概念和数据持久化。但理论终究要落地到实践。今天，我们将把Redis真正集成到Asp.Net Core项目中，解决真实业务场景中的性能瓶颈和分布式难题。

在现代Web开发中，Redis早已不是可选项，而是构建高性能、高可用系统的必备组件。作为.Net开发者，掌握如何在Asp.Net Core中熟练使用Redis，是你进阶高级开发的必经之路。

一、环境准备：在Asp.Net Core中集成Redis

1. 安装必要的NuGet包

首先，在你的Asp.Net Core项目中安装最常用的Redis客户端：

# 使用Package Manager ConsoleInstall-Package StackExchange.Redis# 或使用.NET CLIdotnet add package StackExchange.Redis

为什么选择StackExchange.Redis？

• 高性能且线程安全
• 支持同步和异步操作
• 活跃的社区支持和持续更新
• Microsoft官方推荐

2. 配置Redis服务

在appsettings.json中添加Redis连接字符串：

{  "ConnectionStrings": {    "Redis": "localhost:6379,password=your_password,abortConnect=false,connectTimeout=30000"  },  // 其他配置...}

在Program.cs中注册Redis服务：

using StackExchange.Redis;var builder = WebApplication.CreateBuilder(args);// 添加Redis服务builder.Services.AddSingleton<IConnectionMultiplexer>(sp =>     ConnectionMultiplexer.Connect(builder.Configuration.GetConnectionString("Redis")));// 注册自定义Redis服务（推荐）builder.Services.AddScoped<IRedisService, RedisService>();var app = builder.Build();

3. 创建Redis服务封装

为了更好地使用Redis，我们创建一个服务封装类：

public interface IRedisService{    Task<T> GetAsync<T>(string key);    Task SetAsync<T>(string key, T value, TimeSpan? expiry = null);    Task<bool> RemoveAsync(string key);    Task<bool> ExistsAsync(string key);}public class RedisService : IRedisService{    private readonly IConnectionMultiplexer _redis;    private readonly IDatabase _database;    public RedisService(IConnectionMultiplexer redis)    {        _redis = redis;        _database = redis.GetDatabase();    }    public async Task<T> GetAsync<T>(string key)    {        var value = await _database.StringGetAsync(key);        if (value.IsNullOrEmpty)            return default;        return JsonSerializer.Deserialize<T>(value);    }    public async Task SetAsync<T>(string key, T value, TimeSpan? expiry = null)    {        var serializedValue = JsonSerializer.Serialize(value);        await _database.StringSetAsync(key, serializedValue, expiry);    }    public async Task<bool> RemoveAsync(string key)    {        return await _database.KeyDeleteAsync(key);    }    public async Task<bool> ExistsAsync(string key)    {        return await _database.KeyExistsAsync(key);    }}

现在，我们的基础环境已经搭建完成，可以开始实战应用了！

二、缓存实战：提升系统性能的利器

缓存是Redis最经典的应用场景。让我们来看几个实际的例子。

1. 商品信息缓存

假设我们有一个电商系统，商品信息的查询非常频繁：

public interface IProductService{    Task<Product> GetProductByIdAsync(int productId);    Task UpdateProductAsync(Product product);}public class ProductService : IProductService{    private readonly IProductRepository _productRepository;    private readonly IRedisService _redisService;    private readonly ILogger<ProductService> _logger;    public ProductService(IProductRepository productRepository,                          IRedisService redisService,                         ILogger<ProductService> logger)    {        _productRepository = productRepository;        _redisService = redisService;        _logger = logger;    }    public async Task<Product> GetProductByIdAsync(int productId)    {        var cacheKey = $"product:{productId}";                // 1. 先查缓存        var product = await _redisService.GetAsync<Product>(cacheKey);        if (product != null)        {            _logger.LogInformation("从缓存获取商品 {ProductId}", productId);            return product;        }        // 2. 缓存不存在，查询数据库        _logger.LogInformation("缓存未命中，从数据库查询商品 {ProductId}", productId);        product = await _productRepository.GetByIdAsync(productId);        if (product == null)            return null;        // 3. 写入缓存，设置30分钟过期        await _redisService.SetAsync(cacheKey, product, TimeSpan.FromMinutes(30));                return product;    }    public async Task UpdateProductAsync(Product product)    {        // 更新数据库        await _productRepository.UpdateAsync(product);                // 删除缓存，保证数据一致性        var cacheKey = $"product:{product.Id}";        await _redisService.RemoveAsync(cacheKey);                _logger.LogInformation("更新商品 {ProductId} 并清除缓存", product.Id);    }}

缓存策略分析：

• 读取时：先查缓存，命中则返回；未命中查数据库并回写缓存
• 更新时：先更新数据库，再删除缓存（Cache-Aside模式）
• 过期时间：设置合理的过期时间，防止数据长期不更新

2. 在Controller中使用缓存服务

[ApiController][Route("api/[controller]")]public class ProductsController : ControllerBase{    private readonly IProductService _productService;    public ProductsController(IProductService productService)    {        _productService = productService;    }    [HttpGet("{id}")]    public async Task<ActionResult<Product>> GetProduct(int id)    {        var product = await _productService.GetProductByIdAsync(id);        if (product == null)            return NotFound();                    return product;    }    [HttpPut("{id}")]    public async Task<IActionResult> UpdateProduct(int id, Product product)    {        if (id != product.Id)            return BadRequest();                    await _productService.UpdateProductAsync(product);        return NoContent();    }}

三、应对缓存"三剑客"：穿透、击穿、雪崩

在实际生产环境中，仅仅实现基础缓存是不够的，我们还需要应对三个经典问题。

1. 缓存穿透：查询不存在的数据

问题：恶意请求查询数据库中不存在的数据，导致请求直接打到数据库。

解决方案：缓存空对象

public async Task<Product> GetProductByIdWithNullCacheAsync(int productId){    var cacheKey = $"product:{productId}";        var product = await _redisService.GetAsync<Product>(cacheKey);    if (product != null)    {        // 如果是特殊的空对象标记，返回null        if (product.Id == -1)            return null;                    return product;    }    product = await _productRepository.GetByIdAsync(productId);    if (product == null)    {        // 缓存空对象，设置较短的过期时间        var nullProduct = new Product { Id = -1 }; // 特殊标记        await _redisService.SetAsync(cacheKey, nullProduct, TimeSpan.FromMinutes(5));        return null;    }    await _redisService.SetAsync(cacheKey, product, TimeSpan.FromMinutes(30));    return product;}

2. 缓存击穿：热点Key突然失效

问题：某个热点Key在失效的瞬间，大量请求同时到达数据库。

解决方案：使用互斥锁

public async Task<Product> GetProductWithMutexAsync(int productId){    var cacheKey = $"product:{productId}";    var mutexKey = $"mutex:product:{productId}";        // 尝试获取缓存    var product = await _redisService.GetAsync<Product>(cacheKey);    if (product != null)        return product;    // 使用Redis实现分布式锁    var lockToken = Guid.NewGuid().ToString();    var locked = await _redisService.AcquireLockAsync(mutexKey, lockToken, TimeSpan.FromSeconds(5));        if (!locked)    {        // 获取锁失败，稍后重试        await Task.Delay(100);        return await GetProductWithMutexAsync(productId);    }    try    {        // 双重检查，防止重复查询数据库        product = await _redisService.GetAsync<Product>(cacheKey);        if (product != null)            return product;        // 查询数据库        product = await _productRepository.GetByIdAsync(productId);        if (product != null)        {            await _redisService.SetAsync(cacheKey, product, TimeSpan.FromMinutes(30));        }                return product;    }    finally    {        // 释放锁        await _redisService.ReleaseLockAsync(mutexKey, lockToken);    }}

3. 缓存雪崩：大量Key同时失效

问题：大量缓存Key在同一时间失效，导致所有请求直接访问数据库。

解决方案：设置不同的过期时间

public async Task SetWithRandomExpiryAsync<T>(string key, T value, TimeSpan baseExpiry){    // 在基础过期时间上增加随机偏差（±10%）    var random = new Random();    var variance = (int)(baseExpiry.TotalMinutes * 0.1); // 10% 偏差    var actualExpiry = baseExpiry.Add(TimeSpan.FromMinutes(random.Next(-variance, variance)));        await _redisService.SetAsync(key, value, actualExpiry);}

四、分布式锁：控制分布式环境下的资源访问

在分布式系统中，我们需要控制多个服务实例对共享资源的访问。

1. 扩展Redis服务支持分布式锁

public interface IRedisService{    // ... 其他方法        Task<bool> AcquireLockAsync(string key, string value, TimeSpan expiry);    Task<bool> ReleaseLockAsync(string key, string value);    Task<bool> ExtendLockAsync(string key, string value, TimeSpan expiry);}public class RedisService : IRedisService{    // ... 其他实现    public async Task<bool> AcquireLockAsync(string key, string value, TimeSpan expiry)    {        // 使用SET NX EX命令原子性地获取锁        return await _database.StringSetAsync(key, value, expiry, When.NotExists);    }    public async Task<bool> ReleaseLockAsync(string key, string value)    {        // 使用Lua脚本保证原子性：只有锁的值匹配时才删除        var luaScript = @"            if redis.call('GET', KEYS[1]) == ARGV[1] then                return redis.call('DEL', KEYS[1])            else                return 0            end";        var result = await _database.ScriptEvaluateAsync(luaScript, new RedisKey[] { key }, new RedisValue[] { value });        return (int)result == 1;    }    public async Task<bool> ExtendLockAsync(string key, string value, TimeSpan expiry)    {        var luaScript = @"            if redis.call('GET', KEYS[1]) == ARGV[1] then                return redis.call('EXPIRE', KEYS[1], ARGV[2])            else                return 0            end";        var result = await _database.ScriptEvaluateAsync(luaScript,             new RedisKey[] { key },             new RedisValue[] { value, (int)expiry.TotalSeconds });                    return (bool)result;    }}

2. 使用分布式锁实现秒杀功能

public class SeckillService{    private readonly IRedisService _redisService;    private readonly IOrderRepository _orderRepository;    public SeckillService(IRedisService redisService, IOrderRepository orderRepository)    {        _redisService = redisService;        _orderRepository = orderRepository;    }    public async Task<bool> ProcessSeckillAsync(int productId, int userId)    {        var lockKey = $"seckill:lock:{productId}";        var lockValue = Guid.NewGuid().ToString();        var stockKey = $"product:stock:{productId}";        try        {            // 获取分布式锁            var locked = await _redisService.AcquireLockAsync(lockKey, lockValue, TimeSpan.FromSeconds(10));            if (!locked)                return false; // 获取锁失败，稍后重试            // 检查库存            var stock = await _redisService.GetAsync<int>(stockKey);            if (stock <= 0)                return false;            // 扣减库存            await _redisService.SetAsync(stockKey, stock - 1);            // 创建订单            await _orderRepository.CreateAsync(new Order            {                ProductId = productId,                UserId = userId,                CreatedAt = DateTime.UtcNow            });            return true;        }        finally        {            // 释放锁            await _redisService.ReleaseLockAsync(lockKey, lockValue);        }    }}

五、会话存储：实现分布式Session

在微服务架构中，我们需要在多台服务器之间共享用户会话状态。

1. 配置Redis作为分布式Session存储

在Program.cs中：

// 添加Redis分布式缓存builder.Services.AddStackExchangeRedisCache(options =>{    options.Configuration = builder.Configuration.GetConnectionString("Redis");    options.InstanceName = "MyApp_";});// 配置Sessionbuilder.Services.AddSession(options =>{    options.IdleTimeout = TimeSpan.FromMinutes(30);    options.Cookie.HttpOnly = true;    options.Cookie.IsEssential = true;});

在Controller中使用：

public class AccountController : Controller{    [HttpPost]    public async Task<IActionResult> Login(LoginModel model)    {        // 验证用户...        var user = await AuthenticateUserAsync(model);        if (user == null)            return Unauthorized();        // 存储用户信息到Session        HttpContext.Session.SetString("UserId", user.Id.ToString());        HttpContext.Session.SetString("UserName", user.UserName);        HttpContext.Session.SetInt32("UserRole", (int)user.Role);        return RedirectToAction("Index", "Home");    }    [HttpGet]    public IActionResult GetUserInfo()    {        if (!HttpContext.Session.TryGetValue("UserId", out _))            return Unauthorized();        var userInfo = new        {            UserId = HttpContext.Session.GetString("UserId"),            UserName = HttpContext.Session.GetString("UserName"),            Role = HttpContext.Session.GetInt32("UserRole")        };        return Ok(userInfo);    }}

六、消息队列：实现异步任务处理

虽然Redis不是专业的消息队列，但对于简单的场景非常实用。

1. 基于List实现简单消息队列

public interface IMessageQueueService{    Task PublishAsync<T>(string queueName, T message);    Task<T> ConsumeAsync<T>(string queueName, TimeSpan? timeout = null);}public class RedisMessageQueueService : IMessageQueueService{    private readonly IConnectionMultiplexer _redis;    private readonly IDatabase _database;    public RedisMessageQueueService(IConnectionMultiplexer redis)    {        _redis = redis;        _database = redis.GetDatabase();    }    public async Task PublishAsync<T>(string queueName, T message)    {        var serializedMessage = JsonSerializer.Serialize(message);        await _database.ListLeftPushAsync(queueName, serializedMessage);    }    public async Task<T> ConsumeAsync<T>(string queueName, TimeSpan? timeout = null)    {        var value = await _database.ListRightPopAsync(queueName);        if (value.IsNullOrEmpty)        {            if (timeout.HasValue)            {                // 可以实现阻塞版本的消费                // 这里简化处理，返回默认值                await Task.Delay(timeout.Value);            }            return default;        }        return JsonSerializer.Deserialize<T>(value);    }}

2. 使用消息队列处理耗时任务

public class EmailService{    private readonly IMessageQueueService _messageQueue;    private readonly ILogger<EmailService> _logger;    public EmailService(IMessageQueueService messageQueue, ILogger<EmailService> logger)    {        _messageQueue = messageQueue;        _logger = logger;    }    // 发送邮件到队列（非阻塞）    public async Task SendWelcomeEmailAsync(string email, string userName)    {        var emailMessage = new EmailMessage        {            To = email,            Subject = "欢迎注册",            Body = $"亲爱的 {userName}，欢迎使用我们的服务！"        };        await _messageQueue.PublishAsync("email_queue", emailMessage);        _logger.LogInformation("欢迎邮件已加入队列，收件人: {Email}", email);    }}// 后台服务处理队列中的邮件public class EmailBackgroundService : BackgroundService{    private readonly IMessageQueueService _messageQueue;    private readonly IEmailSender _emailSender;    private readonly ILogger<EmailBackgroundService> _logger;    public EmailBackgroundService(IMessageQueueService messageQueue,                                  IEmailSender emailSender,                                 ILogger<EmailBackgroundService> logger)    {        _messageQueue = messageQueue;        _emailSender = emailSender;        _logger = logger;    }    protected override async Task ExecuteAsync(CancellationToken stoppingToken)    {        while (!stoppingToken.IsCancellationRequested)        {            try            {                var message = await _messageQueue.ConsumeAsync<EmailMessage>("email_queue");                if (message != null)                {                    await _emailSender.SendEmailAsync(message.To, message.Subject, message.Body);                    _logger.LogInformation("邮件发送成功: {To}", message.To);                }                else                {                    // 队列为空，等待一段时间                    await Task.Delay(1000, stoppingToken);                }            }            catch (Exception ex)            {                _logger.LogError(ex, "处理邮件队列时发生错误");                await Task.Delay(5000, stoppingToken); // 错误时等待更长时间            }        }    }}

七、性能优化与最佳实践

1. 连接复用

确保在整个应用程序中复用IConnectionMultiplexer实例：

// 在Program.cs中注册为Singletonbuilder.Services.AddSingleton<IConnectionMultiplexer>(sp =>     ConnectionMultiplexer.Connect(builder.Configuration.GetConnectionString("Redis")));

2. 使用Pipeline批量操作

public async Task<bool> SetMultipleAsync(Dictionary<string, object> keyValuePairs, TimeSpan? expiry = null){    var batch = _database.CreateBatch();        var tasks = new List<Task>();    foreach (var kvp in keyValuePairs)    {        var serializedValue = JsonSerializer.Serialize(kvp.Value);        tasks.Add(batch.StringSetAsync(kvp.Key, serializedValue, expiry));    }        batch.Execute();    await Task.WhenAll(tasks);        return tasks.All(t => t.IsCompletedSuccessfully);}

3. 合理的序列化选择

// 对于简单类型，考虑使用更高效的序列化方式public async Task SetStringAsync(string key, string value, TimeSpan? expiry = null){    // 对于字符串，直接存储，避免JSON序列化开销    await _database.StringSetAsync(key, value, expiry);}public async Task<string> GetStringAsync(string key){    return await _database.StringGetAsync(key);}

总结

通过本篇的实战演练，我们掌握了在Asp.Net Core项目中集成和使用Redis的完整方案：

1. 环境搭建：配置StackExchange.Redis客户端
2. 缓存应用：商品信息缓存及缓存策略
3. 问题解决：应对缓存穿透、击穿、雪崩的完整方案
4. 分布式锁：实现秒杀等并发控制场景
5. 会话存储：配置分布式Session
6. 消息队列：实现异步任务处理
7. 性能优化：连接复用、批量操作等最佳实践

关键收获：

• Redis在Asp.Net Core中的集成非常简单直接
• 合理使用缓存可以大幅提升系统性能
• 分布式锁是解决并发问题的利器
• 选择合适的序列化方式对性能有重要影响

现在，你可以自信地在你的Asp.Net Core项目中使用Redis来解决实际的性能瓶颈和分布式协调问题了！

欢迎在评论区分享你在集成过程中遇到的问题和解决方案！

Redis的持久化与高可用：如何避免"内存失忆"？

在之前的文章中，我们领略了Redis基于内存的极速性能。但这也引出了一个关键问题：如果服务器重启或宕机，内存中的数据岂不是会全部丢失？ 今天，我们就来深入探讨Redis如何解决这个"阿喀琉斯之踵"，以及如何构建高可用的Redis架构。

一、数据持久化：为什么需要"记忆备份"？

想象一下，Redis就像一个拥有"超强记忆力"的天才，但它的记忆只存在于脑海中（内存）。一旦受到冲击（重启/宕机），所有记忆都会消失。为了避免这种"失忆"悲剧，Redis提供了两种"记忆备份"机制：RDB和AOF。

持久化的本质：将内存中的数据以某种形式保存到磁盘中，确保在服务重启后能够恢复数据。

二、RDB持久化：给数据拍"快照"

1. 工作原理

RDB（Redis DataBase）的机制很简单：在特定时间点，将内存中所有数据生成一个快照文件保存到磁盘。这个文件通常以.rdb为后缀。

你可以把它理解为给整个数据库拍一张全家福，照片记录了那个瞬间的所有数据状态。

2. 触发机制

RDB有三种主要的触发方式：

自动触发（配置策略）

在redis.conf配置文件中，我们可以设置自动触发快照的条件：

# 在900秒（15分钟）内，如果至少有1个key发生变化，则触发bgsavesave 900 1# 在300秒（5分钟）内，如果至少有10个key发生变化，则触发bgsave  save 300 10# 在60秒内，如果至少有10000个key发生变化，则触发bgsavesave 60 10000

手动触发

# 1. SAVE命令（同步）- 阻塞主进程，直到快照完成，期间不处理任何请求127.0.0.1:6379> SAVEOK# 2. BGSAVE命令（异步）- 后台执行快照，主进程继续处理请求127.0.0.1:6379> BGSAVEBackground saving started

其他情况

• 执行SHUTDOWN命令关闭Redis时，如果没有开启AOF，会自动执行RDB快照
• 主从复制时，主节点会向从节点发送RDB文件进行全量同步

3. RDB的工作流程（BGSAVE）

当我们执行BGSAVE时，Redis会：

1. Fork子进程：主进程创建一个子进程（copy-on-write机制，内存占用不会翻倍）
2. 子进程写盘：子进程将内存数据写入临时RDB文件
3. 替换旧文件：写入完成后，用新的RDB文件替换旧的
4. 清理工作：子进程退出，通知主进程完成

4. 优缺点分析

优点：

• ✅ 性能影响小：BGSAVE通过子进程操作，主进程几乎不受影响
• ✅ 文件紧凑：二进制格式，文件较小，适合备份和传输
• ✅ 恢复速度快：恢复大数据集时比AOF快很多

缺点：

• ❌ 可能丢失数据：两次快照之间的数据修改会丢失
• ❌ Fork可能阻塞：数据集很大时，fork操作本身可能耗时较长

三、AOF持久化：记录每一个"成长瞬间"

1. 工作原理

AOF（Append Only File）采用了一种完全不同的思路：记录每一个写操作命令，以日志的形式追加到文件末尾。

这就像是写日记，记录下每一天发生的事情，而不是只拍几张照片。

2. 配置详解

要开启AOF，需要在配置文件中设置：

# 开启AOF持久化appendonly yes# AOF文件名appendfilename "appendonly.aof"# 同步策略appendfsync everysec

3. AOF的三种同步策略

生产环境推荐使用everysec，在性能和数据安全之间取得最佳平衡。

4. AOF重写机制

随着运行时间增长，AOF文件会越来越大，而且包含很多已经过期的命令（比如对同一个key的多次set）。为了解决这个问题，Redis提供了AOF重写机制。

重写的本质：基于当前内存数据，生成一个新的、更精简的AOF文件，只包含恢复当前数据所需的最小命令集合。

# 手动触发AOF重写127.0.0.1:6379> BGREWRITEAOFBackground append only file rewriting started

自动重写配置：

# 当AOF文件体积比上次重写后体积增长100%时，自动触发重写auto-aof-rewrite-percentage 100# AOF文件体积至少达到64MB时才会触发重写auto-aof-rewrite-min-size 64mb

5. 优缺点分析

优点：

• ✅ 数据安全：配置合理时最多丢失1秒数据
• ✅ 可读性强：AOF文件是文本格式，可以人工阅读和修改
• ✅ 容错性好：即使文件尾部有损坏，也可以用redis-check-aof工具修复

缺点：

• ❌ 文件较大：通常比RDB文件大
• ❌ 恢复速度慢：需要重新执行所有命令，恢复大数据集时较慢
• ❌ 性能影响：在高负载下，AOF可能比RDB稍慢

四、RDB vs AOF：如何选择？

生产环境推荐策略

两者结合使用，发挥各自优势：

# 在redis.conf中同时开启RDB和AOFsave 900 1save 300 10save 60 10000appendonly yesappendfsync everysecauto-aof-rewrite-percentage 100auto-aof-rewrite-min-size 64mb

这种组合的优势：

• AOF保证数据安全性，最多丢失1秒数据
• RDB用于冷备份、快速重启和主从同步
• 重启时优先使用AOF恢复（数据更完整），其次使用RDB

五、主从复制：数据备份与读写分离

单机Redis存在单点故障风险，主从复制是构建高可用架构的第一步。

1. 什么是主从复制？

• 主节点（Master）：负责写操作，将数据变化同步给从节点
• 从节点（Slave/Replica）：复制主节点数据，负责读操作

2. 主从复制的工作原理

1. 建立连接：从节点连接到主节点，发送SYNC命令
2. 全量同步：主节点执行BGSAVE生成RDB文件，发送给从节点
3. 增量同步：主节点将期间的写命令缓存起来，RDB传输完成后发送给从节点
4. 命令传播：之后主节点每收到写命令，就异步发送给从节点

3. 如何配置主从复制？

假设我们有：

• 主节点：127.0.0.1:6379
• 从节点：127.0.0.1:6380

方法一：配置文件
在从节点的redis.conf中添加：

replicaof 127.0.0.1 6379# 或者老版本使用：slaveof 127.0.0.1 6379

方法二：运行时命令

# 在从节点上执行127.0.0.1:6380> REPLICAOF 127.0.0.1 6379OK

4. 验证主从状态

# 在主节点查看复制信息127.0.0.1:6379> INFO replication# Replicationrole:masterconnected_slaves:1slave0:ip=127.0.0.1,port=6380,state=online,offset=1234,lag=0# 在从节点查看复制信息  127.0.0.1:6380> INFO replication# Replicationrole:slavemaster_host:127.0.0.1master_port:6379master_link_status:up

5. 主从架构的优势

1. 数据冗余：从节点是主节点的完整备份
2. 读写分离：主节点负责写，从节点负责读，提升读性能
3. 故障恢复基础：为自动故障转移做准备

六、哨兵（Sentinel）模式：实现自动故障转移

主从复制解决了数据备份问题，但如果主节点宕机，需要手动切换，这期间服务会不可用。哨兵模式就是为了解决这个问题。

1. 哨兵是什么？

Redis Sentinel是一个分布式系统，用于管理多个Redis实例，主要功能包括：

• 监控：持续检查主从节点是否正常运行
• 通知：当被监控的Redis实例出现问题时，向管理员发送告警
• 自动故障转移：主节点故障时，自动将一个从节点提升为新主节点，并让其他从节点复制新主节点
• 配置提供者：客户端连接哨兵获取当前的主节点地址

2. 哨兵集群架构

通常我们会部署奇数个哨兵实例（如3个或5个），通过投票机制来决定是否进行故障转移，避免误判。

3. 搭建哨兵模式

假设我们有：

• Redis主节点：127.0.0.1:6379
• Redis从节点：127.0.0.1:6380、127.0.0.1:6381
• 哨兵节点：127.0.0.1:26379、127.0.0.1:26380、127.0.0.1:26381

创建哨兵配置文件 sentinel-26379.conf：

port 26379sentinel monitor mymaster 127.0.0.1 6379 2sentinel down-after-milliseconds mymaster 5000sentinel failover-timeout mymaster 60000sentinel parallel-syncs mymaster 1

参数解释：

• sentinel monitor mymaster 127.0.0.1 6379 2：监控名为mymaster的主节点，至少需要2个哨兵同意才判定主观下线
• down-after-milliseconds：5000毫秒无响应认为节点主观下线
• failover-timeout：故障转移超时时间
• parallel-syncs：故障转移后，同时向新主节点同步的从节点数量

启动哨兵：

redis-sentinel sentinel-26379.confredis-sentinel sentinel-26380.conf  redis-sentinel sentinel-26381.conf

4. 故障转移过程

1. 主观下线：某个哨兵认为主节点不可用
2. 客观下线：多个哨兵（达到quorum数量）都认为主节点不可用
3. 选举领导者：哨兵之间选举一个领导者来执行故障转移
4. 故障转移：领导者哨兵选择一个合适的从节点提升为新主节点
5. 切换配置：通知其他从节点复制新主节点，更新客户端配置

5. 客户端如何连接哨兵？

客户端不再直接连接Redis节点，而是连接哨兵集群来获取当前的主节点地址。

Java客户端示例（使用Jedis）：

Set<String> sentinels = new HashSet<>();sentinels.add("127.0.0.1:26379");sentinels.add("127.0.0.1:26380"); sentinels.add("127.0.0.1:26381");JedisSentinelPool pool = new JedisSentinelPool("mymaster", sentinels);try (Jedis jedis = pool.getResource()) {    // 现在操作的是当前的主节点    jedis.set("key", "value");}

七、持久化与高可用配置总结

总结

通过本篇的学习，我们掌握了构建可靠Redis系统的核心技术：

1. 持久化是基础：RDB提供快照备份，AOF保证命令不丢失，两者结合使用最稳妥
2. 主从复制提供冗余：数据多副本，读写分离提升性能
3. 哨兵实现高可用：自动故障转移，服务不中断

记住这个演进路径：
单机Redis → 开启持久化 → 搭建主从复制 → 部署哨兵集群

现在，你的Redis已经不再是那个"内存失忆"的脆弱系统，而是一个具备数据持久化和自动故障恢复能力的高可用服务！

思考与实践：

1. 在你的开发环境中尝试配置RDB和AOF，观察文件生成情况
2. 搭建一个一主二从的Redis环境，并验证数据同步
3. 部署三节点哨兵集群，模拟主节点宕机，观察自动故障转移过程

欢迎在评论区分享你在配置过程中遇到的问题和解决方案！

玩转Redis五大核心数据结构：从计数器到排行榜

在上一篇中，我们知道了Redis为什么这么快，以及如何搭建环境。但Redis真正的威力，来自于它丰富的数据结构。如果说简单的键值对是Redis的"骨架"，那么这些数据结构就是它的"肌肉"，让Redis能够优雅地解决各种复杂的业务场景。

Redis最吸引人的地方在于，它不仅仅是一个简单的键值存储，而是一个数据结构服务器。今天，我们将深入探索Redis的五大核心数据结构：String（字符串）、Hash（哈希）、List（列表）、Set（集合） 和 Sorted Set（有序集合）。

一、String（字符串）：不止是文本

String是Redis最基本的数据类型，一个Key对应一个Value。但别被它的名字骗了——它不仅可以存储文本，还可以存储数字（整数或浮点数）甚至是二进制数据（如图片或序列化对象）。

核心命令与特性

# 1. 基础设置与获取127.0.0.1:6379> SET username "redis_learner"OK127.0.0.1:6379> GET username"redis_learner"# 2. 数字操作 - Redis知道你是数字时会允许数学运算127.0.0.1:6379> SET page_views 100OK127.0.0.1:6379> INCR page_views        # 增加1(integer) 101127.0.0.1:6379> INCRBY page_views 10   # 增加指定数值(integer) 111127.0.0.1:6379> DECR page_views        # 减少1(integer) 110# 3. 批量操作 - 提升效率127.0.0.1:6379> MSET user:1001:name "Alice" user:1001:age 25 user:1001:city "Beijing"OK127.0.0.1:6379> MGET user:1001:name user:1001:age user:1001:city1) "Alice"2) "25"3) "Beijing"# 4. 条件设置 - 实现分布式锁的基础127.0.0.1:6379> SETNX lock:order_123 "client_1"  # 只有当key不存在时才设置(integer) 1  # 设置成功127.0.0.1:6379> SETNX lock:order_123 "client_2"(integer) 0  # 设置失败，因为key已存在

实战应用场景

1. 缓存：存储序列化的用户信息、页面片段等
2. 计数器：文章阅读量、用户点赞数、网站访问量
3. 分布式锁：通过SETNX实现简单的互斥锁
4. 会话存储：存储用户Session数据

性能提示：对于多个相关的键值对，使用MSET/MGET比多次SET/GET更高效，因为它减少了网络往返次数。

二、Hash（哈希表）：存储对象的最佳选择

如果你需要存储一个对象（如用户信息、商品信息），Hash是你的最佳选择。它类似于编程语言中的字典或Map，适合存储字段-值对的集合。

为什么用Hash而不用多个String？

假设我们要存储用户信息，有两种方案：

• 方案一（多个String）：

  SET user:1001:name "Bob"SET user:1001:age 30SET user:1001:email "bob@example.com"

• 方案二（一个Hash）：

  HSET user:1001 name "Bob" age 30 email "bob@example.com"

Hash的优势：

• 内存效率更高：Redis对Hash有特殊优化，特别是字段较少时
• 原子性操作：可以一次性获取或修改整个对象
• 更少的Key：避免键空间膨胀，管理更方便

核心命令详解

# 1. 设置和获取字段127.0.0.1:6379> HSET product:1001 name "iPhone 15" price 5999 stock 100(integer) 3  # 返回设置的字段数量127.0.0.1:6379> HGET product:1001 name"iPhone 15"127.0.0.1:6379> HGETALL product:1001  # 获取所有字段和值1) "name"2) "iPhone 15"3) "price"4) "5999"5) "stock"6) "100"# 2. 批量操作127.0.0.1:6379> HMGET product:1001 name price  # 获取多个字段1) "iPhone 15"2) "5999"# 3. 数字运算127.0.0.1:6379> HINCRBY product:1001 stock -1  # 库存减1（售出一件）(integer) 99# 4. 检查字段127.0.0.1:6379> HEXISTS product:1001 name(integer) 1127.0.0.1:6379> HKEYS product:1001  # 获取所有字段名1) "name"2) "price"3) "stock"

实战应用场景

1. 用户信息存储：将用户的所有属性存储在一个Hash中
2. 商品信息：商品的名称、价格、库存等信息
3. 配置信息：系统的各种配置参数
4. 购物车：用户ID作为Key，商品ID和数量作为字段-值对

三、List（列表）：实现简单的消息队列

List是一个按插入顺序排序的字符串元素集合，你可以在列表的头部（左边）或尾部（右边）添加元素。Redis的List底层实现是双向链表，这意味着在头部和尾部添加元素的速度极快，但通过索引访问中间元素相对较慢。

核心命令详解

# 1. 从两端添加元素127.0.0.1:6379> LPUSH tasks "send_email"      # 从左边添加(integer) 1127.0.0.1:6379> LPUSH tasks "process_image"(integer) 2127.0.0.1:6379> RPUSH tasks "generate_report" # 从右边添加(integer) 3# 此时列表：["process_image", "send_email", "generate_report"]#               头(左)   <--------->   尾(右)# 2. 从两端弹出元素127.0.0.1:6379> LPOP tasks  # 从左边弹出"process_image"127.0.0.1:6379> RPOP tasks  # 从右边弹出"generate_report"# 3. 查看列表范围（不会弹出元素）127.0.0.1:6379> LRANGE tasks 0 -1  # 查看所有元素，0表示开始，-1表示末尾1) "send_email"# 4. 阻塞操作 - 消息队列的核心# 从一个空列表中阻塞地等待元素，最多等待10秒127.0.0.1:6379> BLPOP message_queue 10(nil)  # 10秒内没有元素，返回nil# 在另一个客户端执行：LPUSH message_queue "new_message"# 此时BLPOP会立即返回："message_queue" "new_message"

实战应用场景

1. 消息队列：使用LPUSH添加任务，BRPOP阻塞获取任务
2. 最新消息列表：使用LPUSH添加新消息，LRANGE 0 9获取最新的10条
3. 历史记录：用户浏览历史、搜索历史
4. 文章评论列表：文章的评论按时间顺序排列

重要特性：List的阻塞操作(BLPOP, BRPOP)使其成为实现简单消息队列的理想选择，消费者可以在队列为空时等待，而不需要轮询。

四、Set（集合）：无序与唯一性的力量

Set是String类型的无序集合，它最大的特点是：元素唯一且无序。底层通过哈希表实现，添加、删除、查找的时间复杂度都是O(1)。

核心命令与集合运算

# 1. 基本操作127.0.0.1:6379> SADD tags "java" "python" "redis" "java"(integer) 3  # "java"重复，只添加了3个元素127.0.0.1:6379> SMEMBERS tags  # 获取所有元素（顺序不确定）1) "redis"2) "python"3) "java"127.0.0.1:6379> SISMEMBER tags "python"  # 检查元素是否存在(integer) 1# 2. 集合运算 - Set的精华所在127.0.0.1:6379> SADD user:1001:follows "user:1002" "user:1003" "user:1004"(integer) 3127.0.0.1:6379> SADD user:1002:follows "user:1003" "user:1005"(integer) 2# 交集 - 共同关注127.0.0.1:6379> SINTER user:1001:follows user:1002:follows1) "user:1003"# 并集 - 所有的关注127.0.0.1:6379> SUNION user:1001:follows user:1002:follows1) "user:1002"2) "user:1003"3) "user:1004"4) "user:1005"# 差集 - A有但B没有的127.0.0.1:6379> SDIFF user:1001:follows user:1002:follows1) "user:1002"2) "user:1004"# 3. 随机元素 - 抽奖功能127.0.0.1:6379> SADD lottery_users "user1" "user2" "user3" "user4" "user5"(integer) 5127.0.0.1:6379> SRANDMEMBER lottery_users 2  # 随机返回2个元素，不删除1) "user3"2) "user5"127.0.0.1:6379> SPOP lottery_users 1         # 随机弹出1个元素并删除1) "user2"

实战应用场景

1. 标签系统：给文章、用户打标签
2. 社交关系：共同好友、共同关注
3. 数据去重：防止重复提交、重复处理
4. 随机抽奖：从参与用户中随机抽取中奖者
5. 黑白名单：IP白名单、用户黑名单

五、Sorted Set（有序集合）：排行榜的灵魂

Sorted Set是Set的增强版，它在保证元素唯一性的基础上，为每个元素关联了一个分数（Score），元素按照分数进行排序。这是Redis中最复杂但也最强大的数据结构之一。

底层实现：跳表（Skip List）

Sorted Set使用跳表（一种类似链表但有多级索引的数据结构）实现，可以在O(logN)时间内完成插入、删除和按分数范围查找，兼具了链表和二分查找的优点。

核心命令详解

# 1. 添加元素（带分数）127.0.0.1:6379> ZADD leaderboard 2500 "Alice" 1800 "Bob" 3200 "Charlie" 1500 "David"(integer) 4# 2. 按分数范围查询（升序）127.0.0.1:6379> ZRANGE leaderboard 0 -1 WITHSCORES  # 获取所有元素（分数从低到高）1) "David"2) "1500"3) "Bob"4) "1800"5) "Alice"6) "2500"7) "Charlie"8) "3200"# 3. 按分数范围查询（降序 - 排行榜常用）127.0.0.1:6379> ZREVRANGE leaderboard 0 2 WITHSCORES  # 获取前三名1) "Charlie"2) "3200"3) "Alice"4) "2500"5) "Bob"6) "1800"# 4. 按分数范围查询127.0.0.1:6379> ZRANGEBYSCORE leaderboard 2000 3000 WITHSCORES  # 分数在2000-3000之间的玩家1) "Alice"2) "2500"# 5. 获取排名和分数127.0.0.1:6379> ZRANK leaderboard "Alice"    # 获取升序排名（从0开始）(integer) 2127.0.0.1:6379> ZREVRANK leaderboard "Alice" # 获取降序排名（排行榜名次）(integer) 1127.0.0.1:6379> ZSCORE leaderboard "Alice"   # 获取分数"2500"# 6. 分数操作127.0.0.1:6379> ZINCRBY leaderboard 500 "Alice"  # Alice增加500分"3000"

实战应用场景

1. 排行榜：游戏积分榜、销量排行榜、热搜榜
2. 带权重的队列：优先级任务调度
3. 时间轴：按时间排序的消息列表（时间戳作为Score）
4. 范围查询：查找分数/价格在某个区间的数据

性能提示：Sorted Set的范围查询（ZRANGEBYSCORE）非常高效，特别适合需要按范围检索数据的场景。

数据结构选择指南

面对具体业务场景时，如何选择合适的数据结构？这里有一个快速参考：

总结

通过本篇的学习，我们已经掌握了Redis五大核心数据结构的特性和应用：

• String：简单但强大，支持数字操作
• Hash：存储对象的理想选择，内存效率高
• List：顺序数据结构，适合消息队列和时间线
• Set：无序唯一集合，强大的集合运算能力
• Sorted Set：有序唯一集合，排行榜和范围查询的利器

Redis的哲学是：将复杂的数据操作下推到存储层，而不是在应用层处理。理解每种数据结构的特性和适用场景，能够让你在设计系统时做出更优雅、更高效的决策。

现在，当你需要实现计数器时，不会选择在应用层读取-计算-保存，而是直接使用INCR命令；当你需要排行榜时，不会在数据库中排序，而是使用Sorted Set。这就是Redis数据结构的威力所在！

动手练习：尝试用你学到的数据结构实现以下功能：

1. 使用String实现一个文章阅读量计数器
2. 使用Hash存储你的个人简历信息
3. 使用List实现一个简单的待办事项列表
4. 使用Set找出你和你朋友共同喜欢的电影
5. 使用Sorted Set创建一个游戏分数排行榜

欢迎在评论区分享你的实现代码和心得体会！

基础题

1. 主键、外键、超键、候选键的区别和用途

• 超键：在关系中能唯一标识元组的属性集称为关系模式的超键。一个属性可以作为一个超键，多个属性组合在一起也可以作为一个超键。超键包含候选键和主键。
• 候选键：是最小超键，即没有冗余元素的超键。
• 主键：数据库表中对储存数据对象予以唯一和完整标识的数据列或属性的组合。一个数据列只能有一个主键，且主键的取值不能缺失，即不能为空值（Null）。
• 外键：在一个表中存在的另一个表的主键称此表的外键。

2. 为什么使用自增列作为主键？

自增列作为主键可以简化数据的插入操作，避免因插入非顺序的主键值导致的索引分裂和碎片化，从而提高数据库性能。自增列也易于分配和管理，且不会与其他记录的主键冲突。

3. 触发器的作用是什么？

触发器是一种特殊的存储过程，它在特定数据库操作（如INSERT、UPDATE、DELETE）执行之前或之后自动触发执行。触发器可以用于维护数据完整性、实施复杂的业务规则、自动更新表中的数据等。

4. 什么是存储过程？使用什么来调用？

存储过程是一组为了执行特定任务而预编译的SQL语句。它们可以提高性能，因为只需编译一次，之后可以重复调用。存储过程可以通过SQL命令直接调用，也可以被应用程序通过特定的API调用来执行。

5. 存储过程的优缺点有哪些？

存储过程的优点包括提高性能（预编译）、减少网络传输、增强安全性（需要特定权限才能执行）、便于代码复用。缺点包括移植性差，因为它们通常与特定的数据库系统紧密相关。

6. 存储过程与函数的区别是什么？

存储过程是一系列为了完成特定功能的SQL语句集合，可以通过参数传递数据，并且可以有多个返回值。函数通常返回一个单一的数据值，并且在使用时作为表达式的一部分。存储过程使用更灵活，而函数则更适用于需要返回特定数据结构的场景。

7. 视图是什么？游标是什么？

视图是基于SQL查询的虚拟表，它像实际的表一样可以进行查询和更新操作，但是不存储数据，而是在查询视图时动态生成结果。游标是一种数据库对象，用于逐行处理查询结果集，常用于需要对结果集进行循环处理的场景。

8. 视图的优缺点有哪些？

视图的优点包括简化复杂的查询、提高数据安全性、实现数据逻辑抽象。缺点包括可能影响性能（尤其是在复杂的视图上执行查询时），以及在某些情况下限制了数据的更新操作。

9. 什么是临时表？临时表什么时候删除？

临时表是在当前会话或事务中创建的表，仅对当前会话可见。当会话结束或事务提交时，临时表及其数据会自动删除。

10. 非关系型数据库和关系型数据库的区别和优势比较是什么？

非关系型数据库（NoSQL）和关系型数据库在数据模型、查询方式、扩展性等方面有本质区别。非关系型数据库通常提供更高的扩展性和灵活性，适合处理大规模分布式数据。关系型数据库则在数据一致性、复杂查询和事务管理方面表现更好。

11. 数据库范式是什么？如何根据某个场景设计数据表？

数据库范式是一套用于指导数据库设计的规范，包括第一范式（1NF）、第二范式（2NF）、第三范式（3NF）等，目的是减少数据冗余和提高数据完整性。设计数据表时，应根据业务需求和数据关系来确定表结构，确保满足相应的范式要求。

12. 内连接、外连接、交叉连接、笛卡尔积等的区别是什么？

内连接只返回两个表中匹配的行；外连接（左外连接、右外连接）会返回一个表的全部行，另一个表中匹配的行，不匹配的行用NULL填充；交叉连接返回两个表的笛卡尔积，即每行与另一个表中每行的组合；笛卡尔积是两个集合所有可能的组合。

13. varchar和char的使用场景是什么？

VARCHAR适用于长度可变的数据，如用户输入的评论或描述，因为它可以根据实际内容长度存储，节省空间。CHAR适用于长度固定的数据，如性别或国家代码，因为它可以提供更快的存取速度，但会使用固定长度的存储空间。

14. SQL语言分类有哪些？

SQL语言主要分为数据查询语言（DQL），数据操纵语言（DML），数据定义语言（DDL）和数据控制语言（DCL）。DQL用于查询数据，如SELECT；DML用于数据的增删改，如INSERT、UPDATE、DELETE；DDL用于数据库对象的定义，如CREATE、ALTER、DROP；DCL用于控制数据库访问权限，如GRANT、REVOKE。

15. like '%xxx%'和’xxx%'的区别是什么？

LIKE '%xxx%'表示匹配包含xxx的任意字符串，无论xxx出现在哪一部分。LIKE 'xxx%'表示匹配以xxx结尾的字符串。两者在模糊匹配时使用不同的通配符，%代表任意字符出现任意次数，而_仅代表单个字符。

16. count(*)、count(1)、count(column)的区别是什么？

COUNT()用于计算表中的总行数，包括NULL值。COUNT(1)是COUNT()的等价操作，用于计算行数。COUNT(column)用于计算特定列中非NULL值的数量。

17. 最左前缀原则是什么？

最左前缀原则是索引创建和使用的一个重要原则，它指的是在多列索引中，数据库查询优化器只会使用索引的最左部分列。这意味着如果查询条件没有使用到索引的第一个列，那么即使后面的列被使用到，索引也可能不会被利用。

18. 索引的作用是什么？它的优点和缺点有哪些？

索引的作用是加快数据检索速度，排序和分组数据，以及保证数据的唯一性。优点包括提高查询速度、加速表连接、支持数据的排序和分组。缺点包括增加存储空间、降低数据更新（INSERT、UPDATE、DELETE）的速度，以及维护索引本身需要额外的开销。

19. 什么样的字段适合建索引？

适合建索引的字段包括经常需要搜索的列、作为主键的列、经常用于连接的列、经常需要进行范围搜索的列、经常需要排序的列，以及经常使用在WHERE子句中的列。

20. 聚集索引和非聚集索引的区别是什么？

聚集索引决定了表中数据的物理存储顺序，使得相关列的数据在物理上连续存放，查询效率较高，但修改数据时可能较慢。非聚集索引指定了表中数据的逻辑顺序，但物理存储顺序与索引可能不一致，通常用于频繁更新的数据列。

21. SQL注入式攻击是什么？

SQL注入式攻击是一种网络安全攻击手段，攻击者通过在Web表单输入域或页面请求的查询字符串中插入恶意SQL命令，欺骗服务器执行这些命令，从而获取、篡改或删除数据库中的数据。

22. 如何防范SQL注入式攻击？

防范SQL注入式攻击的方法包括：对用户输入进行过滤和验证，替换或转义特殊字符；使用预处理语句（参数化查询）；限制数据库权限，使用最小权限原则；使用存储过程；以及在服务器端进行输入验证等。

23. 内存泄漏是什么？

内存泄漏是指在程序运行过程中，由于未能适当释放不再使用的内存，导致随着程序的持续运行，可用内存逐渐减少的现象。在动态内存分配的语言中，如C或C++，如果使用new分配了内存，却忘记使用delete释放，就可能发生内存泄漏。

24. 维护数据库的完整性和一致性，使用触发器还是自写业务逻辑？

维护数据库的完整性和一致性，通常首选使用数据库提供的约束，如CHECK、PRIMARY KEY、FOREIGN KEY等。其次是使用触发器，因为它们可以自动执行，确保数据的完整性和一致性，无论哪种业务逻辑访问数据库。最后考虑自写业务逻辑，但这种方法编程复杂，效率较低。

25. 什么是事务？什么是锁？

事务是一系列操作，它们作为一个整体被执行，以确保数据的完整性。如果事务中的任何操作失败，整个事务将回滚到执行前的状态。锁是数据库管理系统用来保证事务的隔离性和并发控制的一种机制，它可以防止多个事务同时修改同一数据，从而避免数据冲突。

26. 过多索引对数据库性能的影响

过多的索引虽然可以提高查询速度，但在数据的插入、更新和删除操作时，数据库引擎需要更多的时间来维护这些索引，这可能会导致性能下降。因此，需要在索引创建时进行权衡，以确保数据库操作的整体性能。

27. 相关子查询是什么？如何使用这些查询？

相关子查询是一种特殊类型的子查询，它在查询中使用外部查询的值。这种子查询通常用于WHERE或HAVING子句中，可以基于外部查询的结果来动态地定义查询条件。

28. 操作会使⽤到TempDB

TempDB是SQL Server的一个系统数据库，用于存储临时数据，如临时表和表变量。许多操作，包括创建表时的临时数据、执行某些类型的JOIN操作、使用游标以及存储过程和批处理中的一些操作，都可能会用到TempDB。

29. 如果TempDB异常变大，可能的原因是什么，该如何处理？

TempDB异常变大可能是由于大量使用临时表或返回的记录集过大造成的。处理方法包括优化查询以减少返回的数据量，使用分批处理，或者调整TempDB的大小和配置。

30. Index有哪些类型，它们的区别和实现原理是什么，索引有什么优点和缺点

索引类型主要包括聚集索引和非聚集索引。聚集索引决定了表中数据的物理存储顺序，非聚集索引则不改变数据的物理存储顺序。索引的优点包括提高查询速度、确保数据的唯一性和排序。缺点是增加了存储空间和维护成本，降低了数据更新的速度。

31. Job信息可以通过哪些表获取；系统正在运行的语句可以通过哪些视图获取；如何获取某个T-SQL语句的IO、Time等信息

Job信息可以通过SQL Server的msdb数据库中的表，如sysjobs和sysjobhistory获取。系统正在运行的语句可以通过动态管理视图如sys.dm_exec_requests获取。要获取某个T-SQL语句的IO和Time等信息，可以使用SQL Server Profiler或相关的动态管理视图。

确保字段只接受特定范围内的值
可以通过在字段上设置CHECK约束来确保只接受特定范围内的值。CHECK约束允许定义字段值的范围或条件，确保插入或更新数据时满足这些条件。

32. CHAR、VARCHAR、NCHAR 和 NVARCHAR 的区别是什么？

• CHAR(n)： 固定长度，非 Unicode 字符数据。无论实际内容多长，它都会占用 n 个字节的存储空间。适合存储长度相对固定的数据（如身份证号、电话号码）。
• VARCHAR(n)： 可变长度，非 Unicode 字符数据。它只占用实际数据长度 + 2 个字节（用于存储长度信息）的存储空间。适合存储长度变化较大的数据。
• NCHAR(n)： 固定长度，Unicode 字符数据。存储 Unicode 字符（如中文、日文等），每个字符占用 2 个字节。长度为 n，表示最多可存储 n 个字符（无论中英文）。
• NVARCHAR(n)： 可变长度，Unicode 字符数据。同样存储 Unicode 字符，每个字符 2 字节，但只占用（实际字符数 * 2） + 2 字节的空间。

核心区别： CHAR/VARCHAR 用于非 Unicode，一个英文字符占1字节，一个中文字符可能占2字节（取决于编码）。NCHAR/NVARCHAR 用于 Unicode，任何字符都占2字节，能全球通用。

33. TRUNCATE、DELETE 和 DROP 的区别？

34. 什么是索引？聚集索引和非聚集索引的区别？

• 索引：相当于书籍的目录，它能帮助数据库引擎快速找到数据，而无需扫描整个表。
• 聚集索引：
  • 决定了表中数据的物理存储顺序。一张表只能有一个聚集索引。
  • 叶子节点存储的是实际的数据行。
  • 例如，在主键上默认创建的通常是聚集索引。
• 非聚集索引：
  • 不影响数据的物理存储顺序。一张表可以有多个非聚集索引。
  • 叶子节点存储的是索引键值 + 指向数据行的指针（聚集索引键或RID）。
  • 查询时需要先查非聚集索引，再通过指针去查找实际数据，这个过程称为 “键查找” 或 “书签查找”。

35. 内连接（INNER JOIN）和外连接（OUTER JOIN）的区别？

• 内连接：返回两个表中连接条件匹配的所有行。不匹配的行不会出现在结果中。
• 外连接：
  • 左外连接（LEFT JOIN）：返回左表的所有行，以及右表中连接条件匹配的行。如果右表无匹配，则右表部分为 NULL。
  • 右外连接（RIGHT JOIN）：返回右表的所有行，以及左表中连接条件匹配的行。如果左表无匹配，则左表部分为 NULL。
  • 全外连接（FULL JOIN）：返回左表和右表中的所有行。当某一行在另一个表中没有匹配时，另一个表的部分为 NULL。

36. 什么是执行计划？如何查看和分析？

• 执行计划：是 SQL Server 查询优化器生成的、关于如何执行一个查询的“路线图”。它显示了数据获取的步骤、使用的索引、连接类型、数据量估计和成本等。
• 查看方法：
  • 在 SSMS 中，在查询前按下 Ctrl + M（显示实际执行计划）或 Ctrl + L（显示估计执行计划），然后执行查询。
  • 使用 SET 语句：SET SHOWPLAN_TEXT ON 或 SET STATISTICS PROFILE ON。
• 分析要点：
  • 高成本操作：找到成本最高的步骤。
  • 表扫描（Table Scan）：警惕！这通常意味着没有合适的索引。
  • 索引扫描（Index Scan） vs 索引查找（Index Seek）： Seek 效率远高于 Scan。Scan 意味着遍历了整个索引。
  • 键查找（Key Lookup）：如果开销很大，考虑创建覆盖索引。
  • 警告标志：如转换警告（隐式类型转换）等。

37. 什么是覆盖索引？

一个覆盖索引是指一个非聚集索引，它包含了查询中需要的所有字段。当查询的所有列都包含在索引的键或包含列中时，引擎可以直接从索引页中获取数据，而无需再去查找数据页，从而避免昂贵的键查找操作，极大提升性能。

创建覆盖索引示例：

CREATE INDEX IX_Covering ON Orders (CustomerID) INCLUDE (OrderDate, TotalAmount);-- 对于查询： SELECT OrderDate, TotalAmount FROM Orders WHERE CustomerID = @ID-- 这个索引就是覆盖索引。

38. 什么是索引碎片？如何维护？

• 索引碎片：当索引页的逻辑顺序与物理顺序不匹配，或者页的数据填充度很低时，就产生了碎片。碎片会导致更多的物理 I/O，降低查询性能。
• 类型：
  • 外部碎片：页的逻辑顺序与物理顺序不符。
  • 内部碎片：页中存在大量空闲空间。
• 维护方法：
  • 重组（REORGANIZE）：对叶级页以物理方式重新排序，并压缩索引页。是在线操作，干扰小。适用于轻度碎片。
  • 重建（REBUILD）：删除旧索引并创建一个新的索引。可以最大限度地减少碎片，是离线操作（在 Enterprise 版中可以在线）。适用于重度碎片。

39. 什么时候不适合创建索引？

• 表非常小（数据量很少）。
• 列的值重复度很低（如性别列，只有‘男’，‘女’），索引效果不佳。
• 列经常被频繁进行 INSERT/UPDATE/DELETE 操作，因为维护索引需要开销。
• 不会在查询的 WHERE 或 JOIN 条件中使用的列。

40. 谈谈 ACID 属性。

• 原子性（Atomicity）：事务是一个不可分割的工作单位，事务中的操作要么都发生，要么都不发生。
• 一致性（Consistency）：事务必须使数据库从一个一致性状态变换到另一个一致性状态。
• 隔离性（Isolation）：一个事务的执行不能被其他事务干扰。
• 持久性（Durability）：一旦事务提交，它对数据库中数据的改变就是永久性的。

41. SQL Server 的隔离级别有哪些？脏读、不可重复读、幻读分别是什么？

• 读未提交（Read Uncommitted）：可以读取其他事务未提交的数据。会导致脏读。
• 读已提交（Read Committed）：只能读取其他事务已提交的数据。这是 SQL Server 的默认级别。避免了脏读，但可能导致不可重复读。
• 可重复读（Repeatable Read）：保证在同一个事务中，多次读取同一数据的结果是一致的。避免了脏读和不可重复读，但可能导致幻读。
• 快照（Snapshot）：在事务开始时提供数据的一个一致性版本。读取的是事务开始时的数据快照，不会阻塞写操作。避免了脏读、不可重复读和幻读。
• 可序列化（Serializable）：最高隔离级别，强制事务串行执行。避免了所有并发问题，但性能最差。

名词解释：

• 脏读：事务A读取了事务B未提交的修改数据，之后B回滚了，A读到的就是脏数据。
• 不可重复读：事务A多次读取同一数据，在此期间事务B修改并提交了该数据，导致A多次读取的结果不一致。
• 幻读：事务A多次读取一个范围的数据，在此期间事务B插入或删除了该范围内的数据并提交，导致A多次读取时发现“凭空”多出或少了一些行。

42. 什么是死锁？如何避免和解决？

• 死锁：两个或更多事务相互等待对方释放资源，导致它们都无法继续执行的状态。
• 避免：
  • 以相同的顺序访问表。
  • 保持事务简短，尽快提交。
  • 使用较低的隔离级别（如 Read Committed）。
  • 使用 LOCK_TIMEOUT 设置。
• 解决：SQL Server 内置的死锁监视器会检测到死锁，并选择一个作为“牺牲品”将其回滚，从而让其他事务继续进行。牺牲品事务会收到 1205 错误。

43. 谈谈 CTE（公用表表达式）、临时表和表变量。

• CTE：
  • 更像一个临时的视图，只在查询期间存在。
  • 可读性好，特别适合递归查询。
  • 不能创建索引。
• 临时表（#Temp）：
  • 存储在 TempDB 中，存在于会话或嵌套作用域中。
  • 可以创建索引和统计信息。
  • 适合存储较大的中间结果集。
• 表变量（@Table）：
  • 也存储在 TempDB 中，存在于批处理/函数/存储过程的作用域中。
  • 通常认为它更快（对于小数据量），因为它没有统计信息，导致优化器总是假设它只有1行。
  • 不能创建索引（除了主键和唯一约束）。

44. 行版本控制和乐观并发控制是什么？

这是基于快照隔离级别的机制。当数据被修改时，SQL Server 会在 TempDB 中保存被修改行的旧版本。其他正在读取的事务可以从 TempDB 中读取这个旧版本，从而不会与写事务发生阻塞。READ_COMMITTED_SNAPSHOT 和 ALLOW_SNAPSHOT_ISOLATION 数据库选项与此相关。

45. SQL Server 的高可用性方案有哪些？

• AlwaysOn 故障转移集群实例（FCI）：基于 Windows 故障转移集群，共享存储。实例级别的高可用。
• AlwaysOn 可用性组（AG）：SQL Server 的核心高可用和灾难恢复解决方案。数据库级别，不共享存储，可读副本，功能最强大。
• 数据库镜像（已弃用，被AG取代）：主库和镜像库之间同步数据。
• 日志传送：通过定期备份主数据库的事务日志并还原到辅助服务器来实现。恢复时间较长。

46. 你知道哪些 SQL Server 的新特性？（根据面试公司使用的版本准备）

• JSON 支持：FOR JSON PATH/AUTO, OPENJSON 等。
• STRING_AGG 函数：将多行字符串值合并成一个字符串。
• 查询存储（Query Store）：用于跟踪查询执行计划、性能历史，并强制特定计划。
• 时态表（Temporal Tables）：自动跟踪和管理数据的历史变化。
• 内存优化表（In-Memory OLTP）：将表和存储过程放入内存，极大提升性能。

47. 存储过程和函数的区别是什么？

48. 什么时候应该使用存储过程？什么时候应该使用函数？**

• 使用存储过程：

  • 执行复杂的业务逻辑
  • 需要返回多个结果集
  • 需要进行 DML 操作（INSERT/UPDATE/DELETE）
  • 需要事务控制
  • 性能要求高（预编译、执行计划重用）

• 使用函数：

  • 封装可重用的计算逻辑
  • 在查询中作为列使用
  • 简化复杂的 JOIN 或 WHERE 条件
  • 返回表值供 FROM 子句使用

49. 什么是触发器？INSTEAD OF 和 AFTER 触发器的区别？**

• 触发器：一种特殊的存储过程，在特定数据库事件（INSERT/UPDATE/DELETE）发生时自动执行。

• AFTER 触发器（FOR 触发器）：

  • 在 DML 操作执行完成后 触发
  • 可以访问 inserted 和 deleted 魔术表
  • 常用于审计、日志记录、数据一致性检查

• INSTEAD OF 触发器：

  • 取代 原始的 DML 操作执行
  • 在约束检查之前触发
  • 常用于实现复杂的视图更新逻辑，或对不可更新视图进行更新

50. inserted 和 deleted 魔术表是什么？**

这两个是触发器中的特殊内存表：

• inserted：包含 INSERT 或 UPDATE 操作的新数据
• deleted：包含 DELETE 或 UPDATE 操作的旧数据

-- 在 UPDATE 触发器中CREATE TRIGGER trg_AuditUpdate ON Employees AFTER UPDATEASBEGIN    INSERT INTO AuditTable (EmployeeID, OldSalary, NewSalary)    SELECT d.EmployeeID, d.Salary, i.Salary    FROM deleted d     INNER JOIN inserted i ON d.EmployeeID = i.EmployeeID    WHERE d.Salary <> i.Salary;END;

51. ROW_NUMBER()、RANK()、DENSE_RANK() 的区别？**

这三个都是窗口函数，用于为结果集的行分配排名：

• ROW_NUMBER()：为每一行分配一个唯一的连续序号（1, 2, 3, 4…）
• RANK()：相同的值获得相同排名，但会跳过后续排名（1, 2, 2, 4…）
• DENSE_RANK()：相同的值获得相同排名，但不跳过后续排名（1, 2, 2, 3…）

SELECT     Name, Score,    ROW_NUMBER() OVER (ORDER BY Score DESC) as RowNum,    RANK() OVER (ORDER BY Score DESC) as Rank,    DENSE_RANK() OVER (ORDER BY Score DESC) as DenseRankFROM Students;

52. 什么是公用表表达式（CTE）的递归查询？

递归 CTE 用于处理层次结构数据（如组织结构、菜单树等）：

-- 查询某个部门及其所有子部门WITH DepartmentCTE AS (    -- 锚定成员：根节点    SELECT DepartmentID, DepartmentName, ParentDepartmentID    FROM Departments    WHERE DepartmentID = @RootDepartmentID        UNION ALL        -- 递归成员：子节点    SELECT d.DepartmentID, d.DepartmentName, d.ParentDepartmentID    FROM Departments d    INNER JOIN DepartmentCTE cte ON d.ParentDepartmentID = cte.DepartmentID)SELECT * FROM DepartmentCTE;

53. 什么是参数嗅探问题？如何解决？

• 参数嗅探：SQL Server 在编译存储过程时，使用第一次执行时的参数值来生成执行计划。如果后续执行的参数值数据分布差异很大，可能导致性能问题。

• 解决方案：

  • 使用 OPTION (RECOMPILE)：每次执行都重新编译
  • 使用 OPTION (OPTIMIZE FOR UNKNOWN)：使用平均数据分布
  • 使用局部变量：将参数赋值给局部变量，在查询中使用局部变量
  • 使用 WITH RECOMPILE 选项创建存储过程

54. 如何查找和优化慢查询？

• 查找慢查询：

  • 使用 SQL Server Profiler
  • 使用扩展事件（Extended Events）
  • 查询动态管理视图（DMV）：

  -- 查找最耗时的查询SELECT TOP 10     total_elapsed_time/execution_count AS avg_elapsed_time,    execution_count,    SUBSTRING(st.text, (qs.statement_start_offset/2)+1,         ((CASE qs.statement_end_offset WHEN -1 THEN DATALENGTH(st.text)            ELSE qs.statement_end_offset END - qs.statement_start_offset)/2) + 1) AS statement_textFROM sys.dm_exec_query_stats qsCROSS APPLY sys.dm_exec_sql_text(qs.sql_handle) stORDER BY avg_elapsed_time DESC;

• 优化方法：

  • 添加合适的索引
  • 重写查询逻辑
  • 避免在 WHERE 子句中对字段进行函数操作
  • 减少不必要的列查询

55. 数据库的三大范式是什么？

• 第一范式（1NF）：每个列都是原子的，不可再分
• 第二范式（2NF）：满足 1NF，且非主属性完全依赖于主键（消除部分依赖）
• 第三范式（3NF）：满足 2NF，且非主属性之间没有传递依赖

56. 什么时候应该反范式化？

虽然范式化减少了数据冗余，但在以下情况可以考虑反范式化：

• 频繁的 JOIN 操作影响性能时
• 需要提高查询性能的读密集型场景
• 数据仓库或报表数据库
• 历史数据表，数据不再变更

57. 完整备份、差异备份和事务日志备份的区别？

• 完整备份：备份整个数据库，是其他备份的基础
• 差异备份：只备份自上次完整备份以来发生变化的数据页
• 事务日志备份：备份事务日志，允许时间点恢复

恢复场景示例：

完整备份 (周日) → 差异备份 (周一) → 日志备份 (周二 10:00) → 日志备份 (周二 11:00)

如果周二 11:30 发生故障，可以恢复到：周日完整备份 + 周一差异备份 + 周二 10:00 日志 + 周二 11:00 日志

58. 简单恢复模式 vs 完整恢复模式

• 简单恢复模式：

  • 不备份事务日志
  • 不能进行时间点恢复
  • 日志空间自动回收
  • 适合测试环境或可接受数据丢失的场景

• 完整恢复模式：

  • 需要定期备份事务日志
  • 支持时间点恢复
  • 可以防止数据丢失
  • 生产环境推荐使用

59. 如何设计一个支持软删除的系统？

-- 在表中添加删除标记字段ALTER TABLE Products ADD IsDeleted BIT NOT NULL DEFAULT 0;ALTER TABLE Products ADD DeletedDate DATETIME NULL;-- 使用视图过滤已删除的记录CREATE VIEW vw_ActiveProducts ASSELECT * FROM Products WHERE IsDeleted = 0;-- 使用 INSTEAD OF DELETE 触发器实现软删除CREATE TRIGGER trg_SoftDeleteProductON Products INSTEAD OF DELETEASBEGIN    UPDATE Products     SET IsDeleted = 1, DeletedDate = GETDATE()    WHERE ProductID IN (SELECT ProductID FROM deleted);END;

60. 如何处理数据库中的循环引用？

• 方案1：使用延迟约束检查

  ALTER TABLE TableA ADD CONSTRAINT FK_TableA_TableB FOREIGN KEY (BID) REFERENCES TableB(BID)-- 在某些版本中可以使用 DEFERRABLE

• 方案2：允许 NULL 值，先插入部分数据再更新

• 方案3：使用触发器代替外键约束

• 方案4：重新设计表结构，消除循环引用

61. 如何实现数据库的审计功能？

• 方法1：使用触发器记录数据变更
• 方法2：使用 SQL Server 的变更数据捕获（CDC）功能
• 方法3：使用 SQL Server Audit 功能（企业版）
• 方法4：在应用层实现审计逻辑

62. SQL Server 2019 的新特性有哪些？

• 智能查询处理：自适应连接、行模式内存授予反馈等
• 数据虚拟化：通过 PolyBase 查询外部数据源
• Java 语言扩展：在 SQL Server 中执行 Java 代码
• 加速数据库恢复：大幅减少数据库恢复时间
• 列存储索引增强：可更新的非聚集列存储索引

63. 什么是内存优化表？适用场景？

内存优化表将数据完全存储在内存中，提供极高的吞吐量：

• 适用场景：

  • 高频读写的高并发场景
  • 会话状态管理
  • 实时数据处理
  • 需要亚毫秒级响应的应用

• 创建示例：

CREATE TABLE dbo.SessionState(    SessionID nvarchar(64) NOT NULL PRIMARY KEY NONCLUSTERED,    UserData varbinary(MAX) NOT NULL,    CreatedDate datetime2 NOT NULL) WITH (MEMORY_OPTIMIZED = ON, DURABILITY = SCHEMA_AND_DATA);

数据库表题

1. 电商产品有颜色(红、蓝、黑)、尺寸(S、M、L)等变体信息，买家购买一个红色中码的商品对应的变体名为：红_M。那么给定一组变体信息，用程序生成所有变体组合数据。

变体示例：Color: Red,Green Size: S,M Style: A
变体组合结果：Red_S_A; Red_M_A; Green_S_A; Green_M_A

//测试用例 var list = new List<string[]>{ new string[]{"Red","Green"}, new string[]{"S","M"}, new string[]{"A"}};var result = Combine(list);//期望result为：Red_S_A; Red_M_A; Green_S_A; Green_M_Apublic List<string> Combine(List<string[]> list){ … }

• 答案：

public static List<string> Combine(List<string[]> list){    List<string> result = new List<string>();    int[] indices = new int[list.Count]; // 用于跟踪每个字符串数组中当前选取的元素的索引    while (true)    {        string combined = "";        for (int i = 0; i < list.Count; i++)        {            combined += "_" + list[i][indices[i]]; // 将当前索引对应的元素添加到组合中        }        result.Add(combined); // 将组合添加到结果列表中        // 更新索引        int j = list.Count - 1;        while (j >= 0 && indices[j] == list[j].Length - 1)        {            indices[j] = 0;            j--;        }        // 检查是否所有索引都已经达到最大值        if (j < 0)        {            break;        }        indices[j]++; // 增加索引    }    return result;}

2. 内部系统的产品库中有一个产品拥有三个维度的变体，分别是：Color、Size 和 Style。现在要将其上传到平台 A，但平台 A 仅支持两个维度的变体，因此需要对变体进行降维。那么给定一组变体信息，用程序实现变体降维操作。

变体示例: Color: Red,Green Size: S,M Style: A,B

降维后:Color: Red,Green Size: S_A,S_B,M_A,M_B

var pair = new Dictionary<string, List<string>> { {"Color",new List<string>{ "Red","Green" }}, {"Size",new List<string>{ "S","M" }}, {"Style",new List<string>{ "A","B" }},};var result = Reduce(pair);public Dictionary<string, List<string>> Reduce(Dictionary<string, List<string>> pair){...}

• 答案：

class Program { static void Main() { // 定义变体维度 string[] colors = { "Red", "Green" }; string[] sizes = { "S", "M" }; string[] styles = { "A", "B" };    // 降维操作    Dictionary<string, string[]> reducedDimensions = ReduceDimensions(colors, sizes, styles);        // 打印降维后的变体    Console.WriteLine("Color: " + string.Join(",", reducedDimensions["Color"]));    Console.WriteLine("Size: " + string.Join(",", reducedDimensions["Size"]));}static Dictionary<string, string[]> ReduceDimensions(string[] colors, string[] sizes, string[] styles){    var reduced = new Dictionary<string, string[]>    {        { "Color", colors },        { "Size", sizes.SelectMany(size => styles.Select(style => size + "_" + style)).ToArray() }    };    return reduced;}}

3. 试用SQL查询语句表达下列对教学数据库中三个基本表 S、SC 、C 的查询：

• S(sno, sname, sage, ssex)：学号、姓名、年龄、性别
• SC(sno, cno, grade)：学号、课程号、成绩
• C(cno, cname, teacher)：课程号、课程名、教师名

3.1. 求年龄大于所有女同学年龄的男学生姓名和年龄

SELECT sname, sage FROM S AS XWHERE x.ssex = '男' AND x.sage > ALL (  SELECT sage FROM S AS Y WHERE y.ssex = '女');

3.2. 求年龄大于女同学平均年龄的男学生姓名和年龄

SELECT sname, sage FROM SWHERE ssex = '男' AND sage > (  SELECT AVG(sage) FROM S WHERE ssex = '女');

3.3. 在SC中检索成绩为空值的学生学号和课程号

SELECT sno, cno FROM SC WHERE grade IS NULL;

3.4. 检索姓名以WANG打头的所有学生的姓名和年龄

SELECT sname, sage FROM S WHERE sname LIKE 'WANG%';

3.5. 检索学号比WANG同学大，而年龄比他小的学生姓名

SELECT sname FROM sWHERE sno > (SELECT sno FROM s WHERE sname = 'WANG')  AND sage < (SELECT sage FROM s WHERE sname = 'WANG');

3.6. 统计每门课程的学生选修人数（超过2人的课程才统计）

SELECT cno, COUNT(sno) AS 人数 FROM SCGROUP BY cno HAVING COUNT(sno) > 2ORDER BY 人数 DESC, cno ASC;

3.7. 求LIU老师所授课程的每门课程的学生平均成绩

SELECT cname, AVG(grade) FROM SC, CWHERE SC.cno = C.cno AND teacher = 'liu'GROUP BY c.cno, cname;

3.8. 求选修C4课程的学生的平均年龄

SELECT AVG(sage) FROM S, SCWHERE S.sno = SC.sno AND cno = '4';

3.9. 统计有学生选修的课程门数

SELECT COUNT(DISTINCT cno) FROM SC;

3.10. 在基本表SC中修改4号课程的成绩

UPDATE SC SET grade = grade * 1.05 WHERE cno = '4' AND grade <= 75;UPDATE SC SET grade = grade * 1.04 WHERE cno = '4' AND grade > 75;

3.11. 把低于总平均成绩的女同学成绩提高5%

UPDATE SC SET grade = grade * 1.05WHERE grade < (SELECT AVG(grade) FROM SC)  AND sno IN (SELECT sno FROM S WHERE ssex = '女');

3.12. 把选修数据库原理课不及格的成绩改为空值

UPDATE SC SET grade = NULLWHERE grade < 60 AND cno IN (  SELECT cno FROM C WHERE cname = '数据库原理');

3.13. 把WANG同学的学习选课和成绩全部删去

DELETE FROM SC WHERE sno IN (  SELECT sno FROM S WHERE sname = 'WANG');

3.14. 在基本表SC中删除尚无成绩的选课元组

DELETE FROM SC WHERE grade IS NULL;

3.15. 在基本表S中插入一个学生元组

INSERT INTO S(sno, sname, sage) VALUES('S9', 'WU', 18);

Jenkins

https://jk.dev.eduwon.cn/

NetData

Gitea

https://git.eduwon.cn/

汉字着重号的网页实现

<span class= 'focus_on'>中文符号</span >
无换行着重符号

带颜色 符号 更换类名即可 默认为黑色

.focus_on{  text-emphasis: black open ;  -webbit-text-emphasis:black open;  text-emphasis-style:'●';  -webkit-text-emphasis-style:'●';  -webkit-text-emphasis-position: under;  text-emphasis-position: under;}.focus_on_red{  text-emphasis: red open ;  -webbit-text-emphasis:black open;  text-emphasis-style:'●';  -webkit-text-emphasis-style:'●';  -webkit-text-emphasis-position: under;  text-emphasis-position: under;}.focus_on_green{  text-emphasis: green open ;  -webbit-text-emphasis:black open;  text-emphasis-style:'●';  -webkit-text-emphasis-style:'●';  -webkit-text-emphasis-position: under;  text-emphasis-position: under;}.focus_on_orange{  text-emphasis: orange open ;  -webbit-text-emphasis:black open;  text-emphasis-style:'●';  -webkit-text-emphasis-style:'●';  -webkit-text-emphasis-position: under;  text-emphasis-position: under;}.focus_on_yellow{  text-emphasis: yellow open ;  -webbit-text-emphasis:black open;  text-emphasis-style:'●';  -webkit-text-emphasis-style:'●';  -webkit-text-emphasis-position: under;  text-emphasis-position: under;}.focus_on_blue{  text-emphasis: blue open ;  -webbit-text-emphasis:black open;  text-emphasis-style:'●';  -webkit-text-emphasis-style:'●';  -webkit-text-emphasis-position: under;  text-emphasis-position: under;}.focus_on_purple{  text-emphasis: purple open ;  -webbit-text-emphasis:black open;  text-emphasis-style:'●';  -webkit-text-emphasis-style:'●';  -webkit-text-emphasis-position: under;  text-emphasis-position: under;}

docker-compose部署LOKI

docker-compose.yaml

version: "3"networks:  loki:services:  loki:    image: grafana/loki:2.5.0    ports:      - "3100:3100"    command: -config.file=/etc/loki/local-config.yaml    networks:      - loki  promtail:    image: grafana/promtail:2.5.0    volumes:      - /var/log:/var/log    command: -config.file=/etc/promtail/config.yml    networks:      - loki  grafana:    image: grafana/grafana:latest    ports:      - "3000:3000"    networks:      - loki

当前目录执行命令

 docker-compose up -d

配置数据源

启动后使用浏览器访问 http://localhost:3000

账号admin 密码admin 登录成功后会提示修改密码

配置所用数据库

黄色代表我们配置成功

配置Asp.Net Core程序

使用到的库

• Serilog
• Serilog.Exceptions
• Serilog.Sinks.Grafana.Loki

初始化NLog时配置

Code

 public static IWebHostBuilder UseSerilogDefault(this IWebHostBuilder hostBuilder)        {            var environmentName = Environment.GetEnvironmentVariable("ASPNETCORE_ENVIRONMENT");            var configuration = new ConfigurationBuilder()                .SetBasePath(Directory.GetCurrentDirectory())                .AddJsonFile("appsettings.json", optional: true, reloadOnChange: true)                .AddJsonFile($"appsettings.{environmentName}.json", optional: false, reloadOnChange: true)                .AddEnvironmentVariables()                .Build();            Log.Logger = new LoggerConfiguration()                .Enrich.FromLogContext()                .Enrich.WithExceptionDetails()                .Enrich.WithMachineName()                .WriteTo.Debug()                .WriteTo.Console()                .WriteTo.GrafanaLoki("http://http:localhost:3100/")                // .WriteTo.Elasticsearch(ConfigureElasticSink(configuration, environmentName))                .Enrich.WithProperty("Client", "appClient")                .ReadFrom.Configuration(configuration)                .CreateLogger();                        hostBuilder.UseSerilog();            return hostBuilder;        }

最终效果

日志浏览

筛选条件

效果图

docker-compose部署LOKI

docker-compose.yaml

version: "3"networks:  loki:services:  loki:    image: grafana/loki:2.5.0    ports:      - "3100:3100"    command: -config.file=/etc/loki/local-config.yaml    networks:      - loki  promtail:    image: grafana/promtail:2.5.0    volumes:      - /var/log:/var/log    command: -config.file=/etc/promtail/config.yml    networks:      - loki  grafana:    image: grafana/grafana:latest    ports:      - "3000:3000"    networks:      - loki

当前目录执行命令

 docker-compose up -d

配置数据源

启动后使用浏览器访问 http://localhost:3000

账号admin 密码admin 登录成功后会提示修改密码

配置所用数据库

黄色代表我们配置成功

配置Asp.Net Core程序

使用到的库

• Nlog
• Nlog.Targets.Loki

初始化NLog时配置

Code

static Logger()        {            var config = new NLog.Config.LoggingConfiguration();            string layout = LoggerConfig.Layout;                        AddTarget(new LokiTarget()            {                Name = "loki",                BatchSize = 200,                TaskDelayMilliseconds = 500,                Endpoint = "http://122.112.248.242:3101/",                OrderWrites = true,                CompressionLevel=CompressionLevel.NoCompression,                Layout = layout,                Labels = { new LokiTargetLabel()                {                    Name= "Client",                    Layout = "AppApi"                }                 }            });                        NLog.LogManager.Configuration = config;            void AddTarget(NLog.Targets.Target target)            {                config.AddTarget(target);                config.AddRuleForAllLevels(target);            }        }

配置文件配置

Nlog.config

<nlog xmlns="http://www.nlog-project.org/schemas/NLog.xsd"      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">    <extensions>    <add assembly="NLog.Loki" />  </extensions>    <targets>    <target       name="loki"       xsi:type="loki"      batchSize="200"      taskDelayMilliseconds="500"      endpoint="http://localhost:3100"      username="myusername"      password="secret"      orderWrites="true"      compressionLevel="noCompression"      layout="${level}|${message}${onexception:|${exception:format=type,message,method:maxInnerExceptionLevel=5:innerFormat=shortType,message,method}}|source=${logger}">            <label name="app" layout="my-app-name" />      <label name="server" layout="${hostname:lowercase=true}" />    </target>  </targets>  <rules>    <logger name="*" minlevel="Info" writeTo="loki" />  </rules></nlog>

最终效果

日志浏览

筛选条件

效果图

为什么要学习kubernetes

使用Kubernetes的理由很多，最重要的理由是，IT 行业从来都是由新技术驱动的。

当前，Docker 容器化技术已经被很多公司采用，从单机走向集群已成为必然。云计算的蓬勃发展正在加速这一进程。Kubernetes作为当前被业界广泛认可和看好的基于 Docker 的大规模容器化分布式系统解决方案，得到了以谷歌为首的 IT 巨头们的大力宣传和维持推进。

主要优点

• 一个平台搞定所有
• 云环境无缝迁移
• 高效的利用资源
• 开箱即用的自动缩放能力
• 使 CI/CD 更简单
• 可靠性

Kubernetes学习地址

学习地址

具体步骤

Kubeasz介绍

项目致力于提供快速部署高可用k8s集群的工具, 同时也努力成为k8s实践、使用的参考书；基于二进制方式部署和利用ansible-playbook实现自动化；既提供一键安装脚本, 也可以根据安装指南分步执行安装各个组件

地址

实现

测试机 CentOS 7.6

请使用root用户进行以下命令的操作

下载命令行工具

systemctl status firewalldsystemctl stop firewalldsystemctl disable firewalldexport release=3.0.0wget https://github.com/easzlab/kubeasz/releases/download/${release}/ezdownchmod +x ./ezdown

下载资源包

./ezdown -D

安装单点集群

./ezdown -Sdocker exec -it kubeasz ezctl start-aio

验证安装

如果提示kubectl: command not found，退出重新ssh登录一下，环境变量生效即可

$ kubectl version         # 验证集群版本     $ kubectl get node        # 验证节点就绪 (Ready) 状态$ kubectl get pod -A      # 验证集群pod状态，默认已安装网络插件、coredns、metrics-server等$ kubectl get svc -A      # 验证集群服务状态

dashboard

# 查看pod 运行状态kubectl get pod -n kube-system | grep dashboarddashboard-metrics-scraper-856586f554-l6bf4   1/1     Running   0          35mkubernetes-dashboard-698d4c759b-67gzg        1/1     Running   0          35m# 查看dashboard servicekubectl get svc -n kube-system|grep dashboardkubernetes-dashboard   NodePort    10.68.219.38   <none>        443:24108/TCP                   53s# 查看pod 运行日志kubectl logs -n kube-system kubernetes-dashboard-698d4c759b-67gzg

获取登录令牌

kubectl -n kube-system describe secret $(kubectl -n kube-system get secret | grep admin-user | awk '{print $1}')

登录地址

因为dashboard 作为k8s 原生UI，能够展示各种资源信息，甚至可以有修改、增加、删除权限，所以有必要对访问进行认证和控制，为演示方便这里使用 https://NodeIP:NodePort 方式访问 dashboard，支持两种登录方式：Kubeconfig、令牌(Token)

注意： 使用chrome浏览器访问 https://NodeIP:NodePort 可能提示安全风险无法访问，可以换firefox浏览器设置安全例外，继续访问。

一定要https

启用 kubectl 自动补全

yum -y install epel-releaseyum install bash-completion bash-completion-extras -ylocate bash_completion.shupdatedbsource /etc/profile.d/bash_completion.shlogout //登出后重新登录echo 'source <(kubectl completion bash)' >>~/.bashrckubectl completion bash >/etc/bash_completion.d/kubectl

清理环境

在宿主机上，按照如下步骤清理

• 清理集群 docker exec -it kubeasz ezctl destroy default
• 清理运行的容器 ./ezdown -C
• 清理容器镜像 docker system prune -a
• 停止docker服务 systemctl stop docker
• 删除docker文件

 umount /var/run/docker/netns/default umount /var/lib/docker/overlay rm -rf /var/lib/docker /var/run/docker

为保证无残留，请重启服务器

参考文档

安装
dashboard

完结撒花

Kubernetes使得应用的启动、迁移、部署变得简单又安全。

不必担心应用迁移后工作出现问题，也不用担心一台服务器无法应付突发的用户量。

需要注意的是，你的应用最好使用微服务架构进行开发，因为微服务应用比单体应用更适合做容器化。

不要为了单纯的使用Kubernetes而引入，要看你的系统规模是否值得去用，否则可能适得其反，毕竟Kubernetes的学习是需要一定成本。