基于并发服务器几种实现方法(总结)

2019-11-25 15:27:14

字体：大中小

来源：转载

供稿：网友

今天主题是实现并发服务器，实现方法有多种版本，先从简单的单进程代码实现到多进程，多线程的实现，最终引入一些高级模块来实现并发TCP服务器。

说到TCP，想起吐槽大会有个段子提到三次握手，也只有程序猿(媛)能get。

UDP服务器数据传输不可靠，这里就忽略了。

>>:

简单的单进程TCP服务器

假代码：

#创建tcp服务器套接字

#绑定端口

#设置正常情况退出的服务器下,端口可以重用

#设置监听,变为主动监听

# 等待客户端的链接,返回新的socket和地址

#关闭tcp服务器套接字

from socket import socket, AF_INET,SOCK_STREAM,SOL_SOCKET,SO_REUSEADDR#创建tcp服务器套接字server_socket = socket(AF_INET,SOCK_STREAM)#绑定端口server_socket.bind(("",9999))#设置正常情况退出的服务器下,端口可以重用server_socket.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)#设置监听,变为主动监听server_socket.listen(5)while True: # 等待客户端的链接,返回新的socket和地址 new_socket,new_address = server_socket.accept() #接收数据,并且发送数据 try: while True: recv_data = new_socket.recv(1024) #当有客户端关闭后,recv解除阻塞,并且返回长度为0 if len(recv_data) > 0: recv_content = recv_data.decode("gb2312") print("收到:%s的信息是:%s" % (str(new_address),recv_content)) new_socket.send("thank you!".encode("gb2312")) else: print("客户端%s已经关闭" % (str(new_address))) break finally: new_socket.close() print("关闭%s客户端" % (str(new_address)))#关闭tcp服务器套接字server_socket.close()

多进程TCP服务器

from socket import socket, AF_INET,SOCK_STREAM,SOL_SOCKET,SO_REUSEADDRfrom multiprocessing import Process#在子进程中接收消息def recv_data(new_socket,new_address): while True:  recv_data = new_socket.recv(1024)  # 当有客户端关闭后,recv解除阻塞,并且返回长度为0  if len(recv_data) > 0:   recv_content = recv_data.decode("gb2312")   print("收到:%s的信息是:%s" % (str(new_address), recv_content))   new_socket.send("thank you!".encode("gb2312"))  else:   print("客户端%s已经关闭" % (str(new_address)))   break #关闭与客户端的连接 print("关闭与客户端的连接") new_socket.close()def main(): #创建tcp服务器套接字 server_socket = socket(AF_INET,SOCK_STREAM) #绑定端口 server_socket.bind(("",8888)) #设置正常情况退出的服务器下,端口可以重用 server_socket.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) #设置监听,变为被动连接 server_socket.listen(3) try:  while True:   # 等待客户端的链接,返回新的socket和地址   new_socket,new_address = server_socket.accept()   #接收数据,并且发送数据   Process(target=recv_data,args=(new_socket,new_address)).start()   #因为主进程和子进程不共享数据   #如果我们直接关闭new_socket,只是关闭主进程的new_socket,而子进程的不受影响   new_socket.close() finally:  #关闭tcp服务器套接字  server_socket.close()if __name__ == "__main__": main()

多进程TCP服务器

from socket import socket, AF_INET,SOCK_STREAM,SOL_SOCKET,SO_REUSEADDRfrom multiprocessing import Process#在子进程中接收消息def recv_data(new_socket,new_address): while True:  recv_data = new_socket.recv(1024)  # 当有客户端关闭后,recv解除阻塞,并且返回长度为0  if len(recv_data) > 0:   recv_content = recv_data.decode("gb2312")   print("收到:%s的信息是:%s" % (str(new_address), recv_content))   new_socket.send("thank you!".encode("gb2312"))  else:   print("客户端%s已经关闭" % (str(new_address)))   break #关闭与客户端的连接 print("关闭与客户端的连接") new_socket.close()def main(): #创建tcp服务器套接字 server_socket = socket(AF_INET,SOCK_STREAM) #绑定端口 server_socket.bind(("",8888)) #设置正常情况退出的服务器下,端口可以重用 server_socket.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) #设置监听,变为被动连接 server_socket.listen(3) try:  while True:   # 等待客户端的链接,返回新的socket和地址   new_socket,new_address = server_socket.accept()   #接收数据,并且发送数据   Process(target=recv_data,args=(new_socket,new_address)).start()   #因为主进程和子进程不共享数据   #如果我们直接关闭new_socket,只是关闭主进程的new_socket,而子进程的不受影响   new_socket.close() finally:  #关闭tcp服务器套接字  server_socket.close()if __name__ == "__main__": main()

多线程TCP服务器

from socket import socket, AF_INET,SOCK_STREAM,SOL_SOCKET,SO_REUSEADDRfrom threading import Thread#接收消息def recv_data(new_socket,new_address): while True:  recv_data = new_socket.recv(1024)  # 当有客户端关闭后,recv解除阻塞,并且返回长度为0  if len(recv_data) > 0:   recv_content = recv_data.decode("gb2312")   print("收到:%s的信息是:%s" % (str(new_address), recv_content))   new_socket.send("thank you!".encode("gb2312"))  else:   print("客户端%s已经关闭" % (str(new_address)))   breakdef main(): #创建tcp服务器套接字 server_socket = socket(AF_INET,SOCK_STREAM) #绑定端口 server_socket.bind(("",9999)) #设置正常情况退出的服务器下,端口可以重用 server_socket.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) #设置监听,变为被动连接 server_socket.listen(3) try:  while True:   # 等待客户端的链接,返回新的socket和地址   new_socket,new_address = server_socket.accept()   #接收数据,并且发送数据   Thread(target=recv_data,args=(new_socket,new_address)).start() finally:  #关闭tcp服务器套接字  server_socket.close()if __name__ == "__main__": main()

多任务协程实现 ――

greenlet和gevent

#coding=utf-8from greenlet import greenletimport timedef test1(): while True:  print "---A--"  gr2.switch()  time.sleep(0.5)def test2(): while True:  print "---B--"  gr1.switch()  time.sleep(0.5)gr1 = greenlet(test1)gr2 = greenlet(test2)#切换到gr1中运行gr1.switch()

import gevent#函数def f(n): for i in range(n):  print("%s:%s" % (gevent.getcurrent(),i))f1 = gevent.spawn(f,5)f2 = gevent.spawn(f,5)f3 = gevent.spawn(f,5)#让主线程等待三个协程执行完毕,否则没有机会执行f1.join()f2.join()f3.join()#可以看到，3个greenlet是依次运行而不是交替运行。要让greenlet交替运行，可以通过gevent.sleep()交出控制权。

#coding=utf-8import geventdef f(n): for i in range(n):  print gevent.getcurrent(), i  #用来模拟一个耗时操作，注意不是time模块中的sleep  gevent.sleep(1)g1 = gevent.spawn(f, 5)g2 = gevent.spawn(f, 5)g3 = gevent.spawn(f, 5)#下面三行代码意思：主线程等待各个协成支持完，否则协成没有机会执行g1.join()g2.join()g3.join()

单进程TCP服务器 ――

非堵塞式

from socket import AF_INET,socket,SO_REUSEADDR,SOCK_STREAM,SOL_SOCKETdef main(): #创建tcp的socket套接字 server_socket = socket(AF_INET,SOCK_STREAM) server_socket.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) #绑定端口 server_socket.bind(("",9999)) #设置非阻塞,也就是说accept方法不阻塞了, # 但是在没有客户端链接且被执行的时候会报错 #有客户端链接的时候正常执行 server_socket.setblocking(False) #设置监听 server_socket.listen(5) #客户端列表 client_lists = [] try:  #不断调用accept  while True:   try:   # print("accept--111")   new_socket,new_address = server_socket.accept()   print("accept--2222")   except Exception as result:   # print(result)   pass   else:   print("新的客户%s链接上" % str(new_address))   #新链接的new_sokect默认也是阻塞,也设置为非阻塞后,recv为非阻塞   new_socket.setblocking(False)   client_lists.append((new_socket,new_address))   # print(111)   for client_sokect,client_address in client_lists:   #接收数据   try:    recv_data = client_sokect.recv(1024)   except Exception as result:    # print(result)    pass   else:    # print("正常数据:%s" %recv_data)    if len(recv_data) > 0 :     print("收到%s:%s" % (str(client_address),recv_data))     client_sokect.send("thank you!".encode("gb2312"))    else:     #客户端已经端口,要把该客户端从列表中异常     client_lists.remove((client_sokect,new_address))     client_sokect.close()     print("%s已经断开" % str(new_address)) finally:  #关闭套接字  server_socket.close()if __name__ == "__main__": main()

单进程TCP服务器 ――

select版

select 原理

其他语言（c或者c++）也有使用select实现多任务服务器。

select 能够完成一些套接字的检查，从头到尾检查一遍后，标记哪些套接字是否可以收数据，返回的时候，就返回能接收数据的套接字，返回的是列表。select是由操作系统提供的，效率要高些,非常快的方式检测哪些套接字可以接收数据。select是跨平台的，在window也可以用。

io多路复用：没有使用多进程和多线程的情况下完成多个套接字的使用。

from socket import AF_INET,socket,SO_REUSEADDR,SOCK_STREAM,SOL_SOCKETfrom select import selectimport sysdef main(): #创建tcp的socket套接字 server_socket = socket(AF_INET,SOCK_STREAM) server_socket.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) #绑定端口 server_socket.bind(("",9999)) #设置监听 server_socket.listen(5) #客户端列表 socket_lists = [server_socket,sys.stdin] wirte_list = [] #是否退出 is_run = False try:  while True:   #检测列表client_lists那些socket可以接收数据,   #检测列表[]那些套接字(socket)可否发送数据   #检测列表[]那些套接字(socket)是否产生了异常   print("select--111")   #这个select函数默认是堵塞,当有客户端链接的时候解除阻塞,   # 当有数据可以接收的时候解除阻塞,当客户端断开的时候解除阻塞   readable, wirteable,excep = select(socket_lists,wirte_list,[])   # print("select--2222")   # print(111)   for sock in wirteable:   #这个会一直发送,因为他是处于已经发的状态   sock.send("thank you!".encode("gb2312"))   for sock in readable:   #接收数据   if sock == server_socket:    print("sock == server_socket")    #有新的客户端链接进来    new_socket,new_address = sock.accept()    #新的socket添加到列表中,便于下次socket的时候能检查到    socket_lists.append(new_socket)   elif sock == sys.stdin:    cmd = sys.stdin.readline()    print(cmd)    is_run = cmd   else:    # print("sock.recv(1024)....")    #此时的套接字sock是直接可以取数据的    recv_data = sock.recv(1024)    if len(recv_data) > 0:     print("从[%s]:%s" % (str(new_address),recv_data))     sock.send(recv_data)     #把链接上有消息接收的socket添加到监听写的列表中     wirte_list.append(sock)    else:     print("客户端已经断开")     #客户端已经断开,要移除     sock.close()     socket_lists.remove(sock)   #是否退出程序   if is_run:   break finally:  #关闭套接字  server_socket.close()if __name__ == "__main__": main()

单进程TCP服务器 ――

epoll版

from socket import *import selectdef main(): #创建tcp服务器套接字 server_socket = socket(AF_INET,SOCK_STREAM) #设置端口可以重用 server_socket.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) #绑定端口 server_socket.bind(("",9999)) #设置监听 server_socket.listen(5) #用epoll设置监听收数据 epoll = select.epoll() #把server_socket注册到epoll的事件监听中,如果已经注册过会发生异常 epoll.register(server_socket.fileno(),select.EPOLLIN|select.EPOLLET) #装socket列表 socket_lists = {} #装socket对应的地址 socket_address = {} while True:  #返回套接字列表[(socket的文件描述符,select.EPOLLIN)],  # 如果有新的链接,有数据发过来,断开链接等都会解除阻塞  print("epoll.poll--111")  epoll_list = epoll.poll()  print("epoll.poll--222")  print(epoll_list)  for fd,event in epoll_list:   #有新的链接   if fd == server_socket.fileno():   print("新的客户fd==%s" % fd)   new_sokect,new_address = server_socket.accept()   #往字典添加数据   socket_lists[new_sokect.fileno()] = new_sokect   socket_address[new_sokect.fileno()] = new_address   #注册新的socket也注册到epoll的事件监听中   epoll.register(new_sokect.fileno(), select.EPOLLIN | select.EPOLLET)   elif event ==select.EPOLLIN:   print("收到数据了")   #根据文件操作符取出对应socket   new_sokect = socket_lists[fd]   address = socket_address[fd]   recv_data = new_sokect.recv(1024)   if len(recv_data) > 0:    print("已经收到[%s]:%s" % (str(address),recv_data.decode("gb2312")))   else:    #客户端端口,取消监听    epoll.unregister(fd)    #关闭链接    new_sokect.close()    print("[%s]已经下线" % str(address)) #关闭套接字链接 server_socket.close()if __name__ == "__main__": main()

单进程TCP服务器 ――

gevent版

gevent原理

greenlet已经实现了协程，但是这个还得人工切换，是不是觉得太麻烦了，莫要捉急，python还有一个比greenlet更强大的并且能够自动切换任务的模块gevent

原理------当一个greenlet遇到IO(指的是input output 输入输出，比如网络、文件操作等)操作时，比如访问网络，就自动切换到其他的greenlet，等到IO操作完成，再在适当的时候切换回来继续执行。

由于IO操作非常耗时，经常使程序处于等待状态，有了gevent为我们自动切换协程，就保证总有greenlet在运行，而不是等待IO.

import sysimport timeimport geventfrom gevent import socket,monkeymonkey.patch_all()def handle_request(conn): while True:  data = conn.recv(1024)  if not data:   conn.close()   break  print("recv:", data)  conn.send(data)def server(port): s = socket.socket() s.bind(('', port)) s.listen(5) while True:  newSocket, addr = s.accept()  gevent.spawn(handle_request, newSocket)if __name__ == '__main__': server(7788)

首先基于以上代码模块，撒点概念问题：

1.什么是协程？

协程：存在线程中，是比线程更小的执行单元，又称微线程，纤程。自带cpu上下文，操作协程由程序员决定，它可以将一个线程分解为多个微线程，每个协程间共享全局空间的变量，每秒钟切换频率高达百万次。

2. 什么是计算密集型和IO密集型

计算密集型：要进行大量的计算，消耗cpu资源。如复杂计算，对视频进行高清解码等，全靠cpu的运算能力。而计算密集型任务完成多任务切换任务比较耗时，cpu执行任务效率就越低。在python中，多进程适合计算密集型任务。

IO密集型：涉及到网络、磁盘io的任务都是io密集型。cpu消耗少，计算量小，如请求网页，读写文件等。在python中，使用sleep达到IO密集型任务的目的，多线程适合IO密集型任务。

各大实现版本对比：

select：

1）支持跨平台，最大缺陷是单个进程打开的FD是有限的，由FD_SETSIZE设置，默认是1024；

2）对socket扫描时是线性扫描，及采用轮询方式，效率低；

3）需要维护一个存放大量FD的数据结构，使得用户空间和内核空间在传递该数据结构时复制开销大。

poll：