PHP并发问题，多个同时请求；互斥体？

Question

所以我刚刚意识到 PHP 可能同时运行多个请求。昨晚的日志似乎显示有两个请求进来，是并行处理的；每个都触发了从另一台服务器导入数据；每个人都试图在数据库中插入一条记录。一个请求在尝试插入另一个线程刚刚插入的记录时失败（导入的数据带有 PK；我没有使用递增的 ID）：SQLSTATE[23000]: Integrity constraint violation: 1062 Duplicate entry '865020' for key 'PRIMARY' ...。

1. 我是否正确诊断了这个问题？
2. 我应该如何解决这个问题？

以下是部分代码。我已经删除了大部分内容（日志记录，从数据中创建患者以外的其他实体），但以下内容应包括相关的 sn-ps。请求命中 import() 方法，该方法本质上为要导入的每条记录调用 importOne()。注意 importOne() 中的 save 方法；这是一个 Eloquent 方法（使用 Laravel 和 Eloquent），它将生成 SQL 以根据需要插入/更新记录。

public function import()
{
        $now = Carbon::now();
        // Get data from the other server in the time range from last import to current import
        $calls = $this->getCalls($this->getLastImport(), $now);
        // For each call to import, insert it into the DB (or update if it already exists)
        foreach ($calls as $call) {
            $this->importOne($call);
        }
        // Update the last import time to now so that the next import uses the correct range
        $this->setLastImport($now);
}

private function importOne($call)
{
    // Get the existing patient for the call, or create a new one
    $patient = Patient::where('id', '=', $call['PatientID'])->first();
    $isNewPatient = $patient === null;
    if ($isNewPatient) {
        $patient = new Patient(array('id' => $call['PatientID']));
    }
    // Set the fields
    $patient->given_name = $call['PatientGivenName'];
    $patient->family_name = $call['PatientFamilyName'];
    // Save; will insert/update appropriately
    $patient->save();
}

我猜该解决方案需要在整个导入块周围使用互斥锁？如果一个请求无法获得互斥锁，它就会继续处理其余的请求。想法？

编辑：请注意，这不是严重故障。异常被捕获并记录下来，然后像往常一样响应请求。并且导入在另一个请求上成功，然后该请求将照常响应。用户并不聪明；他们甚至不知道导入，这不是请求的主要焦点。所以真的，我可以让它保持原样运行，除了偶尔的异常，没有什么不好的事情发生。但是，如果有一个修复程序可以防止完成额外的工作/不必要地向其他服务器发送多个请求，那可能值得追求。

EDIT2：好的，我已经尝试使用flock() 实现锁定机制。想法？下面的工作吗？我将如何对这个添加进行单元测试？

public function import()
{
    try {
        $fp = fopen('/tmp/lock.txt', 'w+');
        if (flock($fp, LOCK_EX)) {
            $now = Carbon::now();
            $calls = $this->getCalls($this->getLastImport(), $now);
            foreach ($calls as $call) {
                $this->importOne($call);
            }
            $this->setLastImport($now);
            flock($fp, LOCK_UN);
            // Log success.
        } else {
            // Could not acquire file lock. Log this.
        }
        fclose($fp);
    } catch (Exception $ex) {
        // Log failure.
    }
}

EDIT3：关于锁的以下替代实现的想法：

public function import()
{
    try {
        if ($this->lock()) {
            $now = Carbon::now();
            $calls = $this->getCalls($this->getLastImport(), $now);
            foreach ($calls as $call) {
                $this->importOne($call);
            }
            $this->setLastImport($now);
            $this->unlock();
            // Log success
        } else {
            // Could not acquire DB lock. Log this.
        }
    } catch (Exception $ex) {
        // Log failure
    }
}

/**
 * Get a DB lock, returns true if successful.
 *
 * @return boolean
 */
public function lock()
{
    return DB::SELECT("SELECT GET_LOCK('lock_name', 1) AS result")[0]->result === 1;
}

/**
 * Release a DB lock, returns true if successful.
 *
 * @return boolean
 */
public function unlock()
{
    return DB::select("SELECT RELEASE_LOCK('lock_name') AS result")[0]->result === 1;
}

Answer 1

您似乎没有竞争条件，因为 ID 来自导入文件，并且如果您的导入算法正常工作，那么每个线程都有自己的工作分片要完成并且应该永远不要与他人发生冲突。现在似乎有 2 个线程正在接收创建同一患者的请求，并且由于算法错误而相互冲突。

确保每个生成的线程都从导入文件中获取一个新行，并且仅在失败时重复。

如果你不能这样做，并且想坚持使用互斥锁，使用文件锁似乎不是一个很好的解决方案，因为现在你解决了应用程序中的冲突，而它实际上发生在你的数据库中。数据库锁也应该快得多，并且总体上是一个更体面的解决方案。

请求一个数据库锁，像这样：

$db -> exec('LOCK TABLES table1 WRITE, table2 WRITE');

当您写入锁定的表时，您可能会遇到 SQL 错误，因此请在您的 Patient->save() 周围加上 try catch。

更好的解决方案是使用条件原子查询。一个数据库查询，其中也有条件。你可以使用这样的查询：

INSERT INTO targetTable(field1) 
SELECT field1
FROM myTable
WHERE NOT(field1 IN (SELECT field1 FROM targetTable))

Answer 2

您的示例代码将阻止第二个请求，直到第一个请求完成。您需要为flock() 使用LOCK_NB 选项来立即返回错误而不是等待。

是的，您可以在文件系统级别或直接在数据库中使用锁定或信号量。

在您需要每个导入文件只处理一次的情况下，最好的解决方案是为每个导入文件创建一个包含行的 SQL 表。在导入开始时，您插入正在导入的信息，因此其他线程将知道不再处理它。导入完成后，您将其标记为这样。 （然后几个小时后，您可以检查表格以查看导入是否真的完成。）

此外，最好在单独的脚本上执行此类一次性持久的操作，例如在单独的脚本上导入，而不是在向访问者提供普通网页时。例如，您可以安排每晚的 cron 作业，该作业将获取导入文件并进行处理。

Answer 3

我看到三个选项：
- 使用互斥量/信号量/其他一些标志 - 不容易编码和维护
- 使用DB内置事务机制
- 使用队列（如 RabbitMQ 或 0MQ）将消息连续写入 DB