这篇文章主要介绍服务器架构。
网络服务需要面对两个挑战。
- 第一个问题是核心挑战，要编写出能够正确处理请求并构造合适响应的代码。
- 第二个挑战是如何将网络代码部署到随系统自动启动的Windows服务或者是Unix守护进程中，将活动日志持久化存储。并且在无法连接到数据库或者后端存储区时发出警告，为其提供完整的保护，以防止所有可能的失败情形，或是确保其在失败时快速重启。
这篇文章直重点说第一个问题。然后会介绍服务器部署，然后把重点放在如何构建网络服务器软件上。
首先，我们可以很自然的把网络服务器分为三大类。
第一类就是简单的单线程服务器（比如UDP服务器和TCP服务器），在这里会详细说明这类服务器的局限性，即同一时刻只能为一个客户端服务，此时其他客户端只能等待。即使为一个客户服务，这时候CPU也可能处于近乎空闲的状态。
第二类就是解决局限性的一个方案，使用多个线程或者进程，每个线程或者进程内都运行一个单线程服务器。
第三类就是与第二类刚好相对的另一种解决方案，在自己的代码中使用异步网络操作来支持多路复用，而不直接使用操作系统提供的多路复用。
然后我们说一下服务器部署，我们可能会把网络服务部署到单台机器上，也可能部署到多台机器上。要使用单台机器上的服务，客户端只要链接到该机器的IP地址即可，而要使用运行在多台机器上的服务，就需要更加复杂的方法。
一种方法是将这个服务的某个实例的地址或者主机名返回给客户端（比如与客户端运行在同一机房中服务实例），但是这种方法没有提供冗余性，如果服务的这一个实例宕机了，那么通过主机名或者IP地址硬编码链接这个服务实例的客户端都无法继续链接。
另外一种更加健壮性的方法就是，当要访问某个服务时候，令DNS服务器返回运行这个服务的所有IP地址，如果客户端无法连接到第一个地址的话，可以连接到第二个地址，然后第三个。工业界一般会在服务前配置一个负载均衡器，客户端直接连接到负载均衡器，然后由负载均衡器将链接请求转发到实际的服务器。如果某台服务器宕机了，那么负载均衡器会将转发至该服务器的连接请求予以停止。直到这个服务器恢复服务为止。这样服务器的故障对于大量的用户来说是不可见的。
大型的互联网服务中结合了这两个方法：每个机房中都配置了一个负载均衡器与服务器群，而公共的DNS名会返回与用户距离最近的机房中的负载均衡器的IP地址。
无论服务器架构多么简单或者多么复杂，都需要使用某种方式在物理或者虚拟机器上运行我们的Python服务器代码，这一个过程叫做部署。
对于部署来说，比较旧式的技术观点就是，为每个服务器程序都编写服务所提供的所有功能：通过两次fork()创建一个Unix守护进程（或者是将自己注册为一个Windows服务），安排进行系统级的日志操作，支持配置文件以及提供启动、关闭和重启的相关机制。可以使用已经解决了相关问题的第三方库来完成服务器程序的编写，也可以在自己的代码中重新实现这些功能。
另外一种方法就是。提倡只是先服务器程序必备功能的最小集合。它将每个服务实现为普通的前台程序，而不是将其实现为守护进程。这样的程序从环境变量（Python中的sys.environ字典）而不是系统级的配置文件中获取所需要的配置选项。他通过环境变量中指定的选项链接到任意的后端服务，并且直接将日志信息输出到屏幕，甚至直接使用Python自己提供的print()函数。另外，这个方法通过打开并且监听环境配置指定的任意端口来接受网络请求。
现在有一些大型的平台服务提供商提供了托管这种程序的功能。他们将应用程序的几十个甚至几百个副本配置在一个公共域名和TCP负载均衡器下，然后将所有输出的日志聚集起来进行分析。这些提供商允许我们直接提交代码。但是更多的提供商，更希望我们提供代码、Python解释器以及所需要的依赖打包入一个容器内。（说到这里，可能你会想起Docker）。我们可以在自己的笔记本电脑上对这个容器进行测试，然后将其部署到生产环境中，从而能够确认，生产环境中运行的Python代码与测试环境中运行的代码使用的是完全相同的镜像。无论是用哪种方法，都无需在单个服务中提供多个功能，服务中所有的冗余和重复都可以让平台来处理。
下面是一段示例代码。这段代码使用的是一个最简单的TCP协议进行说明。在这个协议中，客户端可以询问3个问题，这三个问题都使用纯文本的ASCII字符表示。
这三个问题都是基于The Zen of Python中的格言。可以通过import this来获取这首诗。
为了基于这个协议构建一个客户端和多个服务器，这里面定义了很多规则。这个代码本身并没有命令行接口。该程序编写的模块存在的唯一作用就是作为一个支持性模块让后续的代码导入。接下来的代码也可以用下面这段代码中定义的模式，而不需要重复编写。
import argparse, socket, time
aphorisms = {b'Beautiful is better than?': b'Ugly.',
b'Explicit is better than?': b'Implicit.',
b'Simple is better than?': b'Complex.'}
def get_answer(aphorism):
"""Return the string response to a particular Zen-of-Python aphorism."""
time.sleep(0.0) # increase to simulate an expensive operation
return aphorisms.get(aphorism, b'Error: unknown aphorism.')
def parse_command_line(description):
"""Parse command line and return a socket address."""
parser = argparse.ArgumentParser(description=description)
parser.add_argument('host', help='IP or hostname')
parser.add_argument('-p', metavar='port', type=int, default=1060,
help='TCP port (default 1060)')
args = parser.parse_args()
address = (args.host, args.p)
return address
def create_srv_socket(address):
"""Build and return a listening server socket."""
listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
listener.bind(address)
listener.listen(64)
print('Listening at {}'.format(address))
return listener
def accept_connections_forever(listener):
"""Forever answer incoming connections on a listening socket."""
while True:
sock, address = listener.accept()
print('Accepted connection from {}'.format(address))
handle_conversation(sock, address)
def handle_conversation(sock, address):
"""Converse with a client over `sock` until they are done talking."""
try:
while True:
handle_request(sock)
except EOFError:
print('Client socket to {} has closed'.format(address))
except Exception as e:
print('Client {} error: {}'.format(address, e))
finally:
sock.close()
def handle_request(sock):
"""Receive a single client request on `sock` and send the answer."""
aphorism = recv_until(sock, b'?')
answer = get_answer(aphorism)
sock.sendall(answer)
def recv_until(sock, suffix):
"""Receive bytes over socket `sock` until we receive the `suffix`."""
message = sock.recv(4096)
if not message:
raise EOFError('socket closed')
while not message.endswith(suffix):
data = sock.recv(4096)
if not data:
raise IOError('received {!r} then socket closed'.format(message))
message += data
return message
其中最后的四个函数展示了服务器进程的核心模式。这四个函数层级调用包括了监听套接字来创建TCP服务器的内容，以及关于数据封帧和错误处理的内容。
上面这段代码就是用来构建各种服务器的工具箱。
为了测试这篇文章中的服务器，需要一个客户端程序。下面这段代码提供了一个简单的命令行工具作为客户端。
import argparse, random, socket, zen_utils
def client(address, cause_error=False):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(address)
aphorisms = list(zen_utils.aphorisms)
if cause_error:
sock.sendall(aphorisms[0][:-1])
return
for aphorism in random.sample(aphorisms, 3):
sock.sendall(aphorism)
print(aphorism, zen_utils.recv_until(sock, b'.'))
sock.close()
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='Example client')
parser.add_argument('host', help='IP or hostname')
parser.add_argument('-e', action='store_true', help='cause an error')
parser.add_argument('-p', metavar='port', type=int, default=1060,
help='TCP port (default 1060)')
args = parser.parse_args()
address = (args.host, args.p)
client(address, args.e)
在正常情况下，cause_error为False。此时客户端将创建一个TCP套接字，然后发送3句格言作为请求，每发送一个就等待服务器返回相应的答案。
如果想知道如何处理输入有误的情况，客户端提供了-e的选项，用来发送不完整的问题，使服务器突然挂起。如果服务器已经启动并且正确运行，就能在客户端看到这三个问题以及相应的答案。
下面我们说下一个问题，单线程服务器。上面第一段代码中zen_utils模块提供了丰富的工具程序。减少了很多编写一个简单的单线程服务器的工作量。单线程服务器是最简单的可用设计。下面只用3行代码就可以完成这个单线程服务器。
import zen_utils
if __name__ == '__main__':
address = zen_utils.parse_command_line('simple single-threaded server')
listener = zen_utils.create_srv_socket(address)
zen_utils.accept_connections_forever(listener)
多余的事情不再详细说明了，直接说单线程服务器的缺点。
如果服务器和一个客户端进行会话期间，另外一个客户端也尝试连接服务器，这个设计就会出现问题了。只要服务器与第一个客户端的会话没有完成，新建立的链接就会一直处于操作系统的监听队列中。
对单线程服务器进行拒接服务攻击是非常容易，连接该服务器，并且永远不关闭这个链接就可以了。
而且单线程设计无法在等待客户端发送下一个请求时进行其他操作，因此很浪费服务器的CPU和系统资源。
所以，下面会说如何克服单线程服务器的这些限制。
我们说第一个方法。构建多线程与多进程服务器。服务器可以同时与多个客户端进行会话，利用操作系统的内置支持，使用多个控制线程单独运行同一段代码。可以创建多个共享相同内存空间的县城，也可以常见完全独立运行的线程。
这个方法的优点是：简洁，直接使用单线程服务器的代码，创建多个线程运行它的副本。
这个方法的缺点是：服务器能够同时通信的客户端数量收到操作系统并发机制规模的限制。即使某个客户端处于空闲状态，或者是缓慢运行状态，他也会占用整个线程或进程。就算程序被recv()阻塞,也会占用系统RAM以及进程表中的一个进程槽。当同时运行的线程数量达到几千甚至更多的时候，操作系统很少能够维持良好的表现。此时系统在切换服务的客户端时候，需要进行大量上下文切换，这使得运行效率大大降低。
每个线程都可以拥有服务器监听套接字的一个副本，并运行自己的accept()函数。操作系统会将每个新的客户端链接交给任何运行了accept()函数并处于等待的线程来处理。如果所有线程都处在繁忙状态的话，操作系统会将链接置于队列中，直到每个线程为止。
import zen_utils
from threading import Thread
def start_threads(listener, workers=4):
t = (listener,)
for i in range(workers):
Thread(target=zen_utils.accept_connections_forever, args=t).start()
if __name__ == '__main__':
address = zen_utils.parse_command_line('multi-threaded server')
listener = zen_utils.create_srv_socket(address)
start_threads(listener)
这是多线程程序的一个可能设计。主线程启动n个服务器线程。然后退出。主线程认为这n个工作线程将永远运行。因此运行这些线程的进程也会保持运行状态。
当然也有其他的可选设计。比如主线程可以保持运行，并且成为一个服务器线程。
上面介绍的是使用了操作系统级的控制线程来处理同一时刻的多个客户端对话，而且Python标准库也内置了一个框架socketsever，实现了这一个模式。
这个模块将多线程模式分为了两个：第一个是用于打开监听套接字并接受客户端链接的sever模式，第二个是用于通过某个打开的套接字与特定的客户端进行绘画的handler模式。结合这两个模式，我们需要实例化一个sever对象，然后将一个handler对象作为参数传给sever对象。下面是示例代码：
from socketserver import BaseRequestHandler, TCPServer, ThreadingMixIn
import zen_utils
class ZenHandler(BaseRequestHandler):
def handle(self):
zen_utils.handle_conversation(self.request, self.client_address)
class ZenServer(ThreadingMixIn, TCPServer):
allow_reuse_address = 1
# address_family = socket.AF_INET6 # uncomment if you need IPv6
if __name__ == '__main__':
address = zen_utils.parse_command_line('legacy "SocketServer" server')
server = ZenServer(address, ZenHandler)
server.serve_forever()
当然，我们也可以将ThreadingMixIn改为ForkingMixIn,这样就可以使用完全隔离的进程来处理链接的客户端，而不使用线程。
上述代码的一个缺点就是，不限制服务器最终启动的线程数量，这样很容易过载。
下面我们说另外一个解决方法，那就是异步服务器。如果想在不为每个客户端分配一个操作系统级的控制线程的前提下保证CPU在这段时间内处于繁忙状态。我们可以使用异步模式来编写服务器。
使用这种模式代码不需要等待数据发送至某一个特定的客户端或由这个客户端接收。相反，代码可以从整个处于等待的客户端套接字列表中读取数据。
现代操作系统网络栈让这个模式的应用成为了现实。网络栈提供了一个操作系统调用。支持进程为等待整个客户端套接字列表中的套接字而阻塞。而不只是等待一个单独的客户端套接字。另外一个特点是，可以将一个套接字配置为非阻塞套接字。
异步这个属于表示服务器代码从来不会停下来等待某一个特定的客户端，即代码的控制线程不是同步的。异步服务器可以在所有连接的客户端之前自由切换，并提供相应的服务。
操作系统也有很多调用是支持异步的代码。最古老的就是POSIX的select()调用。不过调用很多方面效率低下。
现在操作系统出现了一些select()的替代品，比如Linux上面的poll()，和BSD系统上的epoll()调用。
下面是一个简单异步服务器的完整内部细节。
import select, zen_utils
def all_events_forever(poll_object):
while True:
for fd, event in poll_object.poll():
yield fd, event
def serve(listener):
sockets = {listener.fileno(): listener}
addresses = {}
bytes_received = {}
bytes_to_send = {}
poll_object = select.poll()
poll_object.register(listener, select.POLLIN)
for fd, event in all_events_forever(poll_object):
sock = sockets[fd]
# Socket closed: remove it from our data structures.
if event & (select.POLLHUP | select.POLLERR | select.POLLNVAL):
address = addresses.pop(sock)
rb = bytes_received.pop(sock, b'')
sb = bytes_to_send.pop(sock, b'')
if rb:
print('Client {} sent {} but then closed'.format(address, rb))
elif sb:
print('Client {} closed before we sent {}'.format(address, sb))
else:
print('Client {} closed socket normally'.format(address))
poll_object.unregister(fd)
del sockets[fd]
# New socket: add it to our data structures.
elif sock is listener:
sock, address = sock.accept()
print('Accepted connection from {}'.format(address))
sock.setblocking(False) # force socket.timeout if we blunder
sockets[sock.fileno()] = sock
addresses[sock] = address
poll_object.register(sock, select.POLLIN)
# Incoming data: keep receiving until we see the suffix.
elif event & select.POLLIN:
more_data = sock.recv(4096)
if not more_data: # end-of-file
sock.close() # next poll() will POLLNVAL, and thus clean up
continue
data = bytes_received.pop(sock, b'') + more_data
if data.endswith(b'?'):
bytes_to_send[sock] = zen_utils.get_answer(data)
poll_object.modify(sock, select.POLLOUT)
else:
bytes_received[sock] = data
# Socket ready to send: keep sending until all bytes are delivered.
elif event & select.POLLOUT:
data = bytes_to_send.pop(sock)
n = sock.send(data)
if n < len(data):
bytes_to_send[sock] = data[n:]
else:
poll_object.modify(sock, select.POLLIN)
if __name__ == '__main__':
address = zen_utils.parse_command_line('low-level async server')
listener = zen_utils.create_srv_socket(address)
serve(listener)
这段事件循环代码的精髓在于，它使用了自己的数据结构来维护每个客户端会话的状态，而没有依赖操作系统在客户端活动改变时进行上下文切换。服务器有两层循环。首先是一个不断调用poll()的while循环。一次poll()调用可能返回多个事件，因此这个while循环内部还有一个循环，用于处理poll()返回的每一个事件。我们将这两层迭代隐藏在一个生成器内，这样就避免了主服务器循环因为这两次循环迭代而多用两个不必要的缩进。
然后程序维护了sockets字典。从poll()获取表示已经准备好进行后续通信对的套接字文件描述符n后，就能够根据该文件描述符从sockets字典中查找到相应的Python套接字了。我们还在里面储存了套接字的地址。这样，即使套接字关闭，操作系统也无法继续提供已经连接好的地址。也能够打印出正确的远程地址作为调试信息。
这个服务器的真正核心其实是它的缓冲区：在等待某个请求完成时，会将收到的数据存储在bytes_received字典中。在等待操作系统安排发送数据时，会将要发送的字节存储在bytes_to_send字典中。这两个缓冲区与我们告知poll()要在每个套接字上等待的时间一起形成了一个完整的状态机。用于一步一步的处理客户端会话。
- 准备连接的客户端首先会将它自身视作服务器监听套接字上的一个事件，要始终将该事件设置为POLLIN（poll input）状态。
- 当套接字本身就是客户端套接字，并且事件类型为POLLIN时。就能够使用recv()方法接收到最多为4KB的数据了。
- 套接字设置为POLLOUT后，只要客户端套接字的发送缓冲区还能够接收一个或者多个字节，那么poll()的调用就会立刻通知我们。
- 最后，如果套接字模式设置为POLLOUT后，并且send()完成了所有的数据发送，那么此时就完成了一个完整的请求-响应循环，因此将套接字模式切换为POLLIN，用于下一个请求。
- 如果客户端套接字返回了错误信息或者是错误状态，就将这个客户端的套接字以及发送缓冲区与接受缓冲区丢弃。
这个异步方法的关键之处在于，可以在一个控制线程中处理成千上万的客户端会话。当每个客户套接字准备好下一个事件时，代码就执行该套接字的下一个操作，接收或发送数据。然后立刻返回到poll()调用，监控更多事件。
如果香江注意力放在客户端代码上，而将与select()、poll()或是epoll()有关的细节交给别人去负责。就可以看一下下面说的。下面是两种风格。
第一种是回调风格的asyncio：
asynic框架支持两种编程风格。第一种风格就是使用Twisted框架时，用户通过对象实例来维护每个打开的客户端连接。在这种设计模式中，使用对象实例上的方法调用代替了上述代码中用来加速客户端会话的各步骤。
下面仍旧是读取问题，然后给出响应。回调风格的asyncio：
import asyncio, zen_utils
class ZenServer(asyncio.Protocol):
def connection_made(self, transport):
self.transport = transport
self.address = transport.get_extra_info('peername')
self.data = b''
print('Accepted connection from {}'.format(self.address))
def data_received(self, data):
self.data += data
if self.data.endswith(b'?'):
answer = zen_utils.get_answer(self.data)
self.transport.write(answer)
self.data = b''
def connection_lost(self, exc):
if exc:
print('Client {} error: {}'.format(self.address, exc))
elif self.data:
print('Client {} sent {} but then closed'
.format(self.address, self.data))
else:
print('Client {} closed socket'.format(self.address))
if __name__ == '__main__':
address = zen_utils.parse_command_line('asyncio server using callbacks')
loop = asyncio.get_event_loop()
coro = loop.create_server(ZenServer, *address)
server = loop.run_until_complete(coro)
print('Listening at {}'.format(address))
try:
loop.run_forever()
finally:
server.close()
loop.close()
可以通过该框架来获取远程地址，而不是直接通过套接字来获取。数据是通过一个方法调用来传输。这个方法只需要将接收到的字符串作为参数。
第二种风格是，协程风格的asyncio：
asyncio框架提供的另外一种构造协议代码的方法就是使用协程。协程是一个函数，它在进行I/O操作时不会阻塞，而是会暂停，并将控制权转移回调用方。Python语言的支持协程的一种标准形式就是生成器——在内部包含一个或多个yield语句的函数。这类函数不会再运行了一条返回语句之后就退出，而是会返回一个序列。
下面是通过协程实现的Zen协议。
import asyncio, zen_utils
@asyncio.coroutine
def handle_conversation(reader, writer):
address = writer.get_extra_info('peername')
print('Accepted connection from {}'.format(address))
while True:
data = b''
while not data.endswith(b'?'):
more_data = yield from reader.read(4096)
if not more_data:
if data:
print('Client {} sent {!r} but then closed'
.format(address, data))
else:
print('Client {} closed socket normally'.format(address))
return
data += more_data
answer = zen_utils.get_answer(data)
writer.write(answer)
if __name__ == '__main__':
address = zen_utils.parse_command_line('asyncio server using coroutine')
loop = asyncio.get_event_loop()
coro = asyncio.start_server(handle_conversation, *address)
server = loop.run_until_complete(coro)
print('Listening at {}'.format(address))
try:
loop.run_forever()
finally:
server.close()
loop.close()
上面介绍的异步服务器都可以在服务的不同客户端会话间切换。要完成切换，只需要扫描协议对象即可。
然而，异步服务器是有硬性限制的。因为所有的操作都在单个操作系统线程中完成。所以一旦CPU使用率达到么100%，异步服务器就无法再为任何客户端提供服务。这个时候即使有多个核心，所有工作也只能在单个处理器上完成。
然后这里有一个两全其美的方法。当我们需要高性能的时候，我们首先使用异步对象或协程来编写服务，并通过异步框架来启动服务。然后再回过头来配置一些运行服务器的操作系统，检查操作系统CPU内核数目。有多少个CPU内核，就启动多少个事件循环。
下面说一下inted守护进程。他解决了下面的问题：在一台特定的服务器上，在系统启动时启动n个不同的后台进程，用于提供n个不同的网络服务。可以简单地在系统的/etc/inted.conf文件中将所有要监听的端口全部列出。
inted守护进程在列出的每个端口都调用了bind()和listen()，不过它只在客户端真正连接时才会启动一个服务器进程。
为每一个连接都建立一个进程的花销是很大的，而且会降低服务器的利用率，不过这种方法也更加简单，要通过这种方式启动服务，只要在该服务的inetd.conf配置文件中将第4个字段设置为nowait即可。
如果是下面的命令：
>>>1060 stream tcp nowait brandon /user/bin/python3 /user/bin/python3 in_zen1.py
这样的服务一经启用，其标准输入输出流、标准输出流以及标准错误流便被连接到客户端套接字。服务只需要与连接的客户端通信，然后退出即可。
下面是和上面的inetd.conf配置结合使用的例子：
import socket, sys, zen_utils
if __name__ == '__main__':
sock = socket.fromfd(0, socket.AF_INET, socket.SOCK_STREAM)
sys.stdin = open('/dev/null', 'r')
sys.stdout = sys.stderr = open('/tmp/zen.log', 'a', buffering=1)
address = sock.getpeername()
print('Accepted connection from {}'.format(address))
zen_utils.handle_conversation(sock, address)
另一种模式是，在配置inetd.conf时将第四个字段指定为wait，表示会监听套接字提供给脚本。脚本需要调用accept()，勇于接受正在等待的客户端的连接请求。
这一个模式优势在于，服务器可以保持运行状态。并不断运行accept()来接受更多的客户端的连接请求，而在这一过程中并需要inetd的介入。如果客户端暂时停止了连接，服务器也可以自由调用exit()，来降低服务器的内存占用。在客户端再次需要服务器的时候再启动服务器即可。inetd会检测到我们的服务已经退出，然后会由inetd来负责监听。
下面是wait模式设计的代码。
import socket, sys, zen_utils
if __name__ == '__main__':
listener = socket.fromfd(0, socket.AF_INET, socket.SOCK_STREAM)
sys.stdin = open('/dev/null', 'r')
sys.stdout = sys.stderr = open('/tmp/zen.log', 'a', buffering=1)
listener.settimeout(8.0)
try:
zen_utils.accept_connections_forever(listener)
except socket.timeout:
print('Waited 8 seconds with no further connections; shutting down')
有的inetd版本还内置了一种基于IP地址与主机名的简单访问控制机制。有兴趣的可以查看相关的资料。
这篇文章到了这里就结束了，将编写的服务器安装在服务器上，并且在系统启动时运行服务器的过程叫做部署。
以后的文章会讲一些Python程序员依赖的基础网络服务，HTTP协议的设计还会讲到一些Python工具。还会说到Tornado的异步框架。
不过，那都是后话了。
新的一年天天开心。所有的一切万事胜意。
