PHP/MySQL 作业队列系统多次执行作业

Question

我使用 PHP、MySQL 和 cron 想出了一个非常简单的作业排队系统。

1. Cron 将调用一个网站，该网站有一个每 2 秒调用一次函数 A() 的函数。 A() 从表 A
中搜索并检索行
2. 检索行后，A() 将使用 working 列中的值 1 更新该行
3. A() 然后对检索到的行中的数据执行某些操作
4. A() 然后在表 B 中插入一行，其中包含处理步骤 3 期间获得的值。

问题:我注意到，由于函数 A() 从表中检索同一行，表 B 中有时会存在重复值A 多次。

上面的设计的哪一部分允许重复处理，应该如何解决？

请不要建议像rabbitMQ这样的东西，至少要展示如何更详细地实现它。我读了他们的一些文档，但不明白如何实现它。谢谢!

更新:我有一个 cron 作业，每分钟调用一个页面(调用函数 c())。此函数 c() 循环 30 次，调用函数 A()，并使用 sleep() 进行延迟。

Answer 1

提供的答案很好，文件锁工作得很好，但是，由于您使用的是 MySQL，我想我也会回答。通过 MySQL，您可以使用 GET_LOCK 实现协作异步锁定。和 RELEASE_LOCK .

*DISCLAIMER: The examples below are untested. I have successfully implemented something very close to this before, and the below was the general idea.

假设您已将此 GET_LOCK 函数包装在名为 Mutex 的 PHP 类中:

class Mutex {
  private $_db = null;
  private $_resource = '';

  public function __construct($resource, Zend_Db_Adapter $db) {
    $this->resource = $resource;
    $this->_db      = $db;
  }

  // gets a lock for $this->_resource; you could add a $timeout value,
  // to pass as a 2nd parameter to GET_LOCK, but I'm leaving that
  // out for now
  public function getLock() {
    return (bool)$this->_db->fetchOne(
      'SELECT GET_LOCK(:resource)',
      array(
        ':resource' => $this->_resource
      ));
  }

  public function releaseLock($resource) {
    // using DO because I really don't care if this succeeds; 
    // when the PHP process terminates, the lock is released
    // so there is no worry about deadlock
    $this->_db->query(
      'DO RELEASE_LOCK(:resource)',
      array(
        ':resource' => $resource
      ));
  }
}

在 A() 从表中获取方法之前，让它请求锁定。您可以使用任何字符串作为资源名称。

class JobA {
  public function __construct(Zend_Db_Adapter $db) {
    $this->_db = $db;
  }

  public function A() {
    // I'm assuming A() is a class method and that the class somehow
    // acquired access to a MySQL database - pretend $this->db is a 
    // Zend_Db instance. The resource name can be an arbitrary 
    // string - I chose the class name in this case but it could be 
    // 'barglefarglenarg' or something.
    $mutex = new Mutex($this->db, get_class($this));

    // I choose to throw an exception but you could just as easily 
    // die silently and get out of the way for the next process, 
    // which often works better depending on the job
    if (!$mutex->getLock())
      throw new Exception('Unable to obtain lock.');

    // Got a lock, now select the rows you need without fear of 
    // any other process running A() getting the same rows as this
    // process - presumably you would update/flag the row so that the
    // next A() process will not select the same row when it finally
    // gets a lock. Once we have our data we release the lock

    $mutex->releaseLock();

    // Now we do whatever we need to do with the rows we selected
    // while we had the lock
  }
}

当您设计一个场景，其中多个进程选择和修改相同的数据时，这种事情会非常方便。使用 MySQL 时，我更喜欢这种数据库方法而不是文件锁定机制，以实现可移植性 - 如果锁定机制位于文件系统外部，则可以更轻松地将应用程序托管在不同平台上。当然可以做到，而且效果很好，但根据我个人的经验，我发现这更容易使用。

如果您计划让您的应用程序跨数据库引擎移植，那么这种方法可能不适合您。