PHP队列文件实现答案

【问题标题】：PHP queue file implementationPHP队列文件实现
【发布时间】：2011-10-24 19:28:03
【问题描述】：

对于我正在处理的项目，我需要一个队列，该队列太大而无法容纳在正常内存中。我一直将它实现为一个简单的文件，它将读取整个文件，获取前几行（〜100）行，处理它们，然后写回更新的队列，添加新指令并删除旧指令。但是，由于队列变得太大而无法像这样保存在内存中，我需要一些不同的东西。最好有人可以告诉我一种方法来剥离文件的前几行，而无需查看其余数据。我曾考虑过使用数据库（MySQL 可能带有排序的插入时间戳），但出于负载和带宽的原因，我更愿意这样做（多台服务器必须全部从数据库发送和接收大量数据）。我正在使用的语言是 PHP，但我想这个问题更多的是关于 unix 文件。任何帮助，将不胜感激。

【问题讨论】：

"多个服务器都必须从数据库发送和接收大量数据" - 这正是数据库服务器设计（和优化）的目的。
我知道数据库可以处理它，但我更关心的是让这个进程不断运行并在我的所有服务器之间拍摄数据所涉及的带宽...... MySQL 方式是我最后的手段但如果有人知道一种方法来更新文件以删除前几行而不必查看所有数据，我真的更愿意。

标签： php file unix memory queue

【解决方案1】：

提取文件的第一行非常简单（fopen() 后跟 fgets()）。重写文件以删除已完成的作业会非常痛苦，尤其是当您有多个并发服务器处理同一个队列文件时。

另一种方法是为每个作业使用单独的文件。如果您有一些并发安全的方法来为这些文件生成递增的 ID，那么只需为最旧的作业挑选具有最低 id 的文件，并为每个新作业生成一个新的 id 就很简单了。但是，您必须弄清楚一些文件锁定，以保持两个以上的服务器同时抓取同一个文件。

【讨论】：

我实际上是在使用锁定——我真正需要的是某种方式将表示文件开始的指针向下移动到已经执行的行上。
您可以使用ftell() 获取队列文件中“从这里开始”标记的位置，并将其存储在单独的文件中。在大约 10 字节的小文件上锁定/读取/更新/写入/解锁会相当快，并且让您可以轻松地直接滚动到队列中的相关位置。
好吧，这听起来更像我正在寻找的......队列文件的实际大小会继续增长，所以我不得不在某个时候将它清空一个内存大小的块时间...

【解决方案2】：

我在处理 enqueue/fs 传输时遇到了同样的问题。我未能在文件请求时修改一小部分，而没有将其复制到内存并保存回来。取而代之的是，但可以在文件末尾执行此操作。您可以阅读一部分，然后将其截断。那不是真正的队列，而是堆栈。因此，如果您依赖消息排序，这将不是一个解决方案。就我而言，当从文件中读取文件时，我锁定了文件，锁定被释放。

这是您可以将消息写入队列文件的方式：

<?php
$rawMessage = 'this your message to put to the queue as a string';

$queueFile = fopen('/path/to/queue/file', '+a');

// here it may add some spaces so the message length is multiples of modular.
// that make it easier to read messages from a file.

// lock file

$rawMessage = str_repeat(' ', 64 - (strlen($rawMessage) % 64)).$rawMessage;
fwrite($queueFile, $rawMessage);

// release lock

这是从队列文件中读取消息的方法：

<?php

$queueFile = fopen('/path/to/queue/file', '+c');

// lock file

$frame = readFrame($file, 1);
ftruncate($file, fstat($file)['size'] - strlen($frame));
rewind($file);
$rawMessage = substr(trim($frame), 1);

// release lock


function readFrame($file, $frameNumber)
{
    $frameSize = 64;
    $offset = $frameNumber * $frameSize;
    fseek($file, -$offset, SEEK_END);
    $frame = fread($file, $frameSize);
    if ('' == $frame) {
        return '';
    }
    if (false !== strpos($frame, '|{')) {
        return $frame;
    }
    return readFrame($file, $frameNumber + 1).$frame;
}

对于锁定，我建议使用Symfony LockHandler 或直接使用 enqueue/fs。

【讨论】：