MySQL读写延迟与并发导致的问题解决方案

-- 场景：刚写入的数据立即查询却查不到
-- 1. 写入主库
INSERT INTO users (name, email) VALUES ('张三', 'zhangsan@example.com');

-- 2. 立即从从库查询（可能查询失败）
SELECT * FROM users WHERE email = 'zhangsan@example.com'; -- 查不到！

// 场景：库存超卖
// 用户A和用户B同时购买最后一件商品
$stock = DB::table('products')->where('id', 1)->value('stock'); // 同时读到 stock=1

// 用户A
if ($stock > 0) {
    DB::table('products')->where('id', 1)->decrement('stock');
    // 处理订单...
}

// 用户B（同时执行）
if ($stock > 0) {
    DB::table('products')->where('id', 1)->decrement('stock');
    // 库存变成 -1！
}

flowchart TD
    A[用户遇到数据不一致] --> B{检查问题类型}
    B -->|立即写入后查不到| C[读写延迟<br>主从同步问题]
    B -->|数据计数错误/超卖| D[并发竞争问题]
    B -->|随机性失败| E[网络/连接池问题]
    C --> F[解决方案: 强制主库读取]
    D --> G[解决方案: 事务锁/队列]
    E --> H[解决方案: 连接优化/重试]

// Laravel 中强制使用主库
// 方法1：使用 writeConnection
$user = DB::connection('mysql::write')->table('users')->find(1);
// 方法2：使用 onWriteConnection
$user = User::onWriteConnection()->find(1);
// 方法3：设置整个查询使用主库
DB::connection('mysql')->select('SET SESSION TRANSACTION READ WRITE');
$users = DB::table('users')->get();

// config/database.php
'mysql' => [
    'sticky' => true, // 启用粘性连接
    'read' => [
        ['host' => 'read1.example.com'],
        ['host' => 'read2.example.com']
    ],
    'write' => [
        'host' => 'write.example.com'
    ],
],
// 写入后的短时间内，同一请求的读操作会使用主库

function getWithRetry($id, $maxRetries = 3)
{
    for ($i = 0; $i < $maxRetries; $i++) {
        $data = User::find($id);
        if ($data) {
            return $data;
        }
        // 延迟后重试
        if ($i < $maxRetries - 1) {
            usleep(100000 * pow(2, $i)); // 指数退避
        }
    }
    // 最后一次尝试从主库读取
    return User::onWriteConnection()->find($id);
}

// 悲观锁：SELECT ... FOR UPDATE
DB::transaction(function () {
    // 锁定行
    $product = DB::table('products')
        ->where('id', 1)
        ->lockForUpdate()  // 行级锁
        ->first();
    
    if ($product->stock > 0) {
        DB::table('products')
            ->where('id', 1)
            ->decrement('stock');
        
        // 创建订单...
    }
});

// 或者使用 sharedLock（共享锁）
$product = DB::table('products')
    ->where('id', 1)
    ->sharedLock()
    ->first();

// 使用原子更新，避免先读后写
$affected = DB::table('products')
    ->where('id', 1)
    ->where('stock', '>', 0)
    ->decrement('stock');

if ($affected > 0) {
    // 库存减少成功，处理订单
} else {
    // 库存不足
}

// 使用 CASE 语句
DB::table('products')
    ->where('id', 1)
    ->update([
        'stock' => DB::raw('CASE WHEN stock > 0 THEN stock - 1 ELSE stock END')
    ]);

use Illuminate\Support\Facades\Redis;
function purchaseWithLock($productId, $userId)
{
    $lockKey = "purchase:lock:{$productId}";
    $lock = Redis::set($lockKey, $userId, 'NX', 'EX', 10); // 10秒超时
    if (!$lock) {
        throw new Exception('系统繁忙，请稍后重试');
    }
    try {
        DB::transaction(function () use ($productId) {
            $product = Product::where('id', $productId)
                ->where('stock', '>', 0)
                ->first();
            if ($product) {
                $product->decrement('stock');
                // 创建订单...
            }
        });
    } finally {
        Redis::del($lockKey);
    }
}

// 将并发请求转为串行处理
class ProcessPurchase implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
    public function handle()
    {
        DB::transaction(function () {
            $product = Product::where('id', $this->productId)
                ->where('stock', '>', 0)
                ->first();
            if ($product) {
                $product->decrement('stock');
                // 创建订单...
            }
        });
    }
}
// 触发购买
ProcessPurchase::dispatch($productId, $userId);

class PurchaseService
{
    public function purchase($productId, $userId, $quantity = 1)
    {
        // 1. 快速失败检查（不涉及数据库）
        if ($quantity <= 0) {
            throw new InvalidArgumentException('数量必须大于0');
        }
        // 2. 使用数据库事务保证原子性
        return DB::transaction(function () use ($productId, $userId, $quantity) {
            // 3. 使用行级锁防止并发
            $product = Product::where('id', $productId)
                ->lockForUpdate()  // 悲观锁
                ->first();
            if (!$product) {
                throw new ModelNotFoundException('商品不存在');
            }
            if ($product->stock < $quantity) {
                throw new Exception('库存不足');
            }
            // 4. 原子更新
            $product->decrement('stock', $quantity);
            // 5. 创建订单
            $order = Order::create([
                'user_id' => $userId,
                'product_id' => $productId,
                'quantity' => $quantity,
                'status' => 'paid'
            ]);
            // 6. 清理缓存
            Cache::forget("product:{$productId}");
            return $order;
        }, 3); // 重试3次
    }
}

// 监控数据库延迟
class DatabaseMonitor
{
    public static function checkReplicationDelay()
    {
        // 检查主从延迟
        $delay = DB::select('SHOW SLAVE STATUS');
        if ($delay['Seconds_Behind_Master'] > 5) {
            Log::warning('数据库主从延迟过高', [
                'delay' => $delay['Seconds_Behind_Master'],
                'master' => config('database.connections.mysql.write.host'),
                'slave' => config('database.connections.mysql.read.host')
            ]);
        }
    }
    public static function logSlowQueries()
    {
        // 启用慢查询日志
        DB::enableQueryLog();
        // 执行查询后
        $queries = DB::getQueryLog();
        foreach ($queries as $query) {
            if ($query['time'] > 1000) { // 超过1秒
                Log::warning('慢查询', $query);
            }
        }
    }
}

// 在代码中添加调试信息
DB::listen(function ($query) {
    Log::info('SQL Query', [
        'sql' => $query->sql,
        'bindings' => $query->bindings,
        'time' => $query->time,
        'connection' => $query->connectionName
    ]);
});
// 检查当前使用的连接
$connection = DB::getDefaultConnection();
$isWriting = DB::connection()->getPdo()->inTransaction();

// 临时解决方案：全局强制走主库
// 在 AppServiceProvider 中
public function boot()
{
    if (app()->environment('production') && request()->has('debug_read')) {
        DB::connection('mysql')->setReadConnection('write');
    }
}