[关闭]
@xuchongfeng 2017-12-03T12:49:09.000000Z 字数 6754 阅读 71

服务器服务模型总结

Linux Socket IO

基本概念

网络服务是资源的提供，服务器的存储资源，计算资源等，高效的服务模型是为了更快的服务更多的服务客户请求；


为了能更快的提供服务，需要减少等待时间，对于串行提供服务的，等待时间可能很长；而对于IO密集型的服务请求，其瓶颈在网络IO，对于计算密集型的服务请求，瓶颈在CPU，如果此时由大量的进程或者线程提供服务，那么就会发生大量的上下文切换，浪费CPU，这种情况需要减少上下文的切换，也就要求在一个进程中需要管理多个服务请求；


计算机操作系统层面，分为用户态和内核态，用户可以通过系统API，软中断和硬件中断进行内核态。传输层的TCP/UDP协议是内核协议栈，意味着接收到的数据先缓存在内核缓冲区，用户程序再将数据从内核缓冲区拷贝到用户存储空间；


如图所示，根据处理数据方式的不同，引入同步/异步，阻塞/非阻塞的概念。同步是指操作不完成就不返回，该调用会阻塞；异步不会阻塞调用者，当IO条件满足时，系统通知调用者操作已经完成。阻塞IO有点不达目的，誓不罢休的意思，读操作，一定要等到数据到达才返回，写操作，一定要写完数据才返回，调用者一直阻塞，直到IO条件满足；非阻塞IO，当条件满足时，则执行操作，当条件不满足时，如读操作，但是没有数据到达，写操作，内核TCP缓冲区已满，则立即返回，再由用户进程轮训状态。

此处输入图片的描述


IO复用允许一个进程同时管理多个请求。如poll，select，epoll，kqueue函数，监视对应请求发生的事件：可读，可写，异常等。在存在大量短连接的业务场景下，能高效的提供服务。

之后的部分是使用Python编写的示例，主要是因为Python比较简单，容易理解模型。

单进程(线程)+阻塞IO

服务器端

  1. #!/usr/bin/python
  2. # -*- coding: utf8 -*-
  4. import socket
  5. import datetime
  7. address = ('localhost', 1234)
  9. def init():
  10.     listenfd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
  11.     listenfd.bind(address)
  12.     listenfd.listen(1)
  13.     return listenfd
  15. def logic(listenfd):
  16.     print "start accept connection"
  17.     while 1:
  18.         conn, addr = listenfd.accept()
  19.         while 1:
  20.             data = conn.recv(1024)
  21.             if not data:
  22.                 break
  23.             print "recv data len is %d" % len(data)
  24.             conn.sendall(data)
  26. def loop():
  27.     listenfd = init()
  28.     while 1:
  29.         try:
  30.             logic(listenfd)
  31.         except socket.error, e:
  32.             print e
  34. if __name__ == "__main__":
  35.     print "listening: %s : %s" % (address)
  36.     loop()

客户端

  1. #!/usr/bin/python
  2. # -*- coding: utf8 -*-
  4. import socket
  5. import time
  6. import sys
  9. address = ("localhost", 1234)
  11. #echo, send data to server and receive data from server
  12. def run(chunk_size):
  13.     connfd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
  14.     connfd.connect(address)
  15.     content = open("/dev/zero", "r")
  16.     chunk = content.read(chunk_size)
  17.     left = chunk_size
  18.     connfd.sendall(chunk)
  19.     left = chunk_size
  20.     while left > 0:
  21.         data = connfd.recv(1024)
  22.         left -= len(data)
  23.         print "recv data len is %d" % len(data)
  24.     #time.sleep(10)
  26. if __name__ == "__main__":
  27.     if len(sys.argv) != 2:
  28.         print "usage: python client.py <chunk_size>"
  29.     else:
  30.         run(int(sys.argv[1]))

多线程+阻塞IO

  1. #!/usr/bin/python
  2. # -*- coding: utf8 -*-
  4. # sever
  5. import socket
  6. import datetime
  7. import threading
  9. address = ('localhost', 1234)
  11. def init():
  12.     listenfd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
  13.     listenfd.bind(address)
  14.     listenfd.listen(1)
  15.     return listenfd
  17. def worker(connfd):
  18.     #print "%s" % threading.current_thread().name
  19.     data = ""
  20.     while 1:
  21.         tmp_data = connfd.recv(1024)
  22.         if not tmp_data:
  23.             break
  24.         data += tmp_data
  25.     connfd.sendall(data)
  27. def logic(listenfd):
  28.     print "start accept connection"
  29.     thread_num = 0
  30.     while 1:
  31.         conn, addr = listenfd.accept()
  32.         worker_thread = threading.Thread(target=worker, args=(conn, ), \
  33.                 name="thread-%d" % thread_num)
  34.         worker_thread.start()
  35.         worker_thread.join()
  36.         thread_num += 1
  38. def loop():
  39.     listenfd = init()
  40.     while 1:
  41.         try:
  42.             logic(listenfd)
  43.         except socket.error, e:
  44.             print e
  46. if __name__ == "__main__":
  47.     print "listening: %s : %s" % (address)
  48.     loop()
  1. #!/usr/bin/python
  2. # -*- coding: utf8 -*-
  4. import socket
  5. import time
  6. import sys
  7. import threading
  8. import datetime
  10. address = ("localhost", 1234)
  12. #echo, send data to server and receive data from server
  13. def run(chunk_size):
  14.     connfd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
  15.     print connfd
  16.     connfd.connect(address)
  17.     content = open("/dev/zero", "r")
  18.     chunk = content.read(chunk_size)
  19.     left = chunk_size
  20.     connfd.sendall(chunk)
  21.     connfd.shutdown(socket.SHUT_WR)
  22.     left = chunk_size
  23.     while left > 0:
  24.         data = connfd.recv(1024)
  25.         if len(data) == 0:
  26.             break
  27.         left -= len(data)
  28.         print "%s: recv data len is %d" % (threading.current_thread().name, len(data))
  29.     #time.sleep(10)
  31. if __name__ == "__main__":
  32.     if len(sys.argv) != 3:
  33.         print "usage: python client.py <thread_num> <chunk_size>"
  34.     else:
  35.         time_start = datetime.datetime.now()
  36.         for i in range(int(sys.argv[1])):
  37.             job = threading.Thread(target=run, \
  38.                     args=(int(sys.argv[2]),), \
  39.                     name=("thread-%d" % i)
  40.             )
  41.             job.start()
  42.             job.join()
  43.         time_end = datetime.datetime.now()
  44.         time_delta = time_end - time_start
  45.         print "start at: %s\nend at: %s\neclipse time: %s" % (time_start,
  46.                 time_end, time_delta)

IO复用+非阻塞IO

  1. #!/usr/bin/python
  2. # -*- coding: utf8 -*-
  4. # server, using poll
  5. import time
  6. import socket
  7. import select
  8. import Queue
  9. import errno
  12. ADDRESS = ("localhost", 1234)
  14. def init():
  15.     listenfd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
  16.     listenfd.bind(ADDRESS)
  17.     listenfd.listen(5)
  18.     return listenfd
  20. def eventloop():
  21.     listenfd = init()
  22.     readset = [listenfd]
  23.     writeset = []
  24.     exceptset = []
  26.     message_queue = {}
  28.     while True:
  29.         rset, wset, exset = select.select(readset, writeset, exceptset, 20)
  31.         for fd in rset:
  32.             # fd == listenfd, accept
  33.             if fd == listenfd:
  34.                 connfd, addr = listenfd.accept()
  35.                 print addr
  36.                 connfd.setblocking(False)
  37.                 readset.append(connfd)
  38.                 message_queue[connfd] = Queue.Queue()
  39.             # recv data
  40.             else:
  41.                 data = ""
  42.                 try:
  43.                     while True:
  44.                         tmp_data = fd.recv(1024)
  45.                         if not tmp_data:
  46.                             break
  47.                         data += tmp_data
  48.                     if len(data) == 0:
  49.                         readset.remove(fd)
  50.                         continue
  51.                 except socket.error, e:
  52.                     err = e.args[0]
  53.                     # signal interrupt or no data receive
  54.                     if err == errno.EAGAIN or err == errno.EWOULDBLOCK:
  55.                         writeset.append(fd)
  56.                     # the peer disconnected
  57.                     else:
  58.                         if fd in readset:
  59.                             readset.remove(fd)
  60.                         print "sendall"
  61.                 message_queue[fd].put(data)
  63.         for fd in wset:
  64.             try:
  65.                 data = message_queue[fd].get_nowait()
  66.             except Queue.Empty:
  67.                 # configure fd closed or temp no data
  68.                 if fd not in readset:
  69.                     writeset.remove(fd)
  70.             else:
  71.                 print len(data)
  72.                 fd.send(data)
  75.         for fd in exset:
  76.             readset.remove(fd)
  77.             if fd in writeset:
  78.                 writeset.remove(fd)
  79.             fd.close()
  80.             del message_queue[fd]
  82. if __name__ == "__main__":
  83.     print "listening %s:%s" % ADDRESS
  84.     eventloop()
  1. #!/usr/bin/python
  2. # -*- coding: utf8 -*-
  4. # server, using epoll
  5. import time
  6. import socket
  7. import select
  8. import Queue
  9. import errno
  12. ADDRESS = ("localhost", 1234)
  14. def init():
  15.     listenfd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
  16.     listenfd.bind(ADDRESS)
  17.     listenfd.listen(5)
  18.     return listenfd
  20. def eventloop():
  21.     listenfd = init()
  22.     listenfd.setblocking(False)
  24.     epoll = select.epoll()
  25.     epoll.register(listenfd.fileno(), select.EPOLLIN)
  27.     message_queue = {}
  28.     connections = {}
  30.     while True:
  31.         events = epoll.poll(1)
  32.         for fd, event in events:
  33.             if fd == listenfd.fileno():
  34.                 connfd, address = listenfd.accept()
  35.                 connfd.setblocking(False)
  36.                 epoll.register(connfd.fileno(), select.EPOLLIN)
  37.                 connections[connfd.fileno()] = connfd
  38.                 message_queue[connfd.fileno()] = Queue.Queue()
  39.             elif event & select.EPOLLIN:
  40.                 data = ""
  41.                 try:
  42.                     while True:
  43.                         tmp_data = connections[fd].recv(1024)
  44.                         #print len(tmp_data)
  45.                         if len(tmp_data) == 0:
  46.                             break;
  47.                         data += tmp_data
  48.                     if len(data) == 0:
  49.                         epoll.modify(fd, select.EPOLLOUT | select.EPOLLET)
  50.                 except socket.error, e:
  51.                     err = e.args[0]
  52.                     if err == errno.EAGAIN or err == errno.EWOULDBLOCK:
  53.                         epoll.modify(fd, select.EPOLLIN | select.EPOLLOUT | select.EPOLLET)
  54.                     message_queue[fd].put(data)
  55.             elif event & select.EPOLLOUT:
  56.                 try:
  57.                     data = message_queue[fd].get_nowait()
  58.                 except Queue.Empty:
  59.                     pass
  60.                 else:
  61.                     connections[fd].sendall(data)
  62.             elif event & select.EPOLLHUP:
  63.                 epoll.unregister(fd)
  64.                 connections[fd].close()
  65.                 del connections[fd]
  66.     epoll.unregister(listenfd.fileno())
  67.     epoll.close()
  68.     listenfd.close()
  70. if __name__ == "__main__":
  71.     print "listening %s:%s" % ADDRESS
  72.     eventloop()

总结

使用线程池，使用非阻塞IO+IO复用+线程池的组合可以高效的提供服务，最近的Nginx1.7版本，使用线程池处理耗时的操作，对于IO循环，操作在一个进程中完成，当某个操作比较耗时时，就会影响整个服务的提供，使用线程池处理耗时的请求能解决这一问题；


协程，协程是比线程更小的单元，它不是系统级的调度单元，它的“作案现场”在用户态保存，减少了用户态和内核态的切换，同时使程序员可以用同步的思维写异步执行的程序。

协程示例，使用同步思维写消费者和生产者。

  1. #!/usr/bin/python
  2. # -*- coding: utf8 -*-
  4. def producer(c):
  5.     c.send(None)
  6.     n = 0
  7.     while n < 5:
  8.         n = n + 1
  9.         print "[producer] producting %s..." % n
  10.         r = c.send(n)
  11.         print "[producer] consumer return: %s" % r
  13. def consumer():
  14.     r = ""
  15.     while True:
  16.         n = yield r
  17.         if not n:
  18.             return
  19.         print "[consumer] consuming %s..." % n
  20.         r = 'consumed'
  22. if __name__ == "__main__":
  23.     c = consumer()
  24.     producer(c)
添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注