我的Unix系統(tǒng)上的

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

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

這是我設(shè)法做到的最好的：

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)

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

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

UNIX domain sockets,因?yàn)樗鼈冎С謹(jǐn)?shù)據(jù)報(bào)：

import errno, fcntl, os, socket

服務(wù)器：

# 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而不是使用非阻塞套接字.

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。