Python基础学习（5）网络编程socket、文件上传、粘包问题、socketserver、IO多路复用、线程与进程、进程池、线程池、上下文管理、协程

1. socket套接字，

简单的Server与Client案例：

import socket
# Server
server = socket.socket()
server.bind(("127.0.0.1", 9999, ))
server.listen(5)

# 接收客户端的请求

# 连接客户端地址信息
while True:
conn, addr = server.accept()
print(conn, addr)

import socket
# Client
client = socket.socket()
client.connect(("127.0.0.1", 9999, ))
client.close()

实现简单交互的socket案例：

import socket
# Server
server = socket.socket()
server.bind(("127.0.0.1", 9999, ))
server.listen(5)

# 接收客户端的请求

# 连接客户端地址信息
while True:
conn, addr = server.accept()
conn.sendall(bytes("Hello World", encoding = "utf-8"))
while True:
rec_bytess = conn.recv(1024)
rec_str = str(rec_bytess, encoding="utf-8")

if rec_str == "exit":
break

conn.sendall(bytes(rec_str + "!", encoding="utf-8"))

print(conn, addr)

import socket
# Client
client = socket.socket()
client.connect(("127.0.0.1", 9999, ))

rec = client.recv(1024)
rec_str = str(rec, encoding="utf-8")
print(rec_str)

while True:
inp = input("请输入要发送的内容：")

if inp == "exit":
client.sendall(bytes(inp, encoding="utf-8"))
break
else:
client.sendall(bytes(inp, encoding="utf-8"))
ret = str(client.recv(1024), encoding="utf-8")
print(ret)

client.close()

2.socket的更多功能

1）sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)

参数一：地址簇

socket.AF_INET IPV4（默认）

socket.AF_INET6 IPV6

socket.AF_UNIX 只能够用于单一的Unix系统进程间tongxin

参数二：类型

socket.SOCK_STREAM 流式socket, for TCP（默认）

socket.SOCK_DGRAM 数据报式socket，for UDP

socket.SOCK_RAW 原始套接字，普通的套接字无法处理ICMP、IGMP等网络报文，二SOCK_RAW可以，其次，SOCK_RAW也可以处理特殊的IPv4报文，此外，利用原始套接字，可以通过IP_HDRINCL套接字选项由用户构造IP头。

socket.SOCK_RDM是一种可靠的UDP形式，即保证交付数据报但不保证顺序。SOCK_RAM用来提供对原始协议的低级访问，在需要执行某些特殊操作时使用，如发送ICMP报文。SOCK_RAM通常仅限于高级用户或管理员运行的程序使用。

socket.SOCK_SEQPACKET 可靠的连续数据包服务

参数三：协议

0 （默认）与特定的地址家族相关的协议，如果是0，则系统就会根据地址格式和套接类别，自动选择一个合适的协议

2）sk.bind(address)

sk.bind(address)将套接字绑定到地址。address地址的格式取决于地址簇。在AF_INET下，以元组(host,port)的形式表示地址。

3）sk.listen(backlog)

开始监听传入连接。backlog指定在拒绝连接之前，可以挂起的最大连接数量。

backlog等于5，表示内核已经接到了连接请求，但服务器还没有调用accept进行处理的连接个数最大为5

这个值不能无限大，因为要在内核中维护连接队列

4）sk.setblocking(bool)

是否阻塞（默认True），如果设置False，那么accept和recv时一旦无数据，则报错。

5）sk.accept()

接受连接并返回（conn, address），其中conn是新的套接字对象，可以用来接收和发送数据。address是连接客户端的地址。

接收TCP客户的连接（阻塞式）等待连接的到来

6）sk.connect(address)

连接到address处的套接字。一般address的格式为元组(hostname, port)，如果连接出错，返回socket.error错误。

7）sk.connect_ex(address)

同上，只不过会返回值，连接成功时返回0，连接失败时返回编码，例如：10061

8）sk.close()

关闭套接字

9）sk.recv(bufsize[,flag])

接收套接字的数据。数据以字符串形式返回，bufsize指定最多可以接收的数量。flag提供有关消息的其他信息，通常可以忽略。

10）sk.recvfrom(bufsize[.flag])

与recv()类似，但返回值是(data, address)。其中data是包含接收数据的字符串，address是发送数据的套接字地址。

11）sk.send(string[,flag])

将string中的数据发送到连接的套接字。返回值是要发送的字节数量，该数量可能小于string的字节大小。即：可能未将指定内容全部发送。

12）sk.sendall(string[,flag])

将string中的数据发送到连接的套接字，但在返回之前会尝试发送所有数据。成功返回None，失败则抛出异常

内部通过递归调用send，将所有内容发送出去。

13）sk.sendto(string[,flag],address)

将数据发送到套接字，address是形式为(ipaddr, port)的元组，指定远程地址。返回值是发送的字节数。该函数主要用于UDP协议。

14）sk.settimeout(timeout)

设置套接字操作的超时期，timeout是一个浮点数，单位是秒。值为None表示没有超时期。一般，超时期应该在刚创建套接字时设置，因为它们可能用于连接的操作（如client连接最多等待5s）

15）sk.getpeername()

返回连接套接字的远程地址。返回值通常是元组(ipaddr, port)

16）sk.getsockname()

返回套接字自己的地址。通常是一个元组(ipaddr, port)

17）sk.fileno()

套接字的文件描述符

3. 使用socket实现文件上传

import socket
# Server
server = socket.socket()
server.bind(("127.0.0.1", 9999, ))
server.listen(5)

# 接收客户端的请求

# 连接客户端地址信息
while True:
conn, addr = server.accept()
# 先接收文件大小
file_size = str(conn.recv(1024), encoding="utf-8")
total_size = int(file_size)
has_recv = 0

f = open("logo.png", "wb")
# 然后再开始接收，直到获取完毕
while True:
if has_recv == total_size:
break
data = conn.recv(1024)
f.write(data)
has_recv += len(data)

f.close()

import socket
import os
# Client
client = socket.socket()
client.connect(("127.0.0.1", 9999, ))

# 发送当前文件大小
file_size = os.stat("fudan.png").st_size
client.sendall(bytes(str(file_size), encoding="utf-8"))

with open("fudan.png", "rb") as f:
for line in f:
client.sendall(line)

client.close()

4. 解决粘包问题

import socket
# Server
server = socket.socket()
server.bind(("127.0.0.1", 9999, ))
server.listen(5)

# 接收客户端的请求

# 连接客户端地址信息
while True:
conn, addr = server.accept()
# 先接收文件大小
file_size = str(conn.recv(1024), encoding="utf-8")
# 接收到文件大小之后，再向客户端发送一个确认信息，以解决粘包问题
conn.sendall(bytes("Start to transfer the file", encoding="utf-8"))
total_size = int(file_size)
has_recv = 0

f = open("logo.png", "wb")
# 然后再开始接收，直到获取完毕
while True:
if has_recv == total_size:
break
data = conn.recv(1024)
f.write(data)
has_recv += len(data)

f.close()

import socket
import os
# Client
client = socket.socket()
client.connect(("127.0.0.1", 9999, ))

# 发送当前文件大小
file_size = os.stat("fudan.png").st_size
client.sendall(bytes(str(file_size), encoding="utf-8"))
# 发送文件大小之后，等待服务器端的确认信息，以解决粘包问题
client.recv()

with open("fudan.png", "rb") as f:
for line in f:
client.sendall(line)

client.close()

5. socketserver，并发处理多个客户端请求

使用方法：

创建类，必须继承socketserver.BaseRequestHandler
实现handle方法
使用server_forever启动循环

socketserver服务器端实现的简单案例：

import socketserver

class MyServer(socketserver.BaseRequestHandler):

def handle(self):
# self.request客户端的连接；
# self.client_address客户端地址；
# self.server当前服务器socket对象
conn = self.request
conn.sendall(bytes("Hello World", encoding="utf-8"))
while True:
ret_bytes = conn.recv(1024)
ret_str = str(ret_bytes, encoding="utf-8")
if ret_str == "q":
break
conn.sendall(bytes(ret_str + "!", encoding="utf-8"))

if __name__ == '__main__':
server = socketserver.ThreadingTCPServer(("127.0.0.1", 8008), MyServer)
server.serve_forever()

6. IO多路复用
IO多路复用：通过一种机制，可以监视多个文件描述符（文件句柄），一旦某个描述符就绪（一般是读就绪或者写就绪），能够通知程序进行相应的读写操作。与python没有关系，是系统底层实现的

简单案例：

import socket
# Server
sk1 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sk1.bind(("127.0.0.1", 8001))
sk1.listen(5)

sk2 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sk2.bind(("127.0.0.1", 8002))
sk2.listen(5)

sk3 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sk3.bind(("127.0.0.1", 8003))
sk3.listen(5)

inputs = [sk1, sk2, sk3]

import select
while True:
# [sk1, sk2, sk3]内部自动监听sk1, sk2连个对象，
# 一旦某个文件描述符发生变化，则进行操作
# 如果有人连接sk1，r_list = [sk1]
# 第一个参数（r_list）为一有变化便能监听到，第三个参数（e_list）为一发生错误能监听到
104eb
# 只要有值w_list就能获取值，第二个参数传了什么会原封不动地传给w_list
# e_list中的数据表示该socket出错
# 最后一个参数为最多等待的时间，1秒
r_list, w_list, e_list = select.select(inputs, [], [], 1)
for sk in r_list:
# 每个连接对象
conn, address = sk.accept()
# 连接之后就发送一条信息
conn.sendall(bytes("Hello World!", encoding="utf-8"))

# 如果有连接出错了，就将其从列表中移除
for sk in e_list:
inputs.remove(sk)

import socket
# Client
client = socket.socket()
client.connect(("127.0.0.1", 8001))

ret = str(client.recv(1024), encoding="utf-8")
print(ret)

while True:
inp = input(">>> ")
if inp == "q":
client.send(bytes(inp, encoding="utf-8"))
break
else:
client.send(bytes(inp, encoding="utf-8"))
ret = str(client.recv(1024), encoding="utf-8")
print(ret)

client.close()

实现监听多用户连接的案例：

import socket
# Server
sk1 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sk1.bind(("127.0.0.1", 8001))
sk1.listen(5)

# sk2 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# sk2.bind(("127.0.0.1", 8002))
# sk2.listen(5)
#
# sk3 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# sk3.bind(("127.0.0.1", 8003))
# sk3.listen(5)

inputs = [sk1,]

import select
while True:
# select内部自动监听socket对象，一旦socket变换感知到
#
r_list, w_list, e_list = select.select(inputs, [], [], 1)

print("正在监听的socket对象%d" % len(inputs))
print(r_list)

for sk_or_conn in r_list:
# 每个连接对象
if sk_or_conn == sk1:
# 表示有新用户来连接
conn, address = sk_or_conn.accept()
inputs.append(conn)
else:
# 有老用户发消息了
try:
data_bytes = sk_or_conn.recv(1024)
data_str = str(data_bytes, encoding="utf-8") + "!"
sk_or_conn.sendall(bytes(data_str, encoding="utf-8"))
except Exception as ex:
# 如果发送错误，则从列表中移除
inputs.remove(sk_or_conn)

# 连接之后就发送一条信息
conn.sendall(bytes("Hello World!", encoding="utf-8"))

# 如果有连接出错了，就将其从列表中移除
for sk in e_list:
inputs.remove(sk)

import socket
# Client
client = socket.socket()
client.connect(("127.0.0.1", 8001))

ret = str(client.recv(1024), encoding="utf-8")
print(ret)

while True:
inp = input(">>> ")
if inp == "q":
client.send(bytes(inp, encoding="utf-8"))
break
else:
client.send(bytes(inp, encoding="utf-8"))
ret = str(client.recv(1024), encoding="utf-8")
print(ret)

client.close()

实现读写分离的Server案例：

import socket
# Server
sk1 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sk1.bind(("127.0.0.1", 8001))
sk1.listen(5)

# sk2 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# sk2.bind(("127.0.0.1", 8002))
# sk2.listen(5)
#
# sk3 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# sk3.bind(("127.0.0.1", 8003))
# sk3.listen(5)

inputs = [sk1,]
outputs = []

message_dict = {}

import select
while True:
# select内部自动监听socket对象，一旦socket变换即感知到
r_list, w_list, e_list = select.select(inputs, outputs, inputs, 1)

print("正在监听的socket对象%d" % len(inputs))
print(r_list)

# 监听并读取数据
for sk_or_conn in r_list:
# 每个连接对象
if sk_or_conn == sk1:
# 表示有新用户来连接
conn, address = sk_or_conn.accept()
inputs.append(conn)
message_dict[conn] = []
else:
# 有老用户发消息了
try:
data_bytes = sk_or_conn.recv(1024)
except Exception as ex:
# 如果发送错误，则从列表中移除
inputs.remove(sk_or_conn)
else:
# 用户正常发送消息
data_str = str(data_bytes, encoding="utf-8") + "!"
message_dict[sk_or_conn].append(data_str)
outputs.append(sk_or_conn)

# w_list仅仅保存了谁给我发过消息
# w_list=[]
for conn in w_list:
recv_str = message_dict[conn][0]
if recv_str:
conn.sendall(bytes(recv_str + "!", encoding="utf-8"))
del message_dict[conn][0]
outputs.remove(conn)

7. 线程与进程

多线程的简单案例：

import threading
import time

def process(arg):
time.sleep(1)
print(arg)

for i in range(10):
t= threading.Thread(target = process, args = (i,))
t.start()

1）进程与线程
（1）进程：不是越多越好，进程数=CPU个数

优点：同时利用多个CPU，能够同时进行多个操作
缺点：耗费资源（重新开辟内存空间）
可以通过设置daemon让进程等待子线程执行：p.daemon = True

主线程等待子线程执行：join(2) # 有参数的话为最多等待时间时间，此处为最多等待2秒

（2）线程：不是越多越好，具体案例具体分析，请求上下文切换耗时

优点：共享内存，IO操作时候，创造并发操作
缺点：抢占资源

设置主线程是否等待子线程：threading.Thread(target = func, args = (*args)).setDaemon(True)

主线程等待子线程执行：join(2) # 有参数的话为最多等待时间时间，此处为最多等待2秒

（3）进程和线程目的：提高执行效率

（4）IO密集型：多线程

（5）计算密集型：多进程

（6）GIL，全局解释器锁，用于管理多线程对CPU的使用

2）线程锁threading.RLock和threading.Lock

使用线程对数据操作的时候，如果多个线程同时修改某个数据，可能会出现不可预料的结果，为了保证数据的准确性，引入了锁的概念

案例：

import threading
import time

global_num = 0

lock = threading.RLock()

def func():
lock.acquire() # 获取锁
global global_num
global_num += 1
time.sleep(1)
print(global_num)
lock.release() # 释放锁

for i in range(10):
t = threading.Thread(target = func)
t.start()

3） threading.Event

Event是线程间通信的机制之一，一个线程发送一个event信号，其他的线程则等待这个信号，用于主线程控制其他线程的执行。Events管理一个flag，这个flag可以使用set()设置成True或使用clear()重置为False，wait()则用于阻塞，在flag为True之前。flag默认为False

Event.wait([timeout])：阻塞线程，直到Event对象内部标识位被设为True或超时（如果提供了参数timeout）
Event.set()：将标识位设为True
Event.clear()：将标识位设为False
Event.isSet()：判断标识位是否为True

import threading

def func(event):
print("start")
event.wait()
print("execute")

event = threading.Event()

for i in range(10):
t = threading.Thread(target = func, args = (event,))
t.start()

event.clear()

inp = input(">>>")

if inp.lower() == "true":
event.set()

4）queue模块

queue就是队列，它是线程安全的

q = queue.Queue(maxsize = 0) # 构造一个先进先出的队列，maxsize指定队列长度，为0时表示队列长度无限制

q.join() # 等到对列为空的时候，再执行别的操作

q.qsize() # 返回队列的大小（不可靠）

q.empty() # 当队列为空的时候，返回True，否则返回False（不可靠）

q.full() # 当队列满的时候，返回True，否则返回False（不可靠）

q.put(item, block = True, timeout = None) # 将item放入queue尾部，item必须存在。参数block默认为True，表示队列满时会等待队列给出可用位置，为False时为非阻塞，此时如果队列已满，会引发queue.Full异常，可选参数timeout表示阻塞设置的时间过后，如果队列无法给出放入item的位置，则引发queue.Full异常

q.get(block = True, timeout = None) # 移除并返回队列头部的一个值，可选参数block默认为True，表示获取值的时候如果队列为空则阻塞，为False时不阻塞，若此时队列为空，则引发queue.Empty异常，可选参数timeout表示会表示会阻塞的时间过后，如果队列为空，则引发Empty异常

q.put_nowait(item) # 等效于put(item, block = False)

q.get_nowait() # 等效于get(item, block = False)

生产者-消费者案例：

import queue
import threading

message = queue.Queue(10)

def producer(i):
print("Produce ", i)
message.put(i)

def consumer():
msg = message.get()
print(msg)

for i in range(10):
t = threading.Thread(target = producer, args = (i,))
t.start()

for i in range(10):
t = threading.Thread(target = consumer)
t.start()

5）multiprocessing模块

multiprocessing是python的多进程管理包，和threading.Thread类似，直接从侧面用subprocessess替换线程使用GIL的方式。由于这一点，multiprocessing模块可以让程序员在给定的机器上充分利用CPU。

在multiprocessing中，通过创建Process对象生成进程，然后调用它的start()方法

from multiprocessing import Process
def func(name):
print("Hello ", name)

if __name__ == "__main__":
p = Process(target = func, args = ("Steve", ))
p.start()
p.join()

6）进程间的数据共享

在使用并发设计时最好尽可能地避免共享数据，尤其是在使用多进程的时候。如若需要共享数据，multiprocessing提供了两种方式

（1）Shared memory

数据可以用Value或Array存储在一个共享内存地图里，案例如下：

from multiprocessing import Process, Value, Array

def func(n, a):
n.value = 3.1415927
for i in range(len(a)):
a[i] = -a[i]

if __name__ == "__main__":
num = Value('d', 0.0)
arr = Array('i', range(10))

p = Process(target = func, args = (num, arr))
p.start()
p.join()

print(num)
print(arr[:])

（2）Server process

由Manager()反悔的manager提供list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value 和Array类型的支持。

from multiprocessing import Process, Manager

def func(d, l):
d[1] = "1"
d["2"] = 2
d[0.25] = None
l.reverse()

if __name__ == "__main__":
with Manager() as m:
d = m.dict()
l = m.list(range(10))

p = Process(target = func, args = (d, l))
p.start()
p.join()

print(d)
print(l)

8. 进程池

Pool类描述了一个工作进程池，有几种不同的方法让任务卸载工作进程。

进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进程，那么程序就会等待，直到进程池中有可用进程为止。

可以用Pool类创建一个进程池，展开提交的任务给进程池。

简单使用进程池案例：

from multiprocessing import Pool
import time

def myFun(i):
time.sleep(2)
return i + 100

def end_call(arg):
print("end_call", arg)

if __name__ == "__main__":
p = Pool(5)
for i in range(10):
p.apply_async(func = myFun, args = (i,), callback= end_call)

print("end")
p.close()
p.join()

class multiprocessing.pool.Pool([processes[, initializer[m intargs[, maxtasksperchild[, context]]]]])

一个进程池对象可以控制工作进程池的哪些工作可以被提交，它支持超时和回调的异步结果，有一个类似map的实现。

processes：使用的工作进程的数量，如果processes是None那么使用os.cpu_count()返回的数量。
initializer：如果initializer是None，那么每一个工作进程在开始的时候会调用initializer(*initargs)。
maxtasksperchild：工作进程退出之前可以完成的任务数，完成后用一个新得工作进程来替代原进程，来让闲置的资源被释放。maxtasksperchild默认是None，意味着只要Pool存在工作进程就会一直存活。
context：用在指定工作进程启动时的上下文，一般使用multiprocessing.Pool()或者一个context对象的Pool()方法来创建一个池，两种方法都适当地设置了context

注意：Pool对象的方法只可以被创建pool的进程所调用

进程池的方法：

apply(func[, args[, kwds]])：使用args和kwds参数调用func函数，结果返回前会一直阻塞，由于这个原因，apply_async()更适合并发执行，另外func函数仅被pool中的一个进程运行。每个任务是排队进行的，而每个进程都执行join()方法
apply_async(func[, args[, kwds[, callback[, error_callback]]]])：apply()方法的一个变体，会返回一个结果对象。如果callback被指定，那么callback可以接受一个参数然后被调用，当结果准备好回调时会调用callback，调用失败时，则用error_callback替换callback。Callbacks应被立即完成，否则处理结果的线程会被阻塞。每个任务都并发进行，可以设置回调函数。每个进程没有执行join()方法，daemon参数默认为True。可以通过调用pool的join方法来等待所有子进程执行完毕，但在执行pool的该方法之前，必须执行pool的close()方法或terminate()方法。
close()：组织更多的任务提交到pool，待任务完成后，工作进程会退出。
terminate()：不管任务是否完成，立即停止工作进程。在对pool对象进程垃圾回收的时候，会立即调用terminate()。
join()：wait工作线程的退出，在调用join()前，必须调用close()或terminate()。这样是因为被终止的进程需要被父进程调用wait（join等价于wait），否则进程会成为僵尸进程。
map(func, iterable[, chunksize])
map_async(func, iterable[, chunksize[, callback[, error_callback]]])
imap(func, iterable,[, chunksize])
imap_unordered(func, iterable[, chunksize])
starmap(func, iterable[, chunksize])
starmap_async(func, iterable[, chunksize[, callback[, error_back]]])

9. 线程池

简单线程池的实现：

import queue
import threading
import time

class ThreadPool(object):

def __init__(self, max_num = 20):
# 创建队列并指定队列大小
self.queue = queue.Queue(max_num)

# 在队列中放入threading.Thread类名
for i in range(max_num):
self.queue.put(threading.Thread)

# 获取队列中的threading.Thread类名
def get_thread(self):
return self.queue.get()

# 往队列中添加threading.Thread类名
def add_thread(self):
self.queue.put(threading.Thread)

def func(pool, i):
time.sleep(1)
print(i)
pool.add_thread()

比较完善的线程池实现：

import queue
import threading
import time

stopEvent = object()

class ThreadPool(object):
def __init__(self, max_num):
# 队列为无限大，用以存储线程要执行的任务
self.q = queue.Queue()
# 最多创建的线程数，即线程池的最大容量
self.max_num = max_num
# 若terminal为False，则不停地去queue中取任务
self.terminal = False
# 真实创建的线程列表
self.generate_list = []
# 空闲线程列表
self.free_list = []

def run(self, func, args, callback = None):
"""
线程池执行一个任务
:param func: 任务函数
:param args: 任务函数所需参数
:param callback: 任务执行失败或成功后执行的回调函数，
:return:
"""
# 将需要执行的任务的方法、参数及回调函数打包成一个元组，传入任务队列
package = (func, args, callback,)
self.q.put(package)

# 若空闲线程数为0，且创建的线程总数小于规定的最大线程数，则创建一个线程
if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
self.generate_thread()

def generate_thread(self):
"""
创建一个线程，并执行call方法
:return:
"""
t = threading.Thread(target = self.call)
t.start()

def call(self):
"""
循环去获取任务函数并执行任务函数
:return:
"""
# 获取当前线程，并放入列表
current_thread = threading.currentThread
self.generate_list.append(current_thread)

# 取任务并执行
event = self.q.get()
while event != stopEvent:
# 元组 =》 任务
# 解开任务包
# 执行任务
func, args, callback = event

try:
# 返回调用方法的返回值
ret = func(args)
# 设置一个返回状态值，若正常执行结束，返回True
status = True
except Exception as e:
# 设置一个返回状态值，若执行出错，返回False
status = False
# 将所有错误的相关信息赋值给ret
ret = e

# 如果回调函数为非空，则执行回调函数
if callback is not None:
try:
callback(status, ret)
except Exception as e:
pass

# 如果terminal为False，则持续去取任务
if self.terminal: # terminal默认为False
event = stopEvent
else:
# 标记：当前空闲了，将当前线程放入空闲线程列表
self.free_list.append(current_thread)
# 执行完上一个任务，再次获取一个任务
event = self.q.get()
self.free_list.remove(current_thread)

else:
# 不是元组，不是任务
self.generate_list.remove(current_thread)

def close(self):
num = len(self.generate_list)

while num:
self.q.put(StopAsyncIteration)
num -= 1

# 终止线程（清空队列）
def terminate(self):
# 将terminal设置为True
self.terminal = True

# 在队列中放入stopEvent
while self.generate_list:
# 在队列中放入stopEvent，以清理线程
self.q.put(stopEvent)

self.q.empty()

def work(i):
time.sleep(1)
print(i)

# 执行ThreadPool的初始化方法
pool = ThreadPool(5)

for i in range(50):
# 将任务放入队列中
# 着手开始处理任务
#   - 创建线程
#       - 有空闲线程时，则不再创建线程
#       - 没有空闲线程时，不能高于线程池的限制
#       - 根据任务个数判断
#   - 线程去队列中去任务
result = pool.run(func = work, args = (i,))

pool.terminate()

10. 上下文管理器
上下文管理器在线程池中的实现：

import queue
import threading
import contextlib
import time

stopEvent = object()

class ThreadPool(object):
def __init__(self, max_num):
# 队列为无限大，用以存储线程要执行的任务
self.q = queue.Queue()
# 最多创建的线程数，即线程池的最大容量
self.max_num = max_num
# 若terminal为False，则不停地去queue中取任务
self.terminal = False
# 真实创建的线程列表
self.generate_list = []
# 空闲线程列表
self.free_list = []

def run(self, func, args, callback = None):
"""
线程池执行一个任务
:param func: 任务函数
:param args: 任务函数所需参数
:param callback: 任务执行失败或成功后执行的回调函数，
:return:
"""
# 将需要执行的任务的方法、参数及回调函数打包成一个元组，传入任务队列
package = (func, args, callback,)
self.q.put(package)

# 若空闲线程数为0，且创建的线程总数小于规定的最大线程数，则创建一个线程
if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
self.generate_thread()

def generate_thread(self):
"""
创建一个线程，并执行call方法
:return:
"""
t = threading.Thread(target = self.call)
t.start()

def call(self):
"""
循环去获取任务函数并执行任务函数
:return:
"""
# 获取当前线程，并放入列表
current_thread = threading.currentThread
self.generate_list.append(current_thread)

# 取任务并执行
event = self.q.get()
while event != stopEvent:
# 元组 =》 任务
# 解开任务包
# 执行任务
func, args, callback = event

try:
# 返回调用方法的返回值
ret = func(args)
# 设置一个返回状态值，若正常执行结束，返回True
status = True
except Exception as e:
# 设置一个返回状态值，若执行出错，返回False
status = False
# 将所有错误的相关信息赋值给ret
ret = e

# 如果回调函数为非空，则执行回调函数
if callback is not None:
try:
callback(status, ret)
except Exception as e:
pass

# 如果terminal为False，则持续去取任务
if self.terminal: # terminal默认为False
event = stopEvent
else:
with worker_state(self.free_list, current_thread):
event = self.q.get()

# # 标记：当前空闲了，将当前线程放入空闲线程列表
# self.free_list.append(current_thread)
# # 执行完上一个任务，再次获取一个任务
# event = self.q.get()
# self.free_list.remove(current_thread)

else:
# 不是元组，不是任务
self.generate_list.remove(current_thread)

def close(self):
num = len(self.generate_list)

while num:
self.q.put(StopAsyncIteration)
num -= 1

# 终止线程（清空队列）
def terminate(self):
# 将terminal设置为True
self.terminal = True

# 在队列中放入stopEvent
while self.generate_list:
# 在队列中放入stopEvent，以清理线程
self.q.put(stopEvent)

self.q.empty()

def work(i):
time.sleep(1)
print(i)

# 实现上下文管理的方法
@contextlib.contextmanager
def worker_state(self, state_list, worker_thread):
state_list.append(worker_thread)
try:
yield
finally:
state_list.remove(worker_thread)

# 执行ThreadPool的初始化方法
pool = ThreadPool(5)

for i in range(50):
# 将任务放入队列中
# 着手开始处理任务
#   - 创建线程
#       - 有空闲线程时，则不再创建线程
#       - 没有空闲线程时，不能高于线程池的限制
#       - 根据任务个数判断
#   - 线程去队列中去任务
result = pool.run(func = work, args = (i,))

pool.terminate()

1）用上下文管理器自定义实现open

import contextlib

@contextlib.contextmanager
def myopen(file_path, mode):
f = open(file_path, mode, encoding = "utf-8")
try:
yield f
finally:
f.close()

11. 协程
简单案例：

from greenlet import greenlet

def func1():
print("In Func1, Phase 1")
gr2.switch()
print("In Func1, Phase 2")
gr2.switch()

def func2():
print("In Func2, Phase 1")
gr1.switch()
print("In Func2, Phase 2")

gr1 = greenlet(func1)
gr2 = greenlet(func2)

gr1.switch()

import gevent

def func1():
print("Running in Func1")
# 一旦遇到gevent.sleep()便会到下一个协程去
gevent.sleep(0)
print("Explicit context switch to Func1 again")

def func2():
print("Explicit context to Func2")
gevent.sleep(0)
print("Implicit context switch back to Func2")

gevent.joinall([
gevent.spawn(func1),
gevent.spawn(func2),
])