我希望能够读取来自Queue.Queue或TCP套接字的消息,无论哪个先到达。
如何在不使用2个线程的情况下实现?
平台是Windows上的CPython 2.7.5。
这里有一个非常好的技巧可以解决你的问题,点击这里查看。
import queue
import socket
import os
class PollableQueue(queue.Queue):
def __init__(self):
super().__init__()
# Create a pair of connected sockets
if os.name == 'posix':
self._putsocket, self._getsocket = socket.socketpair()
else:
# Compatibility on non-POSIX systems
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('127.0.0.1', 0))
server.listen(1)
self._putsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._putsocket.connect(server.getsockname())
self._getsocket, _ = server.accept()
server.close()
def fileno(self):
return self._getsocket.fileno()
def put(self, item):
super().put(item)
self._putsocket.send(b'x')
def get(self):
self._getsocket.recv(1)
return super().get()
select
而不是非阻塞套接字 I/O,因为如果您需要从多个套接字读取数据,它会更加简洁...import socket
import select
import Queue
import time
TIMEOUT = 0.1 # 100ms
def process_queue_item(item):
print 'Got queue item: %r' % item
def process_socket_data(data):
print 'Got socket data: %r' % data
def main():
# Build queue
queue = Queue.Queue()
for i in range(10):
queue.put(i)
queue.put(None) # Using None to indicate no more data on queue
queue_active = True
# Build socket
sock = socket.socket()
sock.connect(('www.google.com', 80))
sock.send('GET / HTTP/1.0\r\n\r\n')
socket_active = True
# Main event loop
while 1:
# If there's nothing to read, bail out
if not (socket_active or queue_active):
break
# By default, sleep at the end of the loop
do_sleep = True
# Get data from socket without blocking if possible
if socket_active:
r, w, x = select.select([sock], [], [], TIMEOUT)
if r:
data = sock.recv(64)
if not data: # Hit EOF
socket_active = False
else:
do_sleep = False
process_socket_data(data)
# Get item from queue without blocking if possible
if queue_active:
try:
item = queue.get_nowait()
if item is None: # Hit end of queue
queue_active = False
else:
do_sleep = False
process_queue_item(item)
except Queue.Empty:
pass
# If we didn't get anything on this loop, sleep for a bit so we
# don't max out CPU time
if do_sleep:
time.sleep(TIMEOUT)
if __name__ == '__main__':
main()
Got socket data: 'HTTP/1.0 302 Found\r\nLocation: http://www.google.co.uk/\r\nCache-Co'
Got queue item: 0
Got socket data: 'ntrol: private\r\nContent-Type: text/html; charset=UTF-8\r\nSet-Cook'
Got queue item: 1
Got socket data: 'ie: PREF=ID=a192ab09b4c13176:FF=0:TM=1373055330:LM=1373055330:S='
Got queue item: 2
etc.
TIMEOUT
值。如果你准备使用多个线程,有很多选择。如果每个线程都能处理自己接收到的数据,那么我会将每个阻塞调用放在它自己的线程中。 - Ayadef check_for_message(queue,socket,sock_accept_size=512):
socket.setblocking(0)
while True:
try:
sock_msg=socket.recv(sock_accept_size)
except socket.error:
"""Do stuff if there is no message"""
sock_msg=None
try:
que_msg=queue.get()
except Queue.Empty:
"""Do stuff if there is no message"""
que_msg=None
yield (que_msg,sock_msg)
然后您可以使用以下方式进行迭代:
for que_message,sock_message in check_for_message(que_instance,socket_instance):
print que_message,sock_message
select(2)
这样的东西适用于两个阻塞套接字,但不适用于队列。如果可以接受的话,最接近的方法是在循环中使用非阻塞方法。 - Aya