如何在不断检查输入线程的同时运行后台程序？

Question

我有小型服务器和客户端Python脚本，客户端发送一个字符串，服务器以相反的方式响应。客户端输入退出字符串时，客户端退出，然后服务器退出。

我希望服务器的“接收，反向，发送”过程在后台运行，而程序不断检查标准输入的退出字符串。

我试过使用threading，但由于阻塞，许多套接字调用导致它无法正常工作。

这样你就可以知道我已经做了什么。

server.py：

import socket 
from time import sleep 

sock = socket.socket() 
sock.bind(("127.0.0.1",12346)) 
sock.listen(3) 
print "Waiting on connection" 
conn = sock.accept() 
print "Client connected" 

while True: 
    m = conn[0].recv(4096) 
    if m == "exit": 
     sleep(1) 
     break 
    else: 
     conn[0].send(m[::-1]) 

sock.shutdown(socket.SHUT_RDWR) 
sock.close() 

client.py：

import socket 

sock = socket.socket() 
sock.connect(("127.0.0.1",12346)) 

while True: 
    s = raw_input("message: ") 
    sock.send(s) 

    if s == "exit": 
     print "Quitting" 
     break 

    print sock.recv(4096) 

sock.shutdown(socket.SHUT_RDWR) 
sock.close() 

Answer 1

既然你希望服务器进程能够同时在同一时间接收输入来处理客户端从服务器的stdin，您可以将当前的整个服务器代码放入Thread，然后等待stdin的输入。

import socket 
from time import sleep 
import threading 

def process(): 
    sock = socket.socket() 
    sock.bind(("127.0.0.1",12346)) 
    sock.listen(3) 
    print "Waiting on connection" 
    conn = sock.accept() 
    print "Client connected" 

    while True: 
     m = conn[0].recv(4096) 
     conn[0].send(m[::-1]) 

    sock.shutdown(socket.SHUT_RDWR) 
    sock.close() 

thread = threading.Thread(target=process) 
thread.daemon = True 
thread.start() 
while True: 
    exit_signal = raw_input('Type "exit" anytime to stop server\n') 
    if exit_signal == 'exit': 
     break 

并且您可以删除客户端中的“退出”检查。

在此代码中，服务器在客户端断开连接后将不会执行任何操作，但它只会等待“退出”在stdin中输入。您可能需要扩展代码以使服务器能够接受新客户端，因为您不希望客户端有能力关闭服务器。在这种情况下，您可以将另一个while循环从conn = sock.accept()更改为sock.close()。

而@usmcs建议，如果您没有任何其他命令要发送到服务器，它会更好，如果你使用CTRL-C（KeyboardInterrupt），所以你不需要线程，而它仍然可以结束服务器正常（意味着没有误差由于CTRL-C报道）与此代码：

import socket 
from time import sleep 
import threading 

sock = socket.socket() 
sock.bind(("127.0.0.1",12346)) 
sock.listen(3) 
print "Waiting on connection" 
conn = sock.accept() 
print "Client connected" 

while True: 
    try: 
     m = conn[0].recv(4096) 
     conn[0].send(m[::-1]) 
    except KeyboardInterrupt: 
     break 

sock.close() 

Answer 2

这是非阻塞套接字接收的例子。如果没有数据接收套接字将抛出异常。

import sys 
import socket 
import fcntl, os 
import errno 
from time import sleep 

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
s.connect(('127.0.0.1',9999)) 
fcntl.fcntl(s, fcntl.F_SETFL, os.O_NONBLOCK) 

while True: 
    try: 
     msg = s.recv(4096) 
    except socket.error, e: 
     err = e.args[0] 
     if err == errno.EAGAIN or err == errno.EWOULDBLOCK: 
      sleep(1) 
      print 'No data available' 
      continue 
     else: 
      # a "real" error occurred 
      print e 
      sys.exit(1) 
    else: 
     # got a message, do something :) 

这里是标准输入非阻塞的例子阅读：

import sys 
import select 

# If there's input ready, do something, else do something 
# else. Note timeout is zero so select won't block at all. 
while sys.stdin in select.select([sys.stdin], [], [], 0)[0]: 
    line = sys.stdin.readline() 
    if line: 
    something(line) 
    else: # an empty line means stdin has been closed 
    print('eof') 
    exit(0) 
else: 
    something_else() 

基本上，你希望将它们组合，并且可以添加一些超时强制定期读取标准输入在很多的情况下，连接。

Answer 3

我以前发布过的用于在Python中构建预分叉的JSON-RPC服务器并修改代码以解决此问题的要点。要点在这里：https://gist.github.com/matthewstory/4547282

$ python server.py localhost 9999 5 
exit 
$ 

查看更多为什么这个工程。主叉派生N许多叉（在上面5的示例），其中的每一个进入一个接受循环：

# simple pre-fork server, fork before accept 
for i in range(int(argv[2])): 
    # fork our current process 
    pid = os.fork() 

    # if we are the child fork ... 
    if 0 == pid: 
     # die without unhandled exception 
     for signum in (signal.SIGINT, signal.SIGTERM,): 
      signal.signal(signum, _gogentle) 

     # under the hood, this calls `socket.accept` 
     s.serve_forever() 
     os._exit(0) 

    # if we are the papa fork 
    else: 
     _PIDS.append(pid) 

这些子叉将处理任何传入请求localhost:9999。主叉然后下降到一个选择/ waitpid函数组合循环：

# setup signal relaying for INT and TERM 
for signum in (signal.SIGINT, signal.SIGTERM,): 
    signal.signal(signum, _kronos) 

# wait on the kids 
while len(_PIDS): 
    # 1s timeout here means we're checking for exiting children at most 
    # 1x per second, prevents a busy loop 
    reads, _, _ = select.select([sys.stdin], [], [], 1) 
    if sys.stdin in reads: 
     # blocking, read 1 line 
     cmd = sys.stdin.readline() 
     # kill ourselves ... kronos will propegate 
     if cmd.strip() == 'exit': 
      os.kill(os.getpid(), signal.SIGTERM) 

    # check for exited children, non-blocking 
    while True: 
     pid, rc = os.waitpid(-1, os.WNOHANG) 
     if not pid: 
      break 
     _PIDS.remove(pid) 

的select要么指示stdin是准备进行读取，在这种情况下，我们将从stdin读取1线，或至多后它就会超时1s，在这种情况下，它将直接转入我们对任何已退出子女的支票（使用os.waitpid和WNOHANG标志）。