PHP/CodeIgniter中并发请求下数据库操作失败的同步队列解决方案咨询

Got it, let's tackle this concurrency issue in CodeIgniter—this is super common when dealing with database writes that can't overlap (like inventory updates or transactional user actions). I’ve helped teams solve similar problems before, so here are a few practical approaches that mimic Java’s synchronized method behavior, tailored for CodeIgniter:

This acts exactly like a synchronized method—only one request can execute the protected database logic at a time. CodeIgniter’s built-in cache system works perfectly here; use File cache for single-server setups, or Redis/Memcached for multi-server environments.

Example Code (Model Level) #

class Transaction_model extends CI_Model {
    // 定义锁标识和超时时间（避免死锁）
    private $lock_key = 'global_db_operation_lock';
    private $lock_timeout = 30; // 锁有效期30秒

    public function synchronized_db_write($payload) {
        // 加载缓存驱动（如果没在config里自动加载）
        $this->load->driver('cache', array('adapter' => 'file'));

        // 循环尝试获取锁，直到成功或超时
        $start_time = time();
        while (!$this->cache->save($this->lock_key, 'locked', $this->lock_timeout)) {
            if (time() - $start_time > $this->lock_timeout) {
                log_message('error', 'Failed to acquire lock after timeout');
                return false;
            }
            // 等待100ms再重试，避免CPU空转
            usleep(100000);
        }

        try {
            // 开启事务执行数据库操作
            $this->db->trans_start();
            
            // 替换成你的实际数据库逻辑
            $this->db->insert('transaction_logs', $payload);
            $this->db->update('inventory', ['stock' => 'stock - 1'], ['id' => $payload['item_id']]);
            
            $this->db->trans_complete();

            if (!$this->db->trans_status()) {
                throw new Exception('Database transaction failed');
            }
            return true;
        } catch (Exception $e) {
            log_message('error', 'Synchronized operation failed: ' . $e->getMessage());
            return false;
        } finally {
            // 无论成功失败，必须释放锁
            $this->cache->delete($this->lock_key);
        }
    }
}

If your concurrency issue is isolated to specific database rows (like updating a user's balance), use database-native row-level locking instead of a global lock. This is more efficient than a full synchronized method because it only blocks access to the targeted record.

Example Code #

class User_model extends CI_Model {
    public function update_user_balance($user_id, $amount) {
        $this->db->trans_start();

        // 锁定目标用户行，其他请求会等待直到锁释放
        $this->db->where('id', $user_id);
        $this->db->lock('FOR UPDATE'); // MySQL语法，其他数据库可调整
        $user = $this->db->get('users')->row();

        if (!$user) {
            $this->db->trans_rollback();
            return false;
        }

        // 执行余额更新（确保逻辑基于最新的锁定数据）
        $new_balance = $user->balance + $amount;
        $this->db->where('id', $user_id);
        $this->db->update('users', ['balance' => $new_balance]);

        $this->db->trans_complete();
        return $this->db->trans_status();
    }
}

For high-traffic applications where you don’t want users waiting for the operation to complete, implement a queue system. Requests are added to a queue table, and a background CLI process handles them one at a time.

Step 1: Create Queue Table #

CREATE TABLE `request_queue` (
    `id` INT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    `payload` TEXT NOT NULL,
    `status` ENUM('pending', 'processing', 'completed', 'failed') DEFAULT 'pending',
    `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
    `processed_at` DATETIME NULL
);

Step 2: Add Requests to Queue (Controller) #

class Order_controller extends CI_Controller {
    public function place_order() {
        $order_data = $this->input->post();
        
        // 将请求加入队列，立即返回响应给用户
        $this->db->insert('request_queue', [
            'payload' => json_encode($order_data)
        ]);

        echo json_encode([
            'status' => 'success',
            'message' => 'Your order is being processed'
        ]);
    }
}

Step 3: CLI Queue Processor #

class Queue_processor extends CI_Controller {
    public function __construct() {
        parent::__construct();
        // 只允许CLI访问
        if (!$this->input->is_cli_request()) {
            show_404();
        }
        $this->load->model('Order_model');
    }

    public function run() {
        echo "Queue processor started...\n";
        while (true) {
            // 原子性获取并标记待处理请求（避免重复处理）
            $this->db->trans_start();
            $this->db->where('status', 'pending');
            $this->db->order_by('created_at', 'ASC');
            $this->db->limit(1);
            $this->db->update('request_queue', ['status' => 'processing']);
            
            $request = $this->db->get_where('request_queue', ['status' => 'processing'])->row();
            $this->db->trans_complete();

            if (!$request) {
                // 无待处理请求，休眠5秒再检查
                echo "No pending requests, sleeping...\n";
                sleep(5);
                continue;
            }

            try {
                $payload = json_decode($request->payload, true);
                // 执行实际数据库操作
                $success = $this->Order_model->process_order($payload);

                $update_data = $success ? [
                    'status' => 'completed',
                    'processed_at' => date('Y-m-d H:i:s')
                ] : ['status' => 'failed'];

                $this->db->where('id', $request->id);
                $this->db->update('request_queue', $update_data);

                echo "Processed request ID {$request->id}: " . ($success ? 'Success' : 'Failed') . "\n";
            } catch (Exception $e) {
                log_message('error', "Queue processing failed for request {$request->id}: " . $e->getMessage());
                $this->db->where('id', $request->id);
                $this->db->update('request_queue', ['status' => 'failed']);
            }
        }
    }
}

Run the Processor #

Execute this via CLI to start the background process:

php index.php queue_processor run

For production, use tools like supervisor to keep the process running permanently.

• 锁超时: Always set a lock timeout to avoid deadlocks if a process crashes mid-operation.
• 事务完整性: Wrap all database operations in transactions to ensure data consistency.
• Multi-server Considerations: For multi-server setups, avoid File cache locks—use Redis/Memcached or database-level global locks instead.
• Retry Logic: For queue-based systems, add logic to reprocess failed requests after a delay.

内容的提问来源于stack exchange，提问作者Jagdish Chaudhary