(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()，为了克服上面提到的问题，我将对进程进行分叉，在子进程中打开管道以供写入，这样父进程在打开管道时就不会阻塞。
这是一些代码
<code>
<?php
if (pcntl_fork() == 0)
{
// 子进程
$file = fopen("JobsQueue","w");
sleep(1);
exit(0);
}
else
{
usleep(15); // 确保子进程先执行
$file = fopen("JobsQueue","r");
}
echo "opened the queue \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 "I am not blocked!";
}
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
接口 Communication {
public function receive($bytes = 1024);
public function getData();
public function clearData();
public function send($data);
}
类 FIFOCommunication 实现 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();
?>