(PHP 4, PHP 5, PHP 7, PHP 8)
posix_mkfifo — 创建FIFO特殊文件（命名管道）
posix_mkfifo() 创建一个特殊的FIFO文件，该文件存在于文件系统中，并充当进程的双向通信端点。
filenameFIFO文件的路径。
permissions第二个参数permissions必须以八进制表示法给出（例如0644）。新创建的FIFO的权限也取决于当前umask()的设置。创建的文件的权限为（mode & ~umask）。
对于对“半连接”管道（使用/usr/bin/mkfifo、posix_mkfifo等创建）进行非阻塞、fopen’d读取访问，我只需执行以下操作
<?php
$fh=fopen($fifo, "r+"); // 确保至少有一个写入者（我们），因此将是非阻塞的
stream_set_blocking($fh, false); // 防止fread/fwrite阻塞
?>
"r+"允许fopen立即返回，而不管外部写入者通道如何。然后，您必须使用您自己的约定来跟踪$fh作为伪只读资源，因为fwrite在技术上也是允许的。我已经在Linux上使用PHP 4.3.10和PHP 5.2.4成功地使用了这种方法，包括半连接（尚未写入者）和预连接（写入者已经在等待）管道，并像往常一样使用stream_select轮询。这是一个可能的解决方案，针对- tech at kwur dot com- 提到的问题
我遇到了一个问题，我的一个进程（一个服务器）需要处理套接字连接，同时从数据库获取一些数据。我不想让客户端等待查询执行时间，所以我决定创建一个单独的进程来执行数据库上的查询，这两个进程将通过管道进行通信。当然，如果数据不可用，我不希望服务器阻塞。所以我想到的是使用stream_select()，为了克服上述问题，我会fork进程，在子进程中打开管道进行写入，这样父进程在打开管道时就不会阻塞。
这里有一些代码
<code>
<?php
if (pcntl_fork() == 0)
{
// 子进程
$file = fopen("JobsQueue","w");
sleep(1);
exit(0);
}
else
{
usleep(15); // 确保子进程先执行
$file = fopen("JobsQueue","r");
}
echo "已打开队列 \n";
while (true)
{
$reders = array($file);
if (stream_select($reders,$writers=null,$except=null,0,15) < 1)
{
continue;
}
else
{
// 从FIFO读取数据
$data = fread($file,1024);
echo $data;
}
sleep(1);
}
?>
</code>拥有一个非阻塞管道读取器的方法是首先检查管道是否存在。如果存在，则从管道读取，否则执行其他操作。这将在写入者创建管道、写入管道并在之后删除管道的假设下工作。
这是一个阻塞写入器
<?php
$pipe="/tmp/pipe";
$mode=0600;
if(!file_exists($pipe)) {
// 创建管道
umask(0);
posix_mkfifo($pipe,$mode);
}
$f = fopen($pipe,"w");
fwrite($f,"hello"); // 直到有读取器才阻塞
unlink($pipe); // 删除管道
?>
这是一个非阻塞读取器
<?php
$pipe="/tmp/pipe";
if(!file_exists($pipe)) {
echo "我没有阻塞！";
}
else {
// 阻塞并从管道读取
$f = fopen($pipe,"r");
echo fread($f,10);
}
?>这仍然不是解决方案：如果我在管道上监听命令并在单独的管道上输出状态，PHP 将阻塞这两个打开操作，因为其他进程尚未连接到此管道。因为我无法使用底层的 fcntl() 来设置 O_NONBLOCK 或类似的东西，所以这总是会锁住，非常愚蠢。我唯一能使其工作的方法是使用 system() 生成单独的子 shell 并让它们分别执行 cat 或 echo，然后管道就能正常工作……通常情况下？我们无法在打开时设置阻塞，这实在太麻烦了！注释（摘自 Debian Linux 上的 `man 7 pipe`）
“在某些系统（但不是 Linux）上，管道是双向的：数据可以在管道两端双向传输。根据 POSIX.1-2001，管道只需要单向即可。可移植的应用程序应避免依赖双向管道语义。”
Linux 管道不是双向的。
此外，在我看来，在 php 中使用 fifo（命名）管道几乎毫无意义，因为似乎无法确定打开（更不用说读取）它是否会阻塞。stream_select 应该能够做到这一点，不幸的是，你无法达到这一点，因为即使尝试打开管道进行读取也会阻塞，直到有写入者。
我甚至尝试使用 popen("cat $name_of_pipe", 'r')，它也阻塞了，直到另一个进程打开它进行写入。面向对象的 FIFO 通信进程
<?php
interface Communication {
public function receive($bytes = 1024);
public function getData();
public function clearData();
public function send($data);
}
class FIFOCommunication implements Communication {
private $fifos;
private $data;
static public function stream_fifo_open($fifoPath, $mode) {
if (pcntl_fork() == 0) {
if (! file_exists($fifoPath)) {
posix_mkfifo($fifoPath, $mode);
}
$fifo = fopen($fifoPath, 'w');
sleep(1);
exit(0);
}
else {
usleep(15);
return fopen($fifoPath, 'r');
}
}
static public function stream_fifo_write($fifoPath, $mode, $data) {
if (pcntl_fork() == 0) {
if (! file_exists($fifoPath)) {
posix_mkfifo($fifoPath, $mode);
}
$fifo = fopen($fifoPath, 'w');
fwrite($fifo, $data);
exit(0);
}
}
public function __construct(
$fifoInputName, $fifoInputMode,
$fifoOutputName, $fifoOutputMode
) {
$this->fifos = array(
'input' => array(
'file' => self::stream_fifo_open($fifoInputName, $fifoInputMode),
'mode' => $fifoInputMode,
'name' => $fifoInputName,
'use' => 'r',
),
'output' => array(
'mode' => $fifoOutputMode,
'name' => $fifoOutputName,
'use' => 'w'
)
);
}
public function remove($type = null) {
switch ($type) {
case 'input':
@unlink($this->fifos['input']['name']);
break;
case 'output':
@unlink($this->fifos['output']['name']);
break;
default:
@unlink($this->fifos['input']['name']);
@unlink($this->fifos['output']['name']);
}
}
public function receive($bytes = 1024) {
$readers = array($this->fifos['input']['file']);
if (stream_select($readers, $writers = null, $except = null, 0, 15) == 1) {
$data = fread($this->fifos['input']['file'], $bytes);
}
if (! empty($data)) {
$this->data .= $data;
return true;
}
else {
return false;
}
}
public function getData() {
return $this->data;
}
public function clearData() {
$this->data = null;
}
public function send($data) {
$fifoOutput = & $this->fifos['output'];
self::stream_fifo_write($fifoOutput['name'], $fifoOutput['mode'], $data);
}
}
$fifoCommunication = new FIFOCommunication(
getmypid() . '.input', 0600,
getmypid() . '.output', 0600
);
echo "COMMUNICATION STARTED\n";
while (true) {
if ($fifoCommunication->receive()) {
$data = $fifoCommunication->getData();
if ($data == "EXIT\n") {
break;
}
else {
$fifoCommunication->send('RECEIVED: ' . $fifoCommunication->getData());
}
}
sleep(1);
}
$fifoComunication->remove();
?>