个人中心
文章发布
退出
作者热门文章
- ubuntu12.04环境下使用kvm ioctl接口实现最简单的虚拟机
- Ubuntu 通过无线网络安装Ubuntu Server启动系统后连接无线网络的方法
- 在Ubuntu上搭建网桥的方法
- ubuntu 虚拟机上网方式及相关配置详解
31
4
CFSDN坚持开源创造价值，我们致力于搭建一个资源共享平台，让每一个IT人在这里找到属于你的精彩世界.
这篇CFSDN的博客文章基于并发服务器几种实现方法(总结)由作者收集整理，如果你对这篇文章有兴趣，记得点赞哟.
今天主题是实现并发服务器，实现方法有多种版本，先从简单的单进程代码实现到多进程，多线程的实现，最终引入一些高级模块来实现并发TCP服务器.
说到TCP，想起吐槽大会有个段子提到三次握手，也只有程序猿(媛)能get.
UDP服务器数据传输不可靠，这里就忽略了.
>>
简单的单进程TCP服务器 。
假代码:
#创建tcp服务器套接字 。
#绑定端口 。
#设置正常情况退出的服务器下,端口可以重用 。
#设置监听,变为主动监听 。
# 等待客户端的链接,返回新的socket和地址 。
#关闭tcp服务器套接字 。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
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服务器 。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
|
from socket import socket, AF_INET,SOCK_STREAM,SOL_SOCKET,SO_REUSEADDR
from 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服务器 。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
|
from socket import socket, AF_INET,SOCK_STREAM,SOL_SOCKET,SO_REUSEADDR
from 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服务器 。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
|
from socket import socket, AF_INET,SOCK_STREAM,SOL_SOCKET,SO_REUSEADDR
from 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)))
break
def 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 。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
#coding=utf-8
from greenlet import greenlet
import time
def 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()
|
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
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()交出控制权。
|
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
#coding=utf-8
import gevent
def 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服务器 —— 。
非堵塞式 。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
|
from socket import AF_INET,socket,SO_REUSEADDR,SOCK_STREAM,SOL_SOCKET
def 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多路复用：没有使用多进程和多线程的情况下完成多个套接字的使用.
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
|
from socket import AF_INET,socket,SO_REUSEADDR,SOCK_STREAM,SOL_SOCKET
from select import select
import sys
def 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版 。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
|
from socket import *
import select
def 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. 。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
import sys
import time
import gevent
from gevent import socket,monkey
monkey.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:
1）poll与select本质上没有区别，但poll没有最大连接数的限制； 。
2）大量的fd数组被整体复制于用户态和内核地址空间之间，不管这样的复制是不是有意义； 。
3）‘水平触发'，如果报告了fd后，没有被处理，下次poll时还会再次报告该fd.
epoll:
1）是之前poll和select的增强版，epoll更灵活，没有描述符限制，能打开的fd远大于1024（1G的内存上能监听约10万个端口）； 。
2）‘边缘出发'，事件通知机制，效率提升，最大的特点在于它只管你活跃的连接，而跟连接总数无关。而epoll对文件描述符的操作模式之一ET是一种高效的工作方式，很大程度减少事件反复触发的次数，内核不会发送更多的通知（only once）.
以上这篇基于并发服务器几种实现方法(总结)就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持我.
原文链接：https://www.cnblogs.com/NolaLi/archive/2017/12/28/8137973.html 。
最后此篇关于基于并发服务器几种实现方法(总结)的文章就讲到这里了,如果你想了解更多关于基于并发服务器几种实现方法(总结)的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。
31
4
0
背景: 我最近一直在使用 JPA，我为相当大的关系数据库项目生成持久层的轻松程度给我留下了深刻的印象。 我们公司使用大量非 SQL 数据库，特别是面向列的数据库。我对可能对这些数据库使用 JPA 有一
我已经在我的 maven pom 中添加了这些构建配置，因为我希望将 Apache Solr 依赖项与 Jar 捆绑在一起。否则我得到了 SolarServerException: ClassNotF
interface ITurtle { void Fight(); void EatPizza(); } interface ILeonardo : ITurtle {
我希望可用于 Java 的对象/关系映射 (ORM) 工具之一能够满足这些要求: 使用 JPA 或 native SQL 查询获取大量行并将其作为实体对象返回。 允许在行(实体)中进行迭代，并在对当前
好像没有，因为我有实现From for 的代码, 我可以转换 A到 B与 .into() , 但同样的事情不适用于 Vec .into()一个Vec . 要么我搞砸了阻止实现派生的事情，要么这不应该发
在 C# 中，如果 A 实现 IX 并且 B 继承自 A ，是否必然遵循 B 实现 IX？如果是，是因为 LSP 吗？之间有什么区别吗: 1. Interface IX; Class A : IX;
就目前而言，这个问题不适合我们的问答形式。我们希望答案得到事实、引用资料或专业知识的支持，但这个问题可能会引发辩论、争论、投票或扩展讨论。如果您觉得这个问题可以改进并可能重新打开，visit the
我正在阅读标准haskell库的(^)的实现代码: (^) :: (Num a, Integral b) => a -> b -> a x0 ^ y0 | y0 a -> b ->a expo x0
我将把国际象棋游戏表示为 C++ 结构。我认为，最好的选择是树结构(因为在每个深度我们都有几个可能的移动)。 这是一个好的方法吗？ struct TreeElement{ SomeMoveType
我正在为用户名数据库实现字符串匹配算法。我的方法采用现有的用户名数据库和用户想要的新用户名，然后检查用户名是否已被占用。如果采用该方法，则该方法应该返回带有数据库中未采用的数字的用户名。 例子: “贾
我正在尝试实现 Breadth-first search algorithm , 为了找到两个顶点之间的最短距离。我开发了一个 Queue 对象来保存和检索对象，并且我有一个二维数组来保存两个给定顶点
我目前正在 ika 中开发我的 Python 游戏，它使用 python 2.5 我决定为 AI 使用 A* 寻路。然而，我发现它对我的需要来说太慢了(3-4 个敌人可能会落后于游戏，但我想供应 4-
我正在寻找 Kademlia 的开源实现C/C++ 中的分布式哈希表。它必须是轻量级和跨平台的(win/linux/mac)。 它必须能够将信息发布到 DHT 并检索它。 最佳答案 OpenDHT是
我在一本书中读到这一行:-“当我们要求 C++ 实现运行程序时，它会通过调用此函数来实现。” 而且我想知道“C++ 实现”是什么意思或具体是什么。帮忙!？ 最佳答案 “C++ 实现”是指编译器加上链接
我正在尝试使用分支定界的 C++ 实现这个背包问题。此网站上有一个 Java 版本:Implementing branch and bound for knapsack 我试图让我的 C++ 版本打印
在很多情况下，我需要在 C# 中访问合适的哈希算法，从重写 GetHashCode 到对数据执行快速比较/查找。 我发现 FNV 哈希是一种非常简单/好/快速的哈希算法。但是，我从未见过 C# 实现的
目录 LRU缓存替换策略 核心思想 不适用场景 算法基本实现 算法优化
1. 绪论 在前面文章中提到 空间直角坐标系相互转换 ，测绘坐标转换时，一般涉及到的情况是：两个直角坐标系的小角度转换。这个就是我们经常在测绘数据处理中，WGS-84坐标系、54北京坐标系
在软件开发过程中，有时候我们需要定时地检查数据库中的数据，并在发现新增数据时触发一个动作。为了实现这个需求，我们在 .Net 7 下进行一次简单的演示. PeriodicTimer .
二分查找 二分查找算法，说白了就是在有序的数组里面给予一个存在数组里面的值key，然后将其先和数组中间的比较，如果key大于中间值，进行下一次mid后面的比较，直到找到相等的，就可以得到它的位置。
我是一名优秀的程序员,十分优秀！