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