Python os.mkfifo() 方法（建议收藏）

💡一则或许对你有用的小广告

欢迎加入小哈的星球 ，你将获得：专属的项目实战 / 1v1 提问 / Java 学习路线 / 学习打卡 / 每月赠书 / 社群讨论

• 新项目:《从零手撸：仿小红书（微服务架构）》 正在持续爆肝中，基于 Spring Cloud Alibaba + Spring Boot 3.x + JDK 17...，点击查看项目介绍 ;
• 《从零手撸：前后端分离博客项目（全栈开发）》 2 期已完结，演示链接： http://116.62.199.48/ ;

截止目前， 星球 内专栏累计输出 82w+ 字，讲解图 3441+ 张，还在持续爆肝中.. 后续还会上新更多项目，目标是将 Java 领域典型的项目都整一波，如秒杀系统, 在线商城, IM 即时通讯，权限管理，Spring Cloud Alibaba 微服务等等，已有 2900+ 小伙伴加入学习 ，欢迎点击围观

命名管道：进程间通信的"快递站"

在编程世界中，进程间通信（IPC）如同城市中的快递网络，需要一种高效可靠的传输方式。Python 的 os.mkfifo() 方法正是为此而生——它允许开发者创建一种特殊的文件类型：命名管道（FIFO）。这种管道如同快递站，两端进程可以像寄存和提取包裹一样，通过固定路径实现数据的单向或双向传递。

什么是命名管道？

命名管道是一种基于文件系统的IPC机制，其核心特性包括：

1. 路径绑定：通过文件路径标识，支持跨进程、跨程序通信
2. 单向通信：默认支持单向数据流（需配合读写操作实现双向）
3. 阻塞特性：读写操作会阻塞直到另一端准备好

想象一个小区快递柜：

• 每个格口对应一个命名管道
• 发件人（写入进程）将包裹放入格口
• 收件人（读取进程）从格口取出包裹
• 格口名称（文件路径）是通信双方的共同约定

os.mkfifo() 方法详解

方法语法与参数

os.mkfifo(path, mode=0o666)  

权限模式的"魔法"

mode参数采用八进制表示，例如：

• 0o666 表示所有用户可读写（实际权限需结合系统umask计算）
• 0o777 表示完全开放权限（需谨慎使用）

类比房屋钥匙：

• mode决定谁有钥匙（读/写权限）
• umask是系统管理员设置的"权限过滤器"，会从mode中减去屏蔽位

创建管道的"施工流程"

1. 调用os.mkfifo()在文件系统中建立特殊文件节点
2. 系统分配FIFO设备类型（通过ls -l可见p标识）
3. 任何进程都可通过标准文件操作（open()、read()、write()）与之交互

$ mkfifo fifo_pipe  
$ ls -l fifo_pipe  
prw-r--r-- 1 user user 0 Jan 1 00:00 fifo_pipe  

典型应用场景

场景1：单向数据传输

import os  
fifo_path = "/tmp/data_pipe"  
os.mkfifo(fifo_path)  

with open(fifo_path, 'w') as pipe:  
    pipe.write("Hello FIFO!")  

print("Data sent successfully")  

fifo_path = "/tmp/data_pipe"  
with open(fifo_path, 'r') as pipe:  
    data = pipe.read()  
    print(f"Received: {data}")  

场景2：日志监控系统

import os  
LOG_FIFO = "/var/log/app_pipe"  

os.mkfifo(LOG_FIFO)  
with open(LOG_FIFO, 'r') as log_pipe:  
    for line in log_pipe:  
        if "ERROR" in line:  
            send_alert(line)  

import os  
LOG_FIFO = "/var/log/app_pipe"  

with open(LOG_FIFO, 'w') as log_pipe:  
    log_pipe.write("INFO: User logged in\n")  

实战演练：构建简易聊天系统

步骤分解

1. 创建双向通信管道
2. 启动两个终端进程（Alice和Bob）
3. 实现消息收发机制

import os  
FIFO_A2B = "/tmp/chat_a2b"  
FIFO_B2A = "/tmp/chat_b2a"  

os.mkfifo(FIFO_A2B)  
os.mkfifo(FIFO_B2A)  

with open(FIFO_A2B, 'w') as to_bob, open(FIFO_B2A, 'r') as from_bob:  
    while True:  
        message = input("Alice: ")  
        to_bob.write(f"Alice: {message}\n")  
        to_bob.flush()  
        response = from_bob.readline().strip()  
        print(f"\tBob: {response}")  

with open(FIFO_B2A, 'w') as to_alice, open(FIFO_A2B, 'r') as from_alice:  
    while True:  
        message = input("Bob: ")  
        to_alice.write(f"Bob: {message}\n")  
        to_alice.flush()  
        response = from_alice.readline().strip()  
        print(f"\tAlice: {response}")  

运行效果

Terminal 1 (Alice):  
python chat_server.py  
Alice: Hi Bob!  
    Bob: Hello Alice!  

Terminal 2 (Bob):  
python chat_server.py  
Bob: How are you?  
    Alice: I'm fine, thanks!  

常见问题与解决方案

问题1：PermissionError异常

现象：Permission denied错误
原因：

• 管道路径权限不足
• 目标目录不可写
• SELinux/AppArmor策略限制

解决方案：

os.mkfifo("/tmp/myfifo", 0o600)  # 仅限当前用户  

问题2：进程阻塞问题

现象：程序卡在open()或read()调用
原理：FIFO的阻塞特性要求两端同时存在
解决方法：

• 使用os.O_NONBLOCK标志
• 在多线程/多进程环境下确保端口对齐

import os  
import fcntl  

pipe = os.open("/tmp/myfifo", os.O_RDONLY | os.O_NONBLOCK)  
fcntl.fcntl(pipe, fcntl.F_SETFL, os.O_NONBLOCK)  

进阶技巧与最佳实践

技巧1：结合信号量控制

import os  
from multiprocessing import Process, Value  

fifo_path = "/tmp/protected_pipe"  
semaphore = Value('i', 1)  # 初始化信号量  

def writer():  
    os.mkfifo(fifo_path)  
    with open(fifo_path, 'w') as pipe:  
        if semaphore.value > 0:  
            pipe.write("Critical data\n")  
            semaphore.value -= 1  

def reader():  
    with open(fifo_path, 'r') as pipe:  
        data = pipe.read()  
        semaphore.value += 1  

if __name__ == "__main__":  
    p1 = Process(target=writer)  
    p2 = Process(target=reader)  
    p1.start()  
    p2.start()  
    p1.join()  
    p2.join()  

技巧2：异步I/O集成

import os  
import select  
import sys  

fifo_path = "/tmp/async_pipe"  

def main():  
    os.mkfifo(fifo_path)  
    read_fds = [sys.stdin, open(fifo_path, 'r')]  

    while True:  
        rlist, _, _ = select.select(read_fds, [], [])  
        for fd in rlist:  
            if fd == sys.stdin:  
                msg = fd.readline().strip()  
                print(f"You: {msg}")  
            else:  
                msg = fd.readline().strip()  
                print(f"Remote: {msg}")  

if __name__ == "__main__":  
    main()  

总结与展望

通过本文的深入探讨，我们不仅掌握了os.mkfifo()方法的核心用法，更理解了命名管道在进程间通信中的独特价值。这种基于文件系统的IPC机制，为需要轻量级、跨平台通信的场景提供了优雅的解决方案。

对于开发者来说，合理使用命名管道可以：

• 构建高效的日志分发系统
• 实现服务端-客户端的异步通信
• 设计分布式任务队列

未来，随着异步编程和微服务架构的普及，FIFO技术将与协程、消息队列等现代技术进一步融合。建议读者在掌握基础后，可深入探索os.open()的高级选项、select/poll的多路复用机制，以及结合asyncio框架的异步管道通信方案，从而在复杂系统设计中游刃有余。