使用 MySQL 数据库处理 PHP Symfony 中的队列竞争条件答案

【问题标题】：Handle Queue race condition in PHP Symfony with MySQL Database使用 MySQL 数据库处理 PHP Symfony 中的队列竞争条件
【发布时间】：2017-01-13 18:05:27
【问题描述】：

我在 Symfony 中有一个应用程序，需要从应用程序发送电子邮件/通知。由于电子邮件/通知发送过程需要时间，所以我决定将它们放入队列并定期处理队列。因此，我可以减少涉及电子邮件/通知调度的请求的响应时间。

处理队列的 Cron 作业（一个 php 脚本 - Symfony 路由）每 30 秒运行一次，并检查是否有任何未发送的电子邮件/通知，如果发现它会从队列表中获取所有数据并开始发送它们。发送电子邮件/通知时，队列表行状态标志会更新以显示它已发送。

现在，当队列中有更多电子邮件可能需要 30 多秒才能发送时。另一个 Cron 作业也开始运行并开始从队列中发送电子邮件。因此导致重复发送电子邮件/通知。

我的电子邮件队列表结构如下：

|-------------------------------------|
| id | email | body | status | sentat |
|-------------------------------------|

我解决这个问题的想法如下：

在数据库中设置一个 Cron 作业正在运行的标志，如果发现该标志设置，其他 Cron 作业不应继续。
将所有记录的状态更新为“已发送”，然后开始发送电子邮件/通知。

所以我的问题是，是否有任何有效的方法来处理队列？是否有任何 Symfony 捆绑包/功能来执行此类特定任务？

【问题讨论】：

如果您对工作人员排队感兴趣，请尝试Beanstalkd。还有 upstart、supervisor、runit 以及 Gearman 和 RabbitMQ，尽管这些对于一般邮件队列服务来说要复杂一些。

标签： php mysql symfony cron

【解决方案1】：

关于您的建议：

如果 cronjob 进程死亡（无论出于何种原因）并且无法清理标志怎么办？我认为旗帜不是一个好主意。如果您想采用这种方法，则不应使用布尔值，而应使用进程 ID 或时间戳，以便您可以检查进程是否仍然存在，或者它是否在很久以前就开始了，但没有进行清理。
同样的问题：如果进程死了怎么办？您不想在邮件发送之前将它们标记为已发送。

我想我可能会使用两个字段：一个用于将记录标记为“正在发送”（从而告诉其他进程跳过该记录），另一个用于将其标记为“发送成功完成”。我会给两者都写一个时间戳，这样我就可以（自动或手动）找到那些“正在发送”在过去 > X 秒的记录，这将是一个死亡进程的指标。

【讨论】：

我对这两个想法的想法完全一样，所以没有实现它们。但是有 2 个字段，是否有一种技术不需要手动检查。由于手动检查后发送时造成的延迟会非常显着。
两种可能的方法： 1. 当 cronjob 检测到记录被标记为正在进行中 > 5 分钟。之前（或您认为合适的任何时间），将其视为未标记。 2.不要使用时间戳，而是使用进程ID；如果 cronjob 找到一个“未完成”记录，其中包含不存在任何进程的进程 ID，则该记录是“孤立的”并且可以安全地取消标记。

【解决方案2】：

您可以在此处使用数据库事务。休息将由数据库锁定机制和并发控制来处理。通常，您给出的任何 DML/DCL/DDL 命令都被视为独立事务。在您的问题中，如果第二个 cron 作业将读取行（在第一个 cron 作业将其更新为已发送之前），它将找到未发送的电子邮件，并尝试再次发送。并且在第二个 cron 作业将其更新为已发送之前，如果第三个作业发现它未发送，它将执行相同的操作。所以它会给你带来很大的问题。

无论你采取什么方法，都会有竞争条件。所以让数据库允许这样做。有很多并发控制方法可以参考。

BEGIN_TRANSACTION

/* Perform your actions here. N numbers of read/write */

END_TRANSACTION

这个解决方案仍然存在一个问题。你会发现在某一阶段，当读/写操作的次数增加时，仍然存在一些不一致。

这里是数据库的隔离级别，它定义了 2 个事务相互隔离的程度，以及如何安排它们同时运行。

您可以根据需要设置隔离级别。请记住，并发性与隔离级别成反比。所以分析你的读/写语句，找出你需要的级别。不要使用更高的级别。我提供了一些链接，这可能会对您有所帮助

http://www.ibm.com/developerworks/data/zones/informix/library/techarticle/db_isolevels.html

Difference between read commit and repeatable read

http://dev.mysql.com/doc/refman/5.7/en/innodb-transaction-isolation-levels.htm

如果您可以在此处发布您的数据库操作。我可以建议你一些可能的隔离级别

【讨论】：

【解决方案3】：

所以我的问题是，是否有任何有效的方法来处理队列？是否有任何 Symfony Bundle/Feature 可以执行此类特定任务？

您可以使用enqueue-bundle 加上doctrine dbal transport。

它已经处理了竞争条件和其他问题。

【讨论】：