Python 之网络式编程
发布于2019-08-07 18:24 阅读(643) 评论(0) 点赞(3) 收藏(1)
一 客户端/服务器架构
即C/S架构，包括
1、硬件C/S架构（打印机）
2、软件B/S架构（web服务）
C/S架构与Socket的关系：
我们学习Socket就是为了完成C/S的开发
二 OSI七层
引子：
计算机组成原理：硬件、操作系统、应用软件三者组成。
具备以上条件后，计算机就可以工作，如果你要和别人一起玩，那你就需要上网了。互联网的核心就是由一堆协议组成，协议就是标准。
为什么学习Socket之前要先了解互联网协议？
1、C/S架构的软件（应用软件属于应用层）是基于网络进行通信的
2、网络的核心即一堆协议，协议即标准，想开发一款基于网络通信的软件，就必须遵循这些标准
OSI七层：
三 Socket层
四 Socket是什么
Socket是应用层与TCP/IP协议族通信的中间软件抽象层，它是一组接口，在设计模式中，Socket其实就是一个门面模式，它把负责的TCP/IP协议族隐藏在Socket接口后面，对用户来说，一组简单的接口就是全部，让Socket去组织数据，以符合指定的协议。
所以，我们无需深入学习理解TCP/UDP协议，Socket已经为我们封装好了，我们只需要遵循Socket的规定去编程，写出的程序自然就是遵循TCP/UDP标准的。
五 套接字发展史及分类
套接字起源于20世纪70年代加利福尼亚大学伯克利分校版本的Unix，即人们所说的BSD Unix。因此，有时人们也把套接字成为“伯克利套接字”或“BSD套接字”。一开始，套接字被设计用在一台主机上多个应用程序之间的通信，这也被称作进程间通许或IPC。套接字有两种（或者称为两个种族），分别是基于文件型和就网络型。
基于文件类型的套接字家族
套接字家族的名字：AF_UNIX
UNIX一切皆文件，基于文件的套接字调用的就是底层的文件系统来取数据，两个套接字进程运行在同一机器上，可以通过访问同一文件系统间接完成通信。
基于网络类型的套接字家族
套接字家族的名字：AF_INET
还有AF_INET6被用于ipv6，还有一些其他的地址家族，不过，他们要么是只用于某个平台，要么就是已经被废弃，或者是很少被使用，或者是根本没有实现，所有地址家族中，AF_INET是使用最广泛的一个，Python支持很多地址家族，但是由于我们只关心网络编程，所以大部分时候我们只使用AF_INET(AF:Address Family;INET:Internet)
六 套接字工作流程
生活中，你要打电话给一个朋友，先拨号，朋友听到电话铃声响后接打电话，这时你和你的朋友就建立起了连接，就可以讲话了，等交流结束，挂断电话结束此次通话。
利用Socket模拟生活中打电话：
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import socket 4 5 phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 买手机;socket.AF_INET:基于网络协议;socket.SOCK_STREAM:基于流的TCP协议 6 phone.bind(('127.0.0.1', 8080)) # 绑定手机卡;元祖形式，ip地址+端口 7 # 注：服务器的ip地址写本机的ip地址 8 phone.listen(5) # 开机 9 conn, addr = phone.accept() # 等电话 10 msg = conn.recv(1024) # 收消息 11 print('客户端发来的消息是:', msg) 12 conn.send(msg.upper()) # 发消息 13 conn.close() 14 phone.close()
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import socket 4 phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 买手机 5 phone.connect(('127.0.0.1', 8080)) # 拨电话 6 # 注：客户端的ip地址，写服务器端的ip地址 7 phone.send('hello'.encode('utf-8')) # 发消息 8 data = phone.recv(1024) # 收消息 9 print('收到服务端发来的消息', data)
服务器和客户端无限循环发送消息：
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 # import socket 4 from socket import * 5 import time 6 ip_port = ('127.0.0.1', 8080) 7 back_log = 5 8 buffer_size = 1024 9 tcp_server = socket(AF_INET, SOCK_STREAM) 10 tcp_server.bind(ip_port) 11 tcp_server.listen(back_log) 12 print('服务端开始运行') 13 conn, addr = tcp_server.accept() # 服务器阻塞 14 print('双向链接', conn) 15 print('客户端地址', addr) 16 while True: 17 time.sleep(1) 18 print('[%s]' % time.time()) 19 data = conn.recv(buffer_size) 20 print('客户端发来的消息是', data.decode('utf-8')) 21 conn.send(data.upper()) 22 conn.close() 23 tcp_server.close()
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from socket import * 4 import time 5 ip_port = ('127.0.0.1', 8080) 6 back_log = 5 7 buffer_size = 1024 8 tcp_client = socket(AF_INET, SOCK_STREAM) 9 tcp_client.connect(ip_port) 10 while True: 11 time.sleep(2) 12 print('[%s]' % time.time()) 13 msg = input('>>:').strip() 14 tcp_client.send(msg.encode('utf-8')) 15 print('客户端已经发送消息') 16 data = tcp_client.recv(buffer_size) 17 print('收到服务端发来消息', data.decode('utf-8')) 18 tcp_client.close()
Socket收发消息原理图：
若重启服务端时，可能会遇到：Address already in use;这个是由于服务端扔然存在四次挥手的time_wait状态占用地址
解决方案：
1 # 加入一条socket配置，重用ip和端口 2 tcp_server = socket(AF_INET, SOCK_STREAM) 3 tcp_server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) # <---就是这条，在bind前加 4 tcp_server.bind(ip_port)
1 发现系统存在大量TIME_WAIT状态的连接，通过调整linux内核参数解决， 2 vi /etc/sysctl.conf 3 4 编辑文件，加入以下内容： 5 net.ipv4.tcp_syncookies = 1 6 net.ipv4.tcp_tw_reuse = 1 7 net.ipv4.tcp_tw_recycle = 1 8 net.ipv4.tcp_fin_timeout = 30 9 10 然后执行 /sbin/sysctl -p 让参数生效。 11 12 net.ipv4.tcp_syncookies = 1 表示开启SYN Cookies。当出现SYN等待队列溢出时，启用cookies来处理，可防范少量SYN攻击，默认为0，表示关闭； 13 14 net.ipv4.tcp_tw_reuse = 1 表示开启重用。允许将TIME-WAIT sockets重新用于新的TCP连接，默认为0，表示关闭； 15 16 net.ipv4.tcp_tw_recycle = 1 表示开启TCP连接中TIME-WAIT sockets的快速回收，默认为0，表示关闭。 17 18 net.ipv4.tcp_fin_timeout 修改系統默认的 TIMEOUT 时间
七 基于UDP的套接字
udp服务端
1 ss = socket() # 创建一个服务器的套接字 2 ss.bind() # 绑定服务器套接字 3 while True: # 服务器无限循环 4 cs = ss.recvfrom()/ss.sendto() # 对话（接收与发送） 5 ss.close() # 关闭服务器套接字
udp客户端
1 cs = socket() # 创建客户套接字 2 while True: 3 cs.sendto()/cs.recvfrom() # 对话（发送/接收） 4 cs.close() # 关闭客户套接字
基于UDP的套接字：
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from socket import * 4 ip_port = ('127.0.0.1', 8080) 5 buffer_size = 1024 6 udp_server = socket(AF_INET, SOCK_DGRAM) # SOCK_DGRAM:数据报式套接字 7 udp_server.bind(ip_port) 8 while True: 9 # data = udp_server.recvfrom(buffer_size) 10 # print(data) # (b'hello', ('127.0.0.1', 65047)) 11 data, addr = udp_server.recvfrom(buffer_size) 12 print(data) 13 print(addr) 14 udp_server.sendto(data.upper(), addr)
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from socket import * 4 ip_port = ('127.0.0.1', 8080) 5 buffer_size = 1024 6 udp_client = socket(AF_INET, SOCK_DGRAM) # SOCK_DGRAM:数据报式套接字 7 while True: 8 msg = input('>>').strip() 9 udp_client.sendto(msg.encode('utf-8'), ip_port) 10 data, addr = udp_client.recvfrom(buffer_size) 11 print(data.decode('utf-8')) 12 print(addr)
八 什么是粘包？
注：只有TCP有粘包现象，UDP永远不会粘包
一个socket收发消息的原理图：
发送端可以是一K一K地发送数据，而接收端的应用程序可以两K两K地提走数据，当然也有可能一次提走3K或6K数据，或者一次只提走几个字节的数据，也就是说，应用程序所看到的数据是一个整体，或说是一个流（stream），一条消息有多少字节对应用程序是不可见的，因此TCP协议是面向流的协议，这也是容易出现粘包问题的原因。而UDP是面向消息的协议，每个UDP段都是一条消息，应用程序必须以消息为单位提取数据，不能一次提取任意字节的数据，这一点和TCP是很不同的。怎样定义消息呢？可以认为对方一次性write/send的数据为一个消息，需要明白的是当对方send一条信息的时候，无论底层怎样分段分片，TCP协议层会把构成整条消息的数据段排序完成后才呈现在内核缓冲区。
例如基于tcp的套接字客户端往服务端上传文件，发送时文件内容是按照一段一段的字节流发送的，在接收方看了，根本不知道该文件的字节流从何处开始，在何处结束
所谓粘包问题主要还是因为接收方不知道消息之间的界限，不知道一次性提取多少字节的数据所造成的。
此外，发送方引起的粘包是由TCP协议本身造成的，TCP为提高传输效率，发送方往往要收集到足够多的数据后才发送一个TCP段。若连续几次需要send的数据都很少，通常TCP会根据优化算法把这些数据合成一个TCP段后一次发送出去，这样接收方就收到了粘包数据。
- TCP（transport control protocol，传输控制协议）是面向连接的，面向流的，提供高可靠性服务。收发两端（客户端和服务器端）都要有一一成对的socket，因此，发送端为了将多个发往接收端的包，更有效的发到对方，使用了优化方法（Nagle算法），将多次间隔较小且数据量小的数据，合并成一个大的数据块，然后进行封包。这样，接收端，就难于分辨出来了，必须提供科学的拆包机制。 即面向流的通信是无消息保护边界的。
- UDP（user datagram protocol，用户数据报协议）是无连接的，面向消息的，提供高效率服务。不会使用块的合并优化算法，, 由于UDP支持的是一对多的模式，所以接收端的skbuff(套接字缓冲区）采用了链式结构来记录每一个到达的UDP包，在每个UDP包中就有了消息头（消息来源地址，端口等信息），这样，对于接收端来说，就容易进行区分处理了。 即面向消息的通信是有消息保护边界的。
- tcp是基于数据流的，于是收发的消息不能为空，这就需要在客户端和服务端都添加空消息的处理机制，防止程序卡住，而udp是基于数据报的，即便是你输入的是空内容（直接回车），那也不是空消息，udp协议会帮你封装上消息头
udp的recvfrom是阻塞的，一个recvfrom(x)必须对唯一一个sendinto(y),收完了x个字节的数据就算完成,若是y>x数据就丢失，这意味着udp根本不会粘包，但是会丢数据，不可靠
tcp的协议数据不会丢，没有收完包，下次接收，会继续上次继续接收，己端总是在收到ack时才会清除缓冲区内容。数据是可靠的，但是会粘包。
粘包解决方案：
方法一、
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from socket import * 4 import subprocess 5 6 ip_port = ('127.0.0.1', 8080) 7 back_log = 5 8 buffer_size = 1024 9 tcp_server = socket(AF_INET, SOCK_STREAM) 10 tcp_server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) 11 tcp_server.bind(ip_port) 12 tcp_server.listen(back_log) 13 while True: 14 conn, addr = tcp_server.accept() 15 while True: 16 try: 17 cmd = conn.recv(buffer_size) 18 # if not cmd:break 19 # 执行命令，得到命令的运行结果cmd_res 20 res = subprocess.Popen(cmd.decode('utf-8'), shell=True, 21 stderr=subprocess.PIPE, 22 stdout=subprocess.PIPE, 23 stdin=subprocess.PIPE 24 ) 25 err = res.stderr.read() 26 if err: 27 cmd_res = err 28 else: 29 cmd_res = res.stdout.read() 30 if not cmd_res: 31 cmd_res = 'excute success'.encode('utf-8') 32 length = len(cmd_res) 33 conn.send(str(length).encode('utf-8')) 34 client_ready = conn.recv(buffer_size) 35 if client_ready == b'ready': 36 conn.send(cmd_res) 37 38 except Exception as EX: 39 print(EX) 40 break 41 conn.close()
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from socket import * 4 5 ip_port = ('127.0.0.1', 8080) 6 back_log = 5 7 buffer_size = 1024 8 tcp_client = socket(AF_INET, SOCK_STREAM) 9 tcp_client.connect(ip_port) 10 while True: 11 cmd = input('>>>').strip() 12 if not cmd:continue 13 if cmd == 'quit':break 14 tcp_client.send(cmd.encode('utf-8')) 15 length = tcp_client.recv(buffer_size) 16 tcp_client.send(b'ready') 17 length = int(length.decode('utf-8')) 18 recv_size = 0 19 recv_msg = b'' 20 while recv_size < length: 21 # 第一种写法 22 # r_m = tcp_client.recv(buffer_size) 23 # recv_msg += r_m 24 # recv_size += len(r_m) 25 26 # 第二种写法 27 recv_msg += tcp_client.recv(buffer_size) 28 recv_size = len(recv_msg) 29 print(recv_msg.decode('gbk')) 30 tcp_client.close()
方法二、
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from socket import * 4 import subprocess 5 import struct 6 7 ip_port = ('127.0.0.1', 8080) 8 back_log = 5 9 buffer_size = 1024 10 tcp_server = socket(AF_INET, SOCK_STREAM) 11 tcp_server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) 12 tcp_server.bind(ip_port) 13 tcp_server.listen(back_log) 14 while True: 15 conn, addr = tcp_server.accept() 16 while True: 17 try: 18 cmd = conn.recv(buffer_size) 19 # if not cmd:break 20 # 执行命令，得到命令的运行结果cmd_res 21 res = subprocess.Popen(cmd.decode('utf-8'), shell=True, 22 stderr=subprocess.PIPE, 23 stdout=subprocess.PIPE, 24 stdin=subprocess.PIPE 25 ) 26 err = res.stderr.read() 27 if err: 28 cmd_res = err 29 else: 30 cmd_res = res.stdout.read() 31 if not cmd_res: 32 cmd_res = 'excute success'.encode('utf-8') 33 length = len(cmd_res) 34 data_length = struct.pack('i', length) 35 conn.send(data_length) 36 conn.send(cmd_res) 37 38 except Exception as EX: 39 print(EX) 40 break 41 conn.close()
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from socket import * 4 import struct 5 6 ip_port = ('127.0.0.1', 8080) 7 back_log = 5 8 buffer_size = 1024 9 tcp_client = socket(AF_INET, SOCK_STREAM) 10 tcp_client.connect(ip_port) 11 while True: 12 cmd = input('>>>').strip() 13 if not cmd:continue 14 if cmd == 'quit':break 15 tcp_client.send(cmd.encode('utf-8')) 16 length_data = tcp_client.recv(4) 17 length = struct.unpack('i', length_data)[0] 18 recv_size = 0 19 recv_msg = b'' 20 while recv_size < length: 21 # 第一种写法 22 # r_m = tcp_client.recv(buffer_size) 23 # recv_msg += r_m 24 # recv_size += len(r_m) 25 26 # 第二种写法 27 recv_msg += tcp_client.recv(buffer_size) 28 recv_size = len(recv_msg) 29 print(recv_msg.decode('gbk')) 30 tcp_client.close()
九 利用socketserver实现并发
基于TCP服务端:
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import socketserver 4 5 6 class MyServer(socketserver.BaseRequestHandler): 7 def handle(self): 8 print('conn is:', self.request) # <==>conn 9 print('addr is:', self.client_address) # <==> addr 10 while True: 11 try: 12 # 收消息 13 data = self.request.recv(1024) 14 if not data:break 15 print('收到客户端的消息是', data) 16 # 发消息 17 self.request.sendall(data.upper()) 18 except Exception as EX: 19 print('错误提示:',EX) 20 break 21 22 23 if __name__ == '__main__': 24 s = socketserver.ThreadingTCPServer(('127.0.0.1', 8080), MyServer) # 多线程;第一个参数，地址+端口；第二个参数，类
25 # s = socketserver.ForkingTCPServer(('127.0.0.1', 8080), MyServer) # 多进程；多进程的开销大于多线程
26 s.serve_forever()
基于TCP客户端:
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from socket import * 4 ip_port = ('127.0.0.1', 8080) 5 back_log = 5 6 buffer_size = 1024 7 tcp_client = socket(AF_INET, SOCK_STREAM) 8 tcp_client.connect(ip_port) 9 while True: 10 msg = input('>>').strip() 11 if not msg:continue 12 if msg == 'quit':break 13 tcp_client.send(msg.encode('utf-8')) 14 data = tcp_client.recv(buffer_size) 15 print('收到服务端发来的消息:', data.decode('utf-8')) 16 tcp_client.close()
基于UDP服务端
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import socketserver 4 5 6 class MyServer(socketserver.BaseRequestHandler): 7 def handle(self): 8 print(self.request) 9 print('收到客户端的消息是:', self.request[0].upper()) 10 self.request[1].sendto(self.request[0].upper(), self.client_address) 11 12 13 if __name__ == '__main__': 14 s = socketserver.ThreadingUDPServer(('127.0.0.1', 8080), MyServer) # 第一个参数，地址+端口；第二个参数，类 15 s.serve_forever()
基于UDP客户端
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from socket import * 4 ip_port = ('127.0.0.1', 8080) 5 buffer_size = 1024 6 udp_client = socket(AF_INET, SOCK_DGRAM) # SOCK_DGRAM:数据报式套接字 7 while True: 8 msg = input('>>').strip() 9 udp_client.sendto(msg.encode('utf-8'), ip_port) 10 data, addr = udp_client.recvfrom(buffer_size) 11 print(data.decode('utf-8')) 12 print(addr)
十 认证客户端的合法性
如果想在分布式系统中实现一个简单的客户端链接认证功能，又不像SSL那么复杂，那么可以利用hmac+加盐的方法来实现。
1 # _*_coding:utf-8_*_ 2 __author__ = 'Linhaifeng' 3 from socket import * 4 import hmac, os 5 6 secret_key = b'alex bang bang bang' 7 8 9 def conn_auth(conn): 10 ''' 11 认证客户端链接 12 :param conn: 13 :return: 14 ''' 15 print('开始验证新链接的合法性') 16 msg = os.urandom(32) # 随机生成的;b'\xa3\x9d\xaa\x94\x9e\x89\xe9\xc9\xc3r\xf9E\xe0w\x82=\xaac-\x04\xd8:\xea\x07\xad\x1dx\x1er\xe0\x7f\x02' 17 conn.sendall(msg) 18 h = hmac.new(secret_key, msg) # <hmac.HMAC object at 0x000000D5DA4824E0> 19 digest = h.digest() # 随机生成的;b'\x17!*\xae6\x81\xfe|)\x138\xfa2o%\x1a' 20 respone = conn.recv(len(digest)) 21 return hmac.compare_digest(respone, digest) # 比较第一个参数和第二个参数；相同，返回True，反之也成立 22 23 24 def data_handler(conn, bufsize=1024): 25 if not conn_auth(conn): 26 print('该链接不合法,关闭') 27 conn.close() 28 return 29 print('链接合法,开始通信') 30 while True: 31 data = conn.recv(bufsize) 32 if not data: break 33 conn.sendall(data.upper()) 34 35 36 def server_handler(ip_port, bufsize, backlog=5): 37 ''' 38 只处理链接 39 :param ip_port: 40 :return: 41 ''' 42 tcp_socket_server = socket(AF_INET, SOCK_STREAM) 43 tcp_socket_server.bind(ip_port) 44 tcp_socket_server.listen(backlog) 45 while True: 46 conn, addr = tcp_socket_server.accept() 47 print('新连接[%s:%s]' % (addr[0], addr[1])) 48 data_handler(conn, bufsize) 49 50 51 if __name__ == '__main__': 52 ip_port = ('127.0.0.1', 9999) 53 bufsize = 1024 54 server_handler(ip_port, bufsize)
1 # _*_coding:utf-8_*_ 2 __author__ = 'Linhaifeng' 3 from socket import * 4 import hmac, os 5 6 secret_key = b'alex bang bang bang' 7 8 9 def conn_auth(conn): 10 ''' 11 认证客户端链接 12 :param conn: 13 :return: 14 ''' 15 print('开始验证新链接的合法性') 16 msg = os.urandom(32) 17 conn.sendall(msg) 18 h = hmac.new(secret_key, msg) 19 digest = h.digest() 20 respone = conn.recv(len(digest)) 21 return hmac.compare_digest(respone, digest) 22 23 24 def data_handler(conn, bufsize=1024): 25 if not conn_auth(conn): 26 print('该链接不合法,关闭') 27 conn.close() 28 return 29 print('链接合法,开始通信') 30 while True: 31 data = conn.recv(bufsize) 32 if not data: break 33 conn.sendall(data.upper()) 34 35 36 def server_handler(ip_port, bufsize, backlog=5): 37 ''' 38 只处理链接 39 :param ip_port: 40 :return: 41 ''' 42 tcp_socket_server = socket(AF_INET, SOCK_STREAM) 43 tcp_socket_server.bind(ip_port) 44 tcp_socket_server.listen(backlog) 45 while True: 46 conn, addr = tcp_socket_server.accept() 47 print('新连接[%s:%s]' % (addr[0], addr[1])) 48 data_handler(conn, bufsize) 49 50 51 if __name__ == '__main__': 52 ip_port = ('127.0.0.1', 9999) 53 bufsize = 1024 54 server_handler(ip_port, bufsize)
十一 FTP服务器
实现如下功能：
1、用户加密认证
2、每个用户都有自己的家目录，且只能访问自己的家目录
3、允许用户在ftp server上随意切换目录(cd)
4、允许用户查看当前目录下的所有文件(ls)
5、允许上传和下载文件
6、文件传输过程中显示进度条
7、支持文件的断点续传
ftp server
目录结构：
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import os,sys 4 PATH = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) 5 sys.path.append(PATH) 6 7 8 from core import main 9 10 11 if __name__ == '__main__': 12 main.ArgvHandler()
1 [DEFAULT] 2 3 [alex] 4 Password =123 5 Quotation = 100 6 7 [root] 8 Password = root 9 Quotation = 100
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import os 4 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) 5 6 ip = '127.0.0.1' 7 port = 8080 8 9 ACCOUNT_PATH = os.path.join(BASE_DIR, 'conf', 'accounts.cfg')
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import socketserver 4 import optparse 5 import socketserver 6 from conf import settings 7 from core import server 8 9 10 class ArgvHandler(): 11 def __init__(self): 12 self.op = optparse.OptionParser() 13 # self.op.add_option('-s', '--server', dest='server') 14 # self.op.add_option('-P', '--port', dest='port') 15 options, args = self.op.parse_args() 16 # print(options) # {'server': '127.0.0.1', 'port': '8080'} 17 # print(type(options)) # <class 'optparse.Values'> 18 # print(options.server) # 127.0.0.1 19 # print(args) 20 21 options, args = self.op.parse_args() 22 self.verify_args(options, args) 23 24 def verify_args(self, options, args): 25 cmd = args[0] 26 # 第一种方法 27 # if cmd == 'start': 28 # pass 29 # else: 30 # pass 31 # 第二种方法 32 if hasattr(self, cmd): 33 func = getattr(self, cmd) 34 func() 35 36 def start(self): 37 print('ths server is working...') 38 s = socketserver.ThreadingTCPServer((settings.ip, settings.port), server.ServerHandler) 39 s.serve_forever() 40 41 def help(self): 42 pass
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import socketserver 4 import json 5 import configparser 6 from conf import settings 7 import os 8 9 STATUS_CODE = { 10 250: "Invalid cmd format,e.g:{'action':'get','filename':'test.py','size':344}", 11 251: "Invalid cmd", 12 252: "Invalid auth data", 13 253: "Wrong username or password", 14 254: "Passed authentication", 15 255: "Filename doesn't provided", 16 256: "File doesn't exist on server", 17 257: "ready to send file", 18 258: "md5 verification", 19 800: "the file exist,but not enough,is continue?", 20 801: "the file exist!", 21 802: "ready to receive datas", 22 900: "md5 valdate success" 23 } 24 25 26 class ServerHandler(socketserver.BaseRequestHandler): 27 def handle(self): 28 while True: 29 data = self.request.recv(1024).strip() # self.request=conn 30 data = json.loads(data.decode('utf8')) 31 if data.get('action'): 32 if hasattr(self, data.get('action')): 33 func = getattr(self, data.get('action')) 34 func(**data) 35 else: 36 print('Invalid cmd') 37 else: 38 print('Invalid cmd') 39 40 def send_response(self, status_code): 41 response = {'status_code': status_code} 42 self.request.sendall(json.dumps(response).encode('utf8')) 43 44 def auth(self, **data): 45 username = data['username'] 46 password = data['password'] 47 username = self.authenticate(username, password) 48 if username: 49 self.send_response(254) 50 else: 51 self.send_response(253) 52 53 def authenticate(self, username, password): 54 cfg = configparser.ConfigParser() 55 cfg.read(settings.ACCOUNT_PATH) 56 if username in cfg.sections(): 57 if cfg[username]['Password'] == password: 58 self.username = username 59 self.mainPath = os.path.join(settings.BASE_DIR, 'home', self.username) 60 print('passed authenticate') 61 return username 62 63 def put(self, **data): 64 print('data', data) 65 file_name = data.get('file_name') 66 file_size = data.get('file_size') 67 target_path = data.get('target_path') 68 abs_path = os.path.join(self.mainPath, target_path, file_name) 69 has_received = 0 70 if os.path.exists(abs_path): 71 file_has_size = os.stat(abs_path).st_size 72 if file_has_size < file_size: 73 # 断点续传 74 self.request.sendall('800'.encode('utf8')) 75 choice = self.request.recv(1024).decode('utf8') 76 if choice == 'Y': 77 self.request.sendall(str(file_has_size).encode('utf8')) 78 has_received = file_has_size 79 f = open(abs_path, 'ab') 80 else: 81 f = open(abs_path, 'wb') 82 else: 83 # 文件完全存在 84 self.request.sendall('801'.encode('utf8')) 85 return 86 else: 87 self.request.sendall('802'.encode('utf8')) 88 f = open(abs_path, 'wb') 89 90 while has_received < file_size: 91 try: 92 data = self.request.recv(1024) 93 except Exception as EX: 94 break 95 f.write(data) 96 has_received += len(data) 97 f.close() 98 99 def ls(self, **data): 100 file_list = os.listdir(self.mainPath) 101 file_str = '\n'.join(file_list) 102 if not len(file_list): 103 file_str = '<empty dir>' 104 self.request.sendall(file_str.encode('utf8')) 105 106 def cd(self, **data): 107 dirname = data.get('dirname') 108 if dirname == '..': 109 self.mainPath = os.path.dirname(self.mainPath) 110 else: 111 self.mainPath = os.path.join(self.mainPath, dirname) 112 self.request.sendall(self.mainPath.encode('utf8')) 113 114 def mkdir(self, **data): 115 dirname = data.get('dirname') 116 path = os.path.join(self.mainPath, dirname) 117 if not os.path.exists(path): 118 if '/' in path: 119 os.makedirs(path) 120 else: 121 os.mkdir(path) 122 self.request.sendall('create success'.encode('utf8')) 123 else: 124 self.request.sendall('dirname exist'.encode('utf8'))
ftp client
目录结构
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 # import socket 4 # 5 # sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 6 # sk.connect(('127.0.0.1', 8080)) 7 import optparse 8 import socket 9 import json 10 import os, sys 11 STATUS_CODE = { 12 250: "Invalid cmd format,e.g:{'action':'get','filename':'test.py','size':344}", 13 251: "Invalid cmd", 14 252: "Invalid auth data", 15 253: "Wrong username or password", 16 254: "Passed authentication", 17 255: "Filename doesn't provided", 18 256: "File doesn't exist on server", 19 257: "ready to send file", 20 258: "md5 verification", 21 800: "the file exist,but not enough,is continue?", 22 801: "the file exist!", 23 802: "ready to receive datas", 24 900: "md5 valdate success" 25 } 26 27 28 class ClientHandler(object): 29 def __init__(self): 30 self.op = optparse.OptionParser() 31 self.op.add_option('-s', '--server', dest='server') 32 self.op.add_option('-P', '--port', dest='port') 33 self.op.add_option('-u', '--username', dest='username') 34 self.op.add_option('-p', '--password', dest='password') 35 36 self.options, self.args = self.op.parse_args() 37 38 self.verify_args(self.options, self.args) 39 self.make_connection() 40 self.mainPath = os.path.dirname(os.path.abspath(__file__)) 41 self.last = 0 42 43 def verify_args(self, options, args): 44 server = options.server 45 port = options.port 46 username = options.username 47 password = options.password 48 if int(port) > 0 and int(port) < 65535: 49 return True 50 else: 51 exit('the port is in 0-65535') 52 53 def make_connection(self): 54 self.sock = socket.socket() 55 self.sock.connect((self.options.server, int(self.options.port))) 56 57 def interactive(self): 58 print('begin to interactive...') 59 # self.authenticate() 60 if self.authenticate(): 61 while 1: 62 cmd_info = input('[%s]' % self.current_dir).strip() 63 cmd_list = cmd_info.split() 64 if hasattr(self, cmd_list[0]): 65 func = getattr(self, cmd_list[0]) 66 func(*cmd_list) 67 68 def put(self, *cmd_list): 69 # put 12.png images 70 actions, local_path, target_path = cmd_list 71 local_path = os.path.join(self.mainPath, local_path) 72 file_name = os.path.basename(local_path) 73 file_size = os.stat(local_path).st_size 74 data = { 75 'action': 'put', 76 'file_name': file_name, 77 'file_size': file_size, 78 'target_path': target_path 79 } 80 self.sock.send(json.dumps(data).encode('utf8')) 81 is_exist = self.sock.recv(1024).decode('utf8') 82 has_send = 0 83 if is_exist == '800': 84 # 文件不完整 85 choice = input('the file exist,but not enough,is continue?[Y/N]').strip() 86 if choice.upper() == 'Y': 87 self.sock.sendall('Y'.encode('utf8')) 88 continue_position = self.sock.recv(1024).decode('utf8') 89 has_send += int(continue_position) 90 else: 91 self.sock.sendall('N'.encode('utf8')) 92 elif is_exist == '801': 93 # 文件完全存在 94 print('the file exist') 95 return 96 else: 97 pass 98 f = open(local_path, 'rb') 99 f.seek(has_send) 100 while has_send < file_size: 101 data = f.read(1024) 102 self.sock.sendall(data) 103 has_send += len(data) 104 self.show_progress(has_send, file_size) 105 f.close() 106 print('successfully upload!') 107 108 def show_progress(self, has, total): 109 rate = float(has)/float(total) 110 rate_num = int(rate*100) 111 if self.last != rate_num: 112 sys.stdout.write('%s%% %s\r' % (rate_num, '#'*rate_num)) 113 self.last = rate_num 114 115 def ls(self, *cmd_list): 116 data = { 117 'action': 'ls' 118 } 119 self.sock.sendall(json.dumps(data).encode('utf8')) 120 data = self.sock.recv(1024).decode('utf8') 121 print(data) 122 123 def cd(self, *cmd_list): 124 data = { 125 'action': 'cd', 126 'dirname': cmd_list[1] 127 } 128 self.sock.sendall(json.dumps(data).encode('utf8')) 129 data = self.sock.recv(1024).decode('utf8') 130 self.current_dir = os.path.basename(data) 131 132 def mkdir(self, *cmd_list): 133 data = { 134 'action': 'mkdir', 135 'dirname': cmd_list[1] 136 } 137 self.sock.sendall(json.dumps(data).encode('utf8')) 138 data = self.sock.recv(1024).decode('utf8') 139 140 def authenticate(self): 141 if self.options.username is None or self.options.password is None: 142 username = input('username:') 143 password = input('password:') 144 return self.get_auth_result(username, password) 145 return self.get_auth_result(self.options.username, self.options.password) 146 147 def response(self): 148 data = self.sock.recv(1024).decode('utf8') 149 data = json.loads(data) 150 return data 151 152 def get_auth_result(self, username, password): 153 data = { 154 'action': "auth", 155 'username': username, 156 'password': password 157 } 158 self.sock.send(json.dumps(data).encode('utf8')) 159 response = self.response() 160 print('response:', response['status_code']) 161 if response['status_code'] == 254: 162 self.username = username 163 self.current_dir = username 164 print(STATUS_CODE[response['status_code']]) 165 return True 166 else: 167 print(STATUS_CODE[response['status_code']]) 168 169 170 ch = ClientHandler() 171 ch.interactive()
十二 进程与线程
1、为什么要有操作系统？
现代计算机系统是由一个或者多个处理器、内存、硬盘、打印机、键盘、鼠标和显示器等组成的。网络接口以及各种其他输入/输出设备组成的复杂系统，每位程序员不可能掌握所有系统实现的细节，并且管理优化这些部件是一件具有挑战性极强的工作。所以，我们需要为计算机安装一层软件，成为操作系统，任务就是用户程序性提供一个简单清晰的计算机模型，并管理以上设备。
定义：操作系统是一个用来协调、管理和控制计算机硬件和软件资源的系统程序，它位于硬件和应用程序之间。程序是运行在系统上的具有某种功能的软件，比如：浏览器，音乐播放器等。
操作系统内部的定义：操作系统的内核是一个管理和控制程序，负责管理计算机的所有物理资源，其中包括：文件系统、内存管理、设备管理、进程管理。
2、什么是进程？
假如有两个程序A和B，程序A在执行到一半的过程中，需要读取大量的数据输入(I/O操作)，而此时CPU只能静静地等待任务A读取完数据才能继续执行，这样就白白浪费了CPU资源。是不是在程序A读取数据的过程中，让程序B去执行，当程序A读取完数据之后，让程序B暂停，然后让程序A继续执行？当然没问题，但这里有一个关键词：切换；既然是切换，那么这就涉及到了状态的保存，状态的恢复，加上程序A与程序B所需要的系统资源(内存，硬盘，键盘等等)是不一样的。自然而然的就需要有一个东西去记录程序A和程序B分别需要什么资源，怎样去识别程序A和程序B等等，所以就有了一个叫进程的抽象。
定义：
进程就是一个程序在一个数据集上的一次动态执行过程。进程一般由程序、数据库、进程控制块三部分组成。我们编写的程序用来描述进程要完成哪些功能以及如何完成。数据集则是程序在执行过程中所需要使用的资源。进程控制块用来记录的外部特征，描述进程的执行变化过程，系统可以利用它来控制和管理进程，它是系统感知进程存在的唯一标志。
本质上就是一段程序的运行过程（抽象的概念）
3、什么是线程？
线程的出现是为了降低上下文切换的消耗，提高系统的并发性，并突破一个进程只能干一样事的缺陷，让进程内并发成为可能。
4、进程与线程区别
1、一个程序至少有一个进程，一个进程至少有一个线程（进程可以理解成线程的容器）
2、进程在执行过程中拥有独立的内存单元，而多个线程共享内存，从而极大地提高了程序的运行效率
3、线程在执行过程中与进程还是有区别的，每个独立的线程有一个程序运行的入口，顺序执行序列和程序的出口。但是线程不能够独立执行，必须依存在应用程序中，由应用程序提供多个线程执行控制。
4、进程是具有一定独立功能的程序关于某个数据集合上的一次运行活动，进程是系统进行资源分配和调度的一个独立单位；线程是进程的一个实体，是CPU调度和分源的基本单位，它是比进程更小的能独立运行的基本单位，线程自己基本上不拥有系统资源，只拥有一点运行中必不可少的资源（如程序计数器，一组寄存器和钱），但是它可与同属一个进程的其他的线程共享进程所拥有的全部资源。一个进程可以创建和撤销另一个线程；同一个进程中的多个线程之间可以并发执行。
5、线程：最小的执行单元（实例）；进程：最小的资源单位
5、Python的GIL(全局解释锁;Global Interpreter Lock)
In CPython, the global interpreter lock, or GIL, is a mutex that prevents multiple native threads from executing Python bytecodes at once. This lock is necessary mainly because CPython’s memory management is not thread-safe. (However, since the GIL exists, other features have grown to depend on the guarantees that it enforces.)
上面的核心意思：无论你启多少个线程，你有多少个CPU，Python在执行的时候会淡定的在同一时刻只允许一个线程运行。
6、 线程的两种调用方式
threading 模块建立在thread 模块之上。thread模块以低级、原始的方式来处理和控制线程，而threading 模块通过对thread进行二次封装，提供了更方便的api来处理线程。
调用方式：
方式一、
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import threading 4 import time 5 6 7 def music(): 8 print('begin to listen % s' % time.ctime()) 9 time.sleep(3) 10 print('stop to listen %s' % time.ctime()) 11 12 13 def game(): 14 print('begin to play game % s \r\n' % time.ctime()) 15 time.sleep(5) 16 print('stop to play game %s \r\n' % time.ctime()) 17 18 19 threads = [] 20 t1 = threading.Thread(target=music) 21 t2 = threading.Thread(target=game) 22 threads.append(t1) 23 threads.append(t2) 24 if __name__ == '__main__': 25 # join()功能：在子线程完成运行之前，这个子线程的父线程讲一直被阻塞 26 # t1 = threading.Thread(target=music) 27 # t2 = threading.Thread(target=game) 28 # t1.start() 29 # t2.start() 30 # 31 # t1.join() 32 # t2.join() 33 # print('end') 34 35 # setDaemon():将线程生命为守护线程 36 t2.setDaemon(True) # 注：一定要在start之前设置 37 for t in threads: 38 t.start() 39 print('end')
方式二、
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import threading 4 import time 5 6 7 class MyThread(threading.Thread): 8 def __init__(self, num): 9 threading.Thread.__init__(self) 10 self.num = num 11 12 def run(self): # 定义每个线程要运行的函数 13 14 print("running on number:%s" % self.num) 15 16 time.sleep(3) 17 18 19 if __name__ == '__main__': 20 t1 = MyThread(1) 21 t2 = MyThread(2) 22 t1.start() 23 t2.start() 24 25 print("ending......")
join():在子线程完成运行之前，这个子线程的父线程将一直被阻塞。
setDaemon(True)：
将线程生命为守护线程，必须在start()方法调用之前设置，如果不设置为守护线程，程序会被无限挂起。这个方法基本和join是相反的。当我们在程序运行中，执行一个主线程，如果主线程又创建一个子线程，主线程和子线程就分兵两路，分别运行，那么当主线程完成想退出时，会验证子线程是否完成。如果子线程未完成，则主线程会等待子线程完成后再退出。但是有时候我们需要的是，只要主线程完成了，不管子线程是否完成，都要和主线程一起退出，这时就可以用setDaemon方法了。
其他方法：
1 # run():用于表示线程活动的方法 2 # start():启动线程活动 3 # isAlive():返回线程是否活动的,返回布尔值,True/False 4 # getName():返回线程名字 5 # setName():设置线程名字 6 7 threading模块提供的一些方法: 8 # threading.currentThread():返回当前的线程变量 9 # threading.enumerate():返回一个包含正在运行的线程的list。正在运行指线程启动后-结束前，不包括启动前和终止后的线程 10 # threading.activeCount():返回正在运行的线程数量，与len(threading.enumerate())有相同的结果
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import threading 4 import time 5 6 7 def music(): 8 print('begin to listen % s' % time.ctime()) 9 time.sleep(3) 10 print('stop to listen %s' % time.ctime()) 11 12 13 def game(): 14 print('begin to play game % s \r\n' % time.ctime()) 15 time.sleep(5) 16 print('stop to play game %s \r\n' % time.ctime()) 17 18 19 threads = [] 20 t1 = threading.Thread(target=music) 21 t2 = threading.Thread(target=game) 22 threads.append(t1) 23 threads.append(t2) 24 if __name__ == '__main__': 25 # join()功能：在子线程完成运行之前，这个子线程的父线程讲一直被阻塞 26 # t1 = threading.Thread(target=music) 27 # t2 = threading.Thread(target=game) 28 # t1.start() 29 # t2.start() 30 # 31 # t1.join() 32 # t2.join() 33 # print('end') 34 35 # setDaemon():将线程生命为守护线程 36 t2.setDaemon(True) # 注：一定要在start之前设置 37 for t in threads: 38 t.start() 39 print(t.getName()) 40 print('count:', threading.activeCount()) 41 while threading.activeCount() == 3: 42 print('end')
7、同步锁(lock)
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import threading 4 import time 5 num = 100 6 7 8 def sub(): 9 global num 10 print('ok') 11 lock.acquire() # 加锁 12 temp = num 13 time.sleep(0.001) 14 num = temp-1 15 lock.release() # 释放锁 16 17 18 li = [] 19 lock = threading.Lock() 20 for i in range(100): 21 t1 = threading.Thread(target=sub) 22 t1.start() 23 li.append(t1) 24 for l in li: 25 l.join() 26 print(num)
注：多个线程都在同时操作同一个共享资源，所以造成了资源破坏(join会造成串行，失去线程的意义)，可以通过同步锁来解决这种问题。
8、递归锁
在线程间共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁，因为系统判断这部分资源都在使用，所有这两个线程在无外力作用下将一直等待下去。
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import threading 4 import time 5 6 7 class MyThread(threading.Thread): 8 9 def actionA(self): 10 r_lock.acquire() 11 print(self.name, 'gotA', time.ctime()) # 重写线程后的self.name --->线程的名字 12 time.sleep(2) 13 r_lock.acquire() 14 print(self.name, 'gotB', time.ctime()) 15 time.sleep(1) 16 r_lock.release() 17 r_lock.release() 18 19 def actionB(self): 20 r_lock.acquire() 21 print(self.name, 'gotB', time.ctime()) # 重写线程后的self.name --->线程的名字 22 time.sleep(2) 23 r_lock.acquire() 24 print(self.name, 'gotA', time.ctime()) 25 time.sleep(1) 26 r_lock.release() 27 r_lock.release() 28 29 def run(self): 30 self.actionA() 31 self.actionB() 32 33 34 if __name__ == '__main__': 35 r_lock = threading.RLock() 36 li = [] 37 for t in range(3): 38 t = MyThread() 39 t.start() 40 li.append(t) 41 42 for i in li: 43 i.join() 44 45 print('end')
为了支持在同一线程中多次请求同一资源，Python提供了“可重入锁”：threading.Rlock。Rlock内部维护着一个Lock和counter变量，counter记录了acquire的次数，从而使得资源可以被多次acquire。直到一个线程所有的acquire都被release，其他的线程才能获得资源。
9、同步对象(Event)
An event is a simple synchronization object;the event represents an internal flag,
and threads can wait for the flag to be set, or set or clear the flag themselves.
event = threading.Event()
# a client thread can wait for the flag to be set
event.wait()
# a server thread can set or reset it
event.set()
event.clear()
If the flag is set, the wait method doesn’t do anything.
If the flag is cleared, wait will block until it becomes set again.
Any number of threads may wait for the same event.
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import threading 4 import time 5 6 7 class Boss(threading.Thread): 8 def run(self): 9 print('Boss:今天加班到22：00！\r\n') 10 print(event.isSet()) # False 11 event.set() 12 time.sleep(6) 13 print('Boss:可以下班了，明天放假！\r\n') 14 print(event.isSet()) 15 event.set() 16 17 18 class Worker(threading.Thread): 19 def run(self): 20 event.wait() # 一旦event被设定，等同于pass 21 print('Worker:唉···命真苦！\r\n') 22 time.sleep(1) 23 event.clear() 24 event.wait() 25 print('Worker:OhYeah!\r\n') 26 27 28 if __name__ == '__main__': 29 event = threading.Event() 30 threads = [] 31 for i in range(5): 32 threads.append(Worker()) 33 threads.append(Boss()) 34 for t in threads: 35 t.start() 36 for t in threads: 37 t.join() 38 print('end')
10、信号量
信号量用来控制线程并发数的，BoundedSemaphore或Semaphore管理一个内置的计数器，每当调用acquire()时-1，调用release()时+1。计数器不能小于0，当计数器为0时，acquire()将阻塞线程至同步锁状态，直到其他线程调用release()。(类似于停车位的概念)BoundedSemaphore与Semaphore的唯一区别在于前者将在调用release()时检查计数器的值是否超过了计数器的初始值，如果超过了将抛出一个异常。
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import threading 4 import time 5 6 7 class MyThread(threading.Thread): 8 def run(self): 9 if semaphore.acquire(): 10 print(self.name, '\r') 11 time.sleep(5) 12 semaphore.release() 13 14 15 if __name__ == '__main__': 16 semaphore = threading.Semaphore(5) 17 threads = [] 18 for i in range(100): 19 threads.append(MyThread()) 20 for t in threads: 21 t.start()
11、队列(queue)
列表是不安全的数据结构：
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import threading 4 import time 5 6 li = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 7 8 9 def Foo(): 10 while li: 11 try: 12 last_li = li[-1] 13 print(last_li) 14 li.remove(last_li) 15 time.sleep(1) 16 except Exception as EX: 17 print('错误提示:', last_li, EX) 18 19 20 t1 = threading.Thread(target=Foo) 21 t1.start() 22 t2 = threading.Thread(target=Foo) 23 t2.start()
queue队列类的方法：
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import queue # 线程队列 4 q = queue.Queue() # 创建q对象,同步实现的。队列长度可为无限或者有限。可通过Queque的构造函数的可选参数maxsize来设定队列长度。如果maxsize小于1就表示队列长度无限 5 q.put(12) # 将一个值放到队列，调用队列对象的put()方法在队尾插入一个项目。put()有两个参数，第一个item为必需的，为插入项目的值；第二个block为可选参数，默认为1.如果队列当前为空且block为1，put()方法就使调用线程暂停，直到空处一个数据单元。如果block为0，put方法将引发Full异常 6 q.put('alex') 7 q.put({"age": 15}) 8 print(q.get()) # 将一个值从队列中取出，调用队列对象的get()方法，从对头删除并返回一个实例。可选参数为block，默认为True。如果队列为空且block为True，get()就使调用线程暂停，直至有项目可用。如果队列为空且block为False，队列将引发Empty异常 9 print(q.qsize()) # 返回队列的大小 10 print(q.empty()) # 判断队列是否为空，返回布尔值，True/False 11 print(q.full()) # 判断队列是否已经满了，返回布尔值，True/False 12 q.join() # 实际上意味着等到队列为空，再执行别的操作 13 14 ''' 15 Queue模块的三种队列及构造函数 16 1、Python Queue模块的FIFO队列，先进先出 class queue.Queue(maxsize) 17 2、LIFO 类似于堆，即先进后处。 class queue.LifoQueue(maxsize) 18 3、优先级队列，级别越低月先出来。 class queue.PriorityQueue(maxsize) 19 '''
生产者消费者模型
为什么要使用生产者和消费者模式？
在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
什么是生产者消费者模式？
生产者和消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import time 4 import random 5 import queue 6 import threading 7 q = queue.Queue() 8 9 10 def Producer(name): 11 count = 0 12 while count < 10: 13 print('making..') 14 time.sleep(random.randrange(3)) 15 q.put(count) 16 print('Producer [%s] has produced %s meat bun。\r' % (name, count)) 17 count += 1 18 print('ok\r') 19 20 21 def Consumer(name): 22 count = 0 23 while count < 10: 24 time.sleep(random.randrange(4)) 25 if not q.empty(): 26 data = q.get() 27 print('lm Consumer [%s] has eat %s meat bun。\r' % (name, data)) 28 else: 29 print('---no meat bun anymore----\r') 30 count += 1 31 32 33 p1 = threading.Thread(target=Producer, args=('alex',)) 34 c1 = threading.Thread(target=Consumer, args=('B',)) 35 p1.start() 36 c1.start()
12、并发&并行
并发：指系统具有处理多个任务(动作)的能力
并行：指系统具有同时处理多个任务(动作)的能力
13、同步&异步
同步：当进程执行到一个IO(等待外部数据)的时候你等
异步：当进程执行到一个IO(等待外部数据)的时候你不等；一直等到数据接收完成，在回来处理
14、任务类型
IO密集型：Python的多线程是有意义的
计算密集型：Python的多线程就不推荐，可以采用多进程+协程
16、多进程模块( multiprocessing)
Multiprocessingmultiprocessingmultiprocessing
由于GIL的存在，Python中的多线程其实并不是真正的多线程，如果想充分地使用多核CPU的资源，在Python中大部分情况下需要使用多进程。multiprocessing包是Python中的多进程管理包。与threading.Thread类似，它可以利用multiprocessing.Process对象来创建一个进程。该进程可以运行在Python程序内部编写的函数。该Process对象与Thread对象的用法，也有start()，run()，join()的方法。此外multiprocessing包中也有Lock/Event/Semaphore/Condition类(这些对象可以像多线程那样，通过参数传递给各个进程)，用以同步进程，其用法与threading包中的同名类一致。所以，multiprocessing的很大一部分与threading使用同一套API，只不过换到了多进程的情景。
调用方式一：
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from multiprocessing import Process 4 import time 5 6 7 def Foo(name): 8 time.sleep(1) 9 print('hello', name, time.ctime()) 10 11 12 if __name__ == '__main__': 13 p_list = [] 14 for i in range(200): 15 p = Process(target=Foo, args=('alex',)) 16 p_list.append(p) 17 p.start() 18 for i in p_list: 19 p.join() 20 print('end')
调用方式二：
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from multiprocessing import Process 4 import time 5 6 7 class MyProcess(Process): 8 def __init__(self): 9 super(MyProcess, self).__init__() 10 11 def run(self): 12 time.sleep(1) 13 print('hello', self.name, time.ctime()) 14 15 16 if __name__ == '__main__': 17 p_list = [] 18 for i in range(3): 19 p = MyProcess() 20 p.start() 21 p_list.append(p) 22 for p in p_list: 23 p.join() 24 print('end')
Process类
构造方法：
Process([group [, target [, name [, args [, kwargs]]]]])
group:线程组，目前还没有实现，库引用中提示必须是None
target:要执行的方法
name:进程名
args/kwargs:要传入方法的参数
实例方法：
is_alive():返回进程是否在运行
join([timeout]):阻塞当前上下文环境的进程，直到调用此方法的进程终止或到达指定的timeout(可选参数)
start():进程准备就绪，等待CPU调度
run():start()调用run方法，如果实例进程时未指定传入target，这start执行t默认run()方法
terminate():不管任务是否完成，立即停止工作进程
属性：
daemon:和线程的setDeamon功能一样
name:进程名字
pid:进程号
17、进程的通信
进程队列Queue
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import queue 4 import multiprocessing 5 6 7 def Foo(q): 8 q.put(123) 9 q.put(456) 10 11 12 if __name__ == '__main__': 13 q = multiprocessing.Queue() # 注意：此处需用进程队列，不能用线程队列，即q=queue.Queue() 14 p = multiprocessing.Process(target=Foo, args=(q,)) 15 p.start() 16 print(q.get()) 17 print(q.get())
管道
The Pipe() function returns a pair of connection objects connected by a pipe which by default is duplex (two-way). For example:
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from multiprocessing import Process, Pipe 4 5 6 def Foo(child_conn): 7 child_conn.send([12, {'name': 'alice'}, 'hello']) 8 response = child_conn.recv() 9 print('response:', response) 10 child_conn.close() 11 print('q_id2:', id(child_conn)) 12 13 14 if __name__ == '__main__': 15 parent_conn, child_conn = Pipe() 16 print('q_ID1', id(child_conn)) 17 p = Process(target=Foo, args=(child_conn,)) 18 p.start() 19 print(parent_conn.recv()) 20 parent_conn.send('早上好!') 21 p.join()
Managers
Queue和pipe只是实现了数据交互，并没实现数据共享，即一个进程去更改另一个进程的数据。
A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.
A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array. For example：
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from multiprocessing import Process, Manager 4 5 6 def Foo(dic, li, i): 7 dic[i] = '1' 8 li.append(i) 9 10 11 if __name__ == '__main__': 12 with Manager() as manager: 13 dic = manager.dict() 14 li = manager.list(range(5)) 15 p_list = [] 16 for i in range(10): 17 p = Process(target=Foo, args=(dic, li, i)) 18 p.start() 19 p_list.append(p) 20 for p in p_list: 21 p.join() 22 print(dic) 23 print(li)
进程同步
Without using the lock output from the different processes is liable to get all mixed up.
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from multiprocessing import Process, Lock 4 5 6 def Foo(lk, i): 7 with lk: # 默认情况下，已经lk.acquire() 8 print('hello world %s' % i) 9 10 11 if __name__ == '__main__': 12 lock = Lock() 13 for num in range(10): 14 Process(target=Foo, args=(lock, num)).start()
进程池
进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进程，那么程序就会等待，直到进程池中有可进程为止。
进程池中两个方法：
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from multiprocessing import Process, Pool 4 import time, os 5 6 7 def Foo(i): 8 time.sleep(1) 9 print('i = \r', i) 10 11 12 def Bar(arg): # 此处arg=Foo()函数的返回值 13 print('pgid-->%s\r' % os.getpid()) 14 print('ppid-->%s\r' % os.getppid()) 15 print('logger:%s\r' % arg) 16 17 18 if __name__ == '__main__': 19 pool = Pool(5) 20 Bar(1) 21 print('------------\r') 22 for i in range(10): 23 # pool.apply(func=Foo, args=(i,)) # 同步接口 24 # pool.apply_async(func=Foo, args=(i,)) 25 pool.apply_async(func=Foo, args=(i,), callback=Bar) # callback-->回调函数：就是某个动作或者函数执行成功后再去执行的函数
26 pool.close() 27 pool.join() # join和close位置不能反 28 print('end\r')
十三 协程
协程：又称微线程，英文名：Coroutine，本质上是一个线程
优点1：协程具有极高的执行效率。因为子程序切换不是线程切换，而是由程序自身控制。因此，没有线程切换的开销，和多线程比，线程数量越多，协程的性能优势就越明显。
优点2：不需要多线程的锁机制，因为只有一个线程，也不存在同时写变量冲突，在协程中控制共享资源不加锁，只需要判断状态就好了，所以执行效率比多线程高很多。
因为协程是一个线程执行，那怎么利用多核CPU呢？最简单的方法就是多进程+协程，即充分利用多核，又充分发挥协程的高效率，可获得极高的性能。
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import time 4 5 6 def consumer(name): 7 print('---->ready to eat meat bun') 8 while True: 9 new_meat_bun = yield 10 print('[%s] is eating meat bun %s' % (name, new_meat_bun)) 11 12 13 def producer(): 14 con1.__next__() 15 con2.__next__() 16 n = 0 17 while 1: 18 time.sleep(1) 19 print('\033[32;1m[producer]\033[0m is making meat bun %s and %s' % (n, n + 1)) 20 con1.send(n) 21 con2.send(n + 1) 22 n += 2 23 24 25 if __name__ == '__main__': 26 con1 = consumer('alex') 27 con2 = consumer('alice') 28 producer()
Greenlet
greenlet是一个用C实现的协程模块，相比与Python自带的yield，它可以使你在任意函数之间随意切换，而不需把这个函数先声明为generator。(注：需要用pip安装包;pip install gevent)
1 from greenlet import greenlet 2 import time 3 4 5 def A(): 6 while 1: 7 print('-------A-------') 8 time.sleep(0.5) 9 g2.switch() 10 11 12 def B(): 13 while 1: 14 print('-------B-------') 15 time.sleep(0.5) 16 g1.switch() 17 18 19 g1 = greenlet(A) # 创建协程g1 20 g2 = greenlet(B) 21 22 g1.switch() # 跳转至协程g1
gevent
1 import gevent 2 import requests, time # 此处requests需安装模块;pip --timeout=100 install requests 3 4 start = time.time() 5 6 7 def f(url): 8 print('GET: %s' % url) 9 resp = requests.get(url) # 爬网页的标签 10 data = resp.text 11 print('%d bytes received from %s.' % (len(data), url)) 12 13 14 gevent.joinall([ 15 gevent.spawn(f, 'https://www.qq.com/'), 16 gevent.spawn(f, 'https://www.baidu.com/'), 17 gevent.spawn(f, 'https://www.taobao.com/'), 18 ]) 19 20 print("cost time:", time.time() - start)
十四 缓存I/O
缓存I/O又被称作标准I/O，大多数文件系统的默认I/O操作都是缓存I/O。在Linux的缓存I/O机制中，操作系统会将I/O的数据缓存在文件系统的页缓存(page cache)中，也就是说，数据会先被拷贝到操作系统内核的缓冲区中，然后才会从操作系统内的缓冲区拷贝到应用程序的地址空间。用户空间没法直接访问内核空间的，内核态到用户态的数据拷贝。
缓存I/O的缺点：数据在传输过程中需要在应用程序地址空间和内核进行多次数据拷贝操作，这些数据拷贝操作所带来的CPU以及内存开销是非常大的。
I/O发生时设计的对象和步骤：
对于一个network IO（以read举例），他会涉及到两个系统对象，一个是调用这个IO的process(or thread)，另一个就是系统内核(kernel)。当一个read操作发生时，它会经历两个阶段：
1、等待数据准备(Waiting for the data to be read)
2、将数据从内核拷贝到进程中(Copying the data from the kernel to the process)
注：这两点很重要，因为这些IO Mode的区别就是在这两个阶段上各有不同的情况。
blocking IO(阻塞IO，Linux下)
在Linux中，默认情况下所有的socket都是blocking，一个典型的读操作大概流程图：
当用户进程调用了recvfrom这个系统调用，kernel就开始了IO的第一阶段:准备数据。对于network IO来说，很多时候数据在一开始还没到达(如：还没收到一个完整的UDP包)，这个时候kernel就要等待足够的数据到来。而在用户进程这边，整个进程会被阻塞。当kernel一直等到数据准备好了，它就将数据从kernel中拷贝到用户内存，然后kernel。所以，blocking IO的特点就是在IO执行的两个阶段都被block了。
non-blocking IO(非阻塞IO,Linux下)
在Linux下，可以通过设置socket使其变为non-blocking。当对一个non-blocking socket执行时大概的流程:
从上图可以看出，当用户进程发出read时，如果kernel中的数据还没准备好，那么它并不会block用户进程，而是立即返回一个error。从用户进程角度讲来讲，它发起一个read操作后，并不需要等待，而是马上就得到了一个结果。用户进程判断结果是一个error时，它就知道数据还没准备好，于是它可以再次发送read操作。一旦kernel中的数据准备好了。所以，用户进程其实是需要不断的主动询问kernel数据好了没有。
1 import time 2 import socket 3 sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 4 sk.bind(('127.0.0.1',8080)) 5 sk.listen(5) 6 sk.setblocking(False) # 设置是否阻塞,默认为True，非阻塞 7 while True: 8 try: 9 print('waiting client connection...') 10 conn,addr = sk.accept() # 进程主动轮询 11 print('+++',addr) 12 data = sk.recv(1024) 13 print(str(data,'utf8')) 14 conn.close() 15 except Exception as EX: 16 print('错误提示:',EX) 17 time.sleep(2)
1 import time 2 import socket 3 sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM) 4 while True: 5 sk.connect(('127.0.0.1',8080)) 6 print('hello') 7 sk.sendall(bytes('hello','utf8')) 8 time.sleep(1) 9 break
IO multiplexing(IO多路复用)
有些地方也称为这种IO方式为event driven IO。它的基本原理就是select/epoll这个function会不断的轮询所负责的所有socket，当某个socket有数据到达了，就通知用户进程，大概流程图：
当用户进程调用了select，那么真个进程会被block。而同时，kernel会“监视”所有select负责的socket，当任何一个socket中的数据准备好了，select就会返回。这时用户进程再调用read操作，将数据从kernel拷贝到用户进程。(如果处理的连接数不是很多的话，使用select/epoll的web server不一定比使用multi-threading+blocking IO的web server性能更好，可能延迟更大；select/epoll的优势并不是对于单个连接能处理得更快，而是在于能处理更多的连接)
注：
1、select函数返回结果中如果有文件可读了，那么进程就可以同故宫调用accept()或recv()来让kernel将位于内核中准备到的数据copy到用户区。
2、select的优势在于可以处理多个连接，不适用于单个连接。
1 import socket 2 import select 3 4 sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 5 sk.bind(('127.0.0.1', 8080)) 6 sk.listen(5) 7 inputs = [sk, ] 8 while True: 9 r, w, e = select.select(inputs, [], [], 3) 10 for obj in r: 11 print('obj:', obj) 12 if obj == sk: 13 conn, addr = obj.accept() 14 print('已连接：', conn) 15 inputs.append(conn) 16 else: 17 data_byte = obj.recv(1024) 18 print(str(data_byte, 'utf8')) 19 inp = input('回答[%s]号客户端>>:' % inputs.index(obj)) 20 obj.sendall(bytes(inp, 'utf8')) 21 print('>>>', r)
1 import socket 2 3 sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 4 sk.connect(('127.0.0.1', 8080)) 5 6 while True: 7 inp = input(">>>>") 8 sk.sendall(bytes(inp, "utf8")) 9 data = sk.recv(1024) 10 print(str(data, 'utf8'))
Asynchronous I/O(异步IO)
流程图：
从图中可以看出，用户进程发起read操作之后，立刻就开始去做其它的事。另一方面，从kernel的角度，当他受到一个asynchronous read之后，首先它会立刻返回，所以不会对用户进程产生任何block。然后，kernel会等待数据准备完成，然后将数据拷贝到用户内存，当这一切都完成之后，kernel会给用户进程发送一个signal，告诉它read操作完成了。
selectors模块：
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import selectors 4 import socket 5 6 sel = selectors.DefaultSelector() # 根据系统，默认选择最优IO多用户模式 7 8 9 def accept(sock, mask): 10 conn, addr = sock.accept() 11 print('accepted', conn, 'from', addr) 12 conn.setblocking(False) 13 sel.register(conn, selectors.EVENT_READ, read) 14 15 16 def read(conn, mask): 17 try: 18 data = conn.recv(1000) 19 if not data: 20 raise Exception 21 print('收到:', data.decode('utf8')) 22 conn.send(data.upper()) # Hope it won't block 23 except Exception as EX: 24 print('closing:', conn) 25 sel.unregister(conn) # 解除绑定 26 conn.close() 27 28 29 sock = socket.socket() 30 sock.bind(('localhost', 8080)) 31 sock.listen(100) 32 sock.setblocking(False) 33 sel.register(sock, selectors.EVENT_READ, accept) # sock与accept绑定 34 while True: 35 events = sel.select() # 监听[sock,conn1,conn2....] 36 for key, mask in events: 37 callback = key.data 38 print('>>callback:', callback) 39 callback(key.fileobj, mask)
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import socket 4 sk = socket.socket() 5 sk.connect(('127.0.0.1', 8080)) 6 while True: 7 inp = input('>>>') 8 sk.send(inp.encode('utf8')) 9 data = sk.recv(1024) 10 print(data.decode('utf8'))
所属网站分类: 技术文章 > 博客
作者：iuie9493
链接：https://www.pythonheidong.com/blog/article/12066/893b6d5d3f394ddad704/
来源：python黑洞网
任何形式的转载都请注明出处,如有侵权 一经发现 必将追究其法律责任
昵称:
评论内容：(最多支持255个字符)
---无人问津也好，技不如人也罢，你都要试着安静下来，去做自己该做的事，而不是让内心的烦躁、焦虑，坏掉你本来就不多的热情和定力