我的Unix系統上的

Python3進程一直在運行,我希望能夠通過偶爾運行的其他進程的命名管道隨機發送數據.如果命名管道沒有數據,我希望我的進程繼續做其他事情,所以我需要檢查它是否有沒有阻塞的數據.

我無法弄清楚如何在不打開它的情況下進行檢查,但是除非我設置非阻塞標志,否則打開塊.如果我設置了標志,如果我在讀取之前或讀取期間碰巧寫入管道,它就會崩潰.

這是我設法做到的最好的：

import os

fifo = "pipe_test.fifo"

done = False

fd = os.open(fifo, os.O_RDONLY | os.O_NONBLOCK)

while not done:

try:

s = os.read(fd, 1024) # buffer size may need tweaking

print(s)

done = True

except BlockingIOError as e:

pass

os.close(fd)

如果管道中沒有數據,我會得到b“”,然后退出.如果管道中有數據,它會獲得一次異常,重試,然后獲取數據.好像我做錯了什么,可能會遇到奇怪的競爭條件.有沒有更好的方法來做到這一點？

最佳答案 如果您可以更改客戶端的代碼,我不會使用命名管道,而是

UNIX domain sockets,因為它們支持數據報：

import errno, fcntl, os, socket

服務器：

# bind socket

sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)

sock.bind('pipe_test.fifo')

# set socket non-blocking

fcntl.fcntl(sock.fileno(), fcntl.F_SETFL, os.O_NONBLOCK)

# get a datagram

try:

datagram = sock.recv(1024)

except (OSError, socket.error) as ex:

if ex.errno not in (errno.EINTR, errno.EAGAIN):

raise

else:

print('Datagram: %r' % datagram)

客戶：

sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)

sock.sendto('Hello!', 'pipe_test.fifo')

但您可能希望查看multithreading而不是使用非阻塞套接字.

總結

以上是生活随笔為你收集整理的python 命名管道_Python：检查命名管道是否有数据的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。