socket
# 一、socket
# TCP服务端
import socket # 导入socket tcp_sk = socket.socket() # 实例化一个服务器对象
tcp_sk.bind(('127.0.0.1', 8080)) # 服务器绑定一个IP地址和端口
tcp_sk.listen() # 监听连接 # conn可以理解为在服务端中与客户端进行交互的操作符(对象)
# addr是客户端的IP地址和端口
conn, addr = tcp_sk.accept() # 接受客户端链接，此时如果没有客户端连接过来，服务器会在此等候，不会向下走(阻塞) while True:
se_msg = input('>>>:')
conn.send(se_msg.encode('utf-8')) # send发送的内容必须是bytes类型
re_msg = conn.recv(1024) # recv接收客户端发送过来的内容，1024是接收的字节长度
print(re_msg.decode('utf-8'))
conn.close() # 关闭与客户端的连接 tcp_sk.close() # 关闭服务器
# 注意：一般情况下服务器是不会关闭的，会一直为用户提供服务。 # 客户端
import socket sk = socket.socket() # 实例化一个客户端对象
sk.connect(('127.0.0.1', 8080)) # 连接服务端的IP和端口
while True:
re_msg = sk.recv(1024)
print(re_msg.decode('utf-8'))
se_msg = input('>>>:')
sk.send(se_msg.encode('utf-8'))
sk.close() # 关闭客户端 # upd服务端
import socket sk = socket.socket(type=socket.SOCK_DGRAM) # 设置type使其成为udp类型的socket
sk.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sk.bind(('127.0.0.1', 8005)) # 绑定IP和端口 msg, addr = sk.recvfrom(1024) # 接收信息和对方IP print(msg.decode('utf-8'))
print(addr) sk.sendto('你好'.encode('utf-8'), addr) # 发送信息给对方 sk.close() # client端：
import socket sk = socket.socket(type=socket.SOCK_DGRAM)
sk.sendto('hello'.encode('utf-8'), ('127.0.0.1', 8005))
msg, addr = sk.recvfrom(1024)
print(msg.decode('utf-8'))
print(addr)
sk.close() # 二、socket黏包
# 黏包现象只发生在tcp协议中：
# 1.从表面上看，黏包问题主要是因为发送方和接收方的缓存机制、tcp协议面向流通信的特点。
# 2.实际上，主要还是因为接收方不知道消息之间的界限，不知道一次性提取多少字节的数据所造成的 # 解决方案
# 首先介绍一个模块struct：该模块可以把一个类型，如数字，转成固定长度(4)的bytes
import struct ret1 = struct.pack('i', 10238976) # i代表把整型的数据转换成bytes类型的数据
ret2 = struct.pack('i', 1) print(ret1, len(ret1)) # b'\x00<\x9c\x00' 4
print(ret2, len(ret2)) # b'\x01\x00\x00\x00' 4
# 可以看到：数字10238976转成bytes后，长度为4，数字1转成bytes后，长度也是为4。 num1 = struct.unpack('i', ret1) # unpack把bytes类型转成第一个参数代表的类型(这里是i，也就是int 整型，但返回的是一个元组)
print(num1) # (10238976,) 元组
print(num1[0]) # 10238976 取元组的第一个值即可 # 服务端：
import socket
import struct sk = socket.socket()
sk.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sk.bind(('127.0.0.1', 8002))
sk.listen() conn, addr = sk.accept()
while True:
msg = input('>>>:').encode('utf-8') # 要发送的内容
pack_num = struct.pack('i', len(msg)) # 计算内容的长度
conn.send(pack_num)
conn.send(msg)
conn.close()
sk.close() # 客户端：
import socket
import struct sk = socket.socket()
sk.connect(('127.0.0.1', 8002)) while True:
pack_num = sk.recv(4)
num = struct.unpack('i', pack_num)[0]
ret = sk.recv(num)
print(ret.decode('utf-8'))
sk.close() # 三、socketserver
# TCP服务器(客户端跟之前原生的一样)
import socketserver # tcp协议的server端就不需要导入socket
class Myserver(socketserver.BaseRequestHandler): # 继承socketserver.BaseRequestHandler这个类
def handle(self): # 必须继承handle方法并重写
print(self.client_address) # 客户端的IP和端口: ('127.0.0.1', 64491)
print(type(self.request)) # <class 'socket.socket'> 与客户端连接的socket套接字
# <socket.socket fd=500, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 8888), raddr=('127.0.0.1', 64491)>
print(self.request) # 与客户端连接的套接字 conn = self.request # self.request就是客户端的对象
conn.send(b'helloworld')
print(conn.recv(1024).decode('utf-8')) # 设置allow_reuse_address允许服务器重用地址
socketserver.TCPServer.allow_reuse_address = True # 创建一个对象server,绑定ip和端口,相当于sk = socket.socket() sk.bind(('127.0.0.1',8888))这两步的结合
server = socketserver.ThreadingTCPServer(('127.0.0.1', 8888), Myserver) # 让server一直运行下去，除非强制停止程序
server.serve_forever() # 客户端
import socket sk = socket.socket()
sk.connect(('127.0.0.1', 8888)) msg = sk.recv(1024)
print(msg.decode('utf-8'))
sk.send(b'hiworld')
sk.close() # UDP服务器
import socketserver class Myserver(socketserver.BaseRequestHandler): # 继承socketserver.BaseRequestHandler这个类
def handle(self): # 必须继承handle方法并重写
print(self.client_address) # 客户端的IP和端口: ('127.0.0.1', 60575)
print(type(self.request)) # udp的时候，request是元组：<class 'tuple'>
print(self.request[0]) # 客户端的消息msg: b'dog'
print(self.request[1]) # udp套接字: <socket.socket fd=480, family=AddressFamily.AF_INET, type=SocketKind.SOCK_DGRAM, proto=0, laddr=('127.0.0.1', 8888)> sk = self.request[1]
sk.sendto(b'cat', self.client_address) # 设置allow_reuse_address允许服务器重用地址
socketserver.UDPServer.allow_reuse_address = True # 创建一个对象server,绑定ip和端口,相当于sk = socket.socket(type=socket.SOCK_DGRAM) sk.bind(('127.0.0.1',8888))这两步的结合
server = socketserver.ThreadingUDPServer(('127.0.0.1', 8888), Myserver) # 让server一直运行下去，除非强制停止程序
server.serve_forever() # 客户端
import socket sk = socket.socket(type=socket.SOCK_DGRAM) sk.sendto(b'dog', ('127.0.0.1', 8888))
msg, addr = sk.recvfrom(1024)
print(msg.decode('utf-8'))
sk.close()
进程
# 一、进程Process
# 在windows中使用需要注意
# 在Windows操作系统中由于没有fork(linux操作系统中创建进程的机制)，在创建子进程的时候会自动 import 启动它的这个文件，而在 import 的时候又执行了整个文件。因此如果将process()直接写在文件中就会无限递归创建子进程报错。
# 所以必须把创建子进程的部分使用if __name__ =='__main__' 判断保护起来，import 的时候 ，就不会递归运行了。
# join：父进程等待子进程结束后才继续执行自己后续的代码
import time
import random
from multiprocessing import Process def func(index):
time.sleep(random.randint(1, 3))
print('第%s封邮件发送完毕' % index) if __name__ == '__main__':
p_lst = []
for i in range(10):
p = Process(target=func, args=(i,))
p.start() # 先让所有子进程都启动
p_lst.append(p)
for p in p_lst: # 再进行join阻塞
p.join()
print('10封邮件全部发送完毕') # 守护进程
# 主进程创建守护进程
# 1：守护进程会在主进程代码执行结束后就终止
# 2：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children
# 注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止 # 1，守护进程会在主进程代码执行结束后就终止
import time
from multiprocessing import Process def func():
print('子进程 start')
time.sleep(3) # 睡3秒的时候主进程的代码已经执行完毕了，所以子进程也会跟着结束
print('子进程end') if __name__ == '__main__':
p = Process(target=func)
p.daemon = True # daemon是Process的属性
p.start()
time.sleep(2) # 睡2秒的时候，执行了子进程
print('主进程')
# 结果：
# 子进程 start
# 主进程 # 锁
# 加锁降低了程序的效率，让原来能够同时执行的代码变成顺序执行了，异步变同步的过程
# 保证了数据的安全 import time
import json
from multiprocessing import Process
from multiprocessing import Lock # 导入Lock类 def search(person):
with open('ticket') as f:
ticketinfo = json.load(f)
print('%s查询余票：' % person, ticketinfo['count']) def get_ticket(person):
with open('ticket') as f:
ticketinfo = json.load(f)
time.sleep(0.2) # 模拟读数据的网络延迟
if ticketinfo['count'] > 0:
print('%s买到票了' % person)
ticketinfo['count'] -= 1
time.sleep(0.2)
with open('ticket', 'w') as f:
json.dump(ticketinfo, f)
else:
print('%s没买到票' % person) def ticket(person, lock):
search(person)
lock.acquire() # 谁获得钥匙 谁才能进入
get_ticket(person)
lock.release() # 用完了，把钥匙给下一个人 if __name__ == '__main__':
lock = Lock() # 创建一个锁对象
for i in range(5):
p = Process(target=ticket, args=('person%s' % i, lock))
p.start() # 结果：
# person1查询余票： 3
# person3查询余票： 3
# person0查询余票： 3
# person2查询余票： 3
# person4查询余票： 3
# person1买到票了
# person3买到票了
# person0买到票了
# person2没买到票
# person4没买到票 # 1、信号量的实现机制：计数器 + 锁实现的
# 信号量同步基于内部计数器，每调用一次acquire()，计数器减1；每调用一次release()，计数器加1.当计数器为0时，acquire()
# 调用被阻塞。
# 互斥锁同时只允许一个线程更改数据，而信号量Semaphore是同时允许一定数量的线程更改数据(Samphore相当于有几把钥匙，lock只能有一把钥匙) import time
import random
from multiprocessing import Process
from multiprocessing import Semaphore def changba(person, sem): # 在唱吧 唱歌
sem.acquire() # 第一次可以同时进来两个人
print('%s走进唱吧' % person)
time.sleep(random.randint(3, 6)) # 每个人唱歌的时间
print('%s走出唱吧' % person) # 唱完走人
sem.release() # 把钥匙给下一个人 if __name__ == '__main__':
sem = Semaphore(2) # 2把钥匙
for i in range(5):
p = Process(target=changba, args=('person%s' % i, sem))
p.start() # 事件 Event
# 事件主要提供了三个方法
# set、wait、clear。
# 事件处理的机制：全局定义了一个“Flag”，
# 如果“Flag”值为False，那么当程序执行event.wait方法时就会阻塞，
# 如果“Flag”值为True，那么event.wait方法时便不再阻塞。 # 阻塞事件 ：wait()方法
# wait是否阻塞是看event对象内部的Flag # 控制Flag的值:
# set() 将Flag的值改成True
# clear() 将Flag的值改成False
# is_set() 判断当前的Flag的值 # 红绿灯：
import time
import random
from multiprocessing import Process
from multiprocessing import Event def traffic_ligth(e): # 红绿灯
print('\033[31m红灯亮\033[0m') # Flag 默认是False
while True:
if e.is_set(): # 如果是绿灯
time.sleep(2) # 2秒后
print('\033[31m红灯亮\033[0m') # 转为红灯
e.clear() # 设置为False else: # 如果是红灯
time.sleep(2) # 2秒后
print('\033[32m绿灯亮\033[0m') # 转为绿灯
e.set() # 设置为True def car(e, i): # 车
if not e.is_set():
print('car %s在等待' % i)
e.wait()
print('car %s 通过了' % i) if __name__ == '__main__':
e = Event()
p = Process(target=traffic_ligth, args=(e,)) # 红绿灯进程
p.daemon = True
p.start()
p_lst = []
for i in range(10): # 10辆车的进程
time.sleep(random.randrange(0, 3, 2))
p = Process(target=car, args=(e, i))
p.start()
p_lst.append(p)
for p in p_lst: p.join() # 二、进程通信
# 队列(先进先出)
# 生产者消费者模型
import time
import random
from multiprocessing import Process, Queue def consumer(q, name):
# 消费者
while True:
food = q.get() # 在队列中取值
if food is None: break
time.sleep(random.uniform(0.3, 1)) # 模拟吃消耗的时间
print('%s偷吃了%s，快打死他' % (name, food)) def producter(q, name, food):
# 生产者
for i in range(10):
time.sleep(random.uniform(0.5, 0.9)) # 模拟生产时间
print('%s生产了%s，序号：%s' % (name, food, i))
q.put(food + str(i)) # 把值存入队列中 if __name__ == '__main__':
q = Queue() # Queue队列对象
c1 = Process(target=consumer, args=(q, '小明'))
c2 = Process(target=consumer, args=(q, '小东'))
c1.start()
c2.start() p1 = Process(target=producter, args=(q, '张三', '面包'))
p2 = Process(target=producter, args=(q, '李四', '可乐'))
p1.start()
p2.start() p1.join()
p2.join() q.put(None) # 有几个consumer进程就需要放几个None,表示生产完毕(这就有点low了)
q.put(None) # JoinableQueue
# JoinableQueue和Queue几乎一样，不同的是JoinableQueue队列允许使用者告诉队列某个数据已经处理了。通知进程是使用共享的信号和条件变量来实现的。
# task_done()：使用者使用此方法发出信号，表示q.get()返回的项目已经被处理
# join()：当队列中有数据的时候，使用此方法会进入阻塞，直到放入队列中所有的数据都被处理掉(都被task_done)才转换成不阻塞 # 解决刚才生产者消费者模型low的问题：
import time
import random
from multiprocessing import Process, JoinableQueue def consumer(jq, name):
# 消费者
while True:
food = jq.get() # 在队列中取值
# if food is None:break
time.sleep(random.uniform(0.3, 1)) # 模拟吃消耗的时间
print('%s偷吃了%s，快打死他' % (name, food))
jq.task_done() # 向jq.join()发送一次信号,证明这个数据已经处理了 def producter(jq, name, food):
# 生产者
for i in range(10):
time.sleep(random.uniform(0.5, 0.9)) # 模拟生产时间
print('%s生产了%s，序号：%s' % (name, food, i))
jq.put(food + str(i)) # 把值存入队列中 if __name__ == '__main__':
jq = JoinableQueue()
c1 = Process(target=consumer, args=(jq, '小明'))
c2 = Process(target=consumer, args=(jq, '小东')) c1.daemon = True # 把消费者设置为守护进程
c2.daemon = True c1.start()
c2.start() p1 = Process(target=producter, args=(jq, '张三', '面包'))
p2 = Process(target=producter, args=(jq, '李四', '可乐'))
p1.start()
p2.start() p1.join()
p2.join() jq.join() # 数据全部被task_done后才不阻塞 # 管道
# Pipe([duplex]):在进程之间创建一条管道，并返回元组（left,right）,其中left,right表示管道两端的连接对象，强调一点：必须在产生Process对象之前产生管道
# duplex:默认管道是全双工的，如果将duplex改成False，left只能用于接收，right只能用于发送。
# 主要方法：
# right.recv(): 接收left.send()发送的内容。如果没有消息可接收，recv方法会一直阻塞。如果连接的另外一端已经关闭，那么recv方法会抛出EOFError。
# letf.send(): 通过连接发送内容。
# close(): 关闭连接。 # pipe的端口管理不会随着某一个进程的关闭就关闭
# 操作系统来管理进程对这些端口的使用，不使用的端口应该关闭它
# 一条管道，两个进程，就有4个端口 每关闭一个端口计数-1，直到只剩下一个端口的时候 recv就会报错
# 如果不关闭不使用的端口，在已经把数据发送完毕的情况下，那么接收端的recv就会一直挂起，等待接收数据，这个进程就一直不能关闭
# 因此不使用的端口就应该关闭它，让recv抛出异常后对这个进程进行处理
from multiprocessing import Process, Pipe def consumer(left, right):
left.close() # 若这里不close，则不会异常EOFError，数据接收完毕后，下面的right.recv()就会一直挂起
while True:
try:
print(right.recv())
except EOFError:
break if __name__ == '__main__':
left, right = Pipe()
Process(target=consumer, args=(left, right)).start()
right.close()
for i in range(10):
left.send('Apple%s' % i)
left.close() # 三、进程池
# 同步、apply
import os
import time
from multiprocessing import Pool def test(num):
time.sleep(1)
print('%s:%s' % (num, os.getpid()))
return num * 2 if __name__ == '__main__':
p = Pool()
for i in range(20):
res = p.apply(test, args=(i,)) # 提交任务的方法 同步提交
print('-->', res) # res就是test的return的值 # 异步、apply_async
import time
from multiprocessing import Pool def func(num):
time.sleep(1)
print('做了%s件衣服' % num) if __name__ == '__main__':
p = Pool(4) # 进程池中创建4个进程，不写的话，默认值为你电脑的CUP数量
for i in range(50):
p.apply_async(func, args=(i,)) # 异步提交func到一个子进程中执行，没有返回值的情况
p.close() # 关闭进程池，用户不能再向这个池中提交任务了
p.join() # 阻塞，直到进程池中所有的任务都被执行完 # 注意：
# 异步提交且没有返回值接收的情况下必须要用close()和join()
# 因为如果没有close()和join()，主进程执行完毕后会立刻把子进程回收了，相当于子进程还没来得及开启
# 所以要join，让子进程结束后再结束父进程，但是进程池中要使用join就必须先进行close import time
import os
from multiprocessing import Pool def test(num):
time.sleep(1)
print('%s:%s' % (num, os.getpid()))
return num * 2 if __name__ == '__main__':
p = Pool()
res_lst = []
for i in range(20):
res = p.apply_async(test, args=(i,)) # 提交任务的方法 异步提交
res_lst.append(res)
for res in res_lst:
print(res.get()) # 注意：
# 异步提交有返回值的情况下，res是一个对象代表的是这个任务的编号，需要用res.get()方法让任务执行且把返回值返回给你。
# get有阻塞效果，拿到子进程的返回值后才不阻塞，所以并不需要再使用close和join。 # map
# map接收一个函数和一个可迭代对象，是异步提交的简化版本，自带close和join方法
# 可迭代对象的每一个值就是函数接收的实参，可迭代对象的长度就是创建的任务数量
# map拿到返回值是所有结果组成的列表
import time
from multiprocessing import Pool def func(num):
print('子进程：', num)
# time.sleep(1)
return num if __name__ == '__main__':
p = Pool()
ret = p.map(func, range(10)) # ret是列表
for i in ret:
print('返回值：', i) # 进程池的回调函数(同步提交apply没有回调函数)
import os
from multiprocessing import Pool def func(i):
print('子进程：', os.getpid())
return i def call_back(res):
print('回调函数：', os.getpid())
print('res--->', res) if __name__ == '__main__':
p = Pool()
print('主进程：', os.getpid())
p.apply_async(func, args=(1,), callback=call_back) # callback关键字传参，参数是回调函数
p.close()
p.join() # 结果：
# 主进程： 4732
# 子进程： 10552
# 回调函数： 4732
# res---> 1
#
# 从结果可以看出：
# 子进程func执行完毕之后才去执行callback回调函数
# 子进程func的返回值会作为回调函数的参数
# 回调函数是在主进程中执行的
线程
# multiprocess模块完全模仿了threading模块的接口，二者在使用层面，有很大的相似性
# 一、线程
# 方式一
from threading import Thread
import time def sleep_boy(name):
time.sleep(1)
print('%s is sleeping' % name) t = Thread(target=sleep_boy, args=('xiaoming',)) # 这里可以不需要main，因为现在只是在一个进程内操作，不需要导入进程就不会import主进程了
t.start()
t.join()
print('主线程') # 方式二
from threading import Thread
import time class Sleep_boy(Thread):
def __init__(self, name):
super().__init__()
self.name = name def run(self):
time.sleep(1)
print('%s is sleeping' % self.name) t = Sleep_boy('xiaoming')
t.start()
print('主线程') # 二、守护线程
# 守护进程和守护线程的区别：
# 主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，才会结束
# 主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，所以主线程结束了之后，守护线程随着主进程的结束自然结束了
# 守护线程(三种方式)
# 1
t1 = Thread(target=func1, daemon=True) # 2
t1 = Thread(target=func1)
t1.daemon = True # 3
t1 = Thread(target=func1)
t1.setDaemon(True) # 示例
import time
from threading import Thread def func1():
while True:
time.sleep(0.5)
print('func1') def func2():
print('func2 start')
time.sleep(3)
print('func2 end') t1 = Thread(target=func1)
t2 = Thread(target=func2)
t1.setDaemon(True) # 设置守护线程
# t1.daemon = True t1 = Thread(target=func1,daemon = True) 这两种方式也是开启守护线程的方法
t1.start()
t2.start()
print('主线程代码结束了') # 三、死锁、信号量、事件、条件、定时器
# 所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法运行下去。
# 此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程。即：操作的时候，抢到一把锁之后还要再去抢第二把锁，
# 但是由于一个线程抢到一把锁，另一个线程抢到了另一把锁，所以导致死锁。
# 例如：(互斥锁才会出现死锁)
import time
from threading import Thread, Lock noodle_lock = Lock()
fork_lock = Lock() def eat1(name):
noodle_lock.acquire()
print('%s拿到面条了' % name) fork_lock.acquire()
print('%s拿到叉子了' % name) print('%s吃面' % name)
time.sleep(0.3) fork_lock.release()
print('%s放下叉子' % name) noodle_lock.release()
print('%s放下面' % name) def eat2(name):
fork_lock.acquire()
print('%s拿到叉子了' % name) noodle_lock.acquire()
print('%s拿到面条了' % name) print('%s吃面' % name)
time.sleep(0.3) noodle_lock.release()
print('%s放下面' % name) fork_lock.release()
print('%s放下叉子' % name) name_list = ['小明', '小红']
name_list2 = ['小白', '小黑']
for name in name_list:
Thread(target=eat1, args=(name,)).start()
for name in name_list2:
Thread(target=eat2, args=(name,)).start() # 解决方法，递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。
# 这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。
# 上面的例子如果使用RLock代替Lock，则不会发生死锁： import time
from threading import Thread, RLock fork_lock = noodle_lock = RLock() def eat1(name):
noodle_lock.acquire()
print('%s拿到面条了' % name) fork_lock.acquire()
print('%s拿到叉子了' % name) print('%s吃面' % name)
time.sleep(0.3) fork_lock.release()
print('%s放下叉子' % name) noodle_lock.release()
print('%s放下面' % name) def eat2(name):
fork_lock.acquire()
print('%s拿到叉子了' % name) noodle_lock.acquire()
print('%s拿到面条了' % name) print('%s吃面' % name)
time.sleep(0.3) noodle_lock.release()
print('%s放下面' % name) fork_lock.release()
print('%s放下叉子' % name) name_list = ['小明', '小红']
name_list2 = ['小白', '小黑']
for name in name_list:
Thread(target=eat1, args=(name,)).start()
for name in name_list2:
Thread(target=eat2, args=(name,)).start() # 信号量
# 跟进程的一样
# Semaphore管理一个内置的计数器，
# 每当调用acquire()时内置计数器-1；
# 调用release() 时内置计数器+1；
# 计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。 import time
from threading import Semaphore, Thread def func(index, sem):
sem.acquire()
print(index)
time.sleep(1)
sem.release() sem = Semaphore(5)
for i in range(10):
Thread(target=func, args=(i, sem)).start() # 事件
# 跟进程的一样
# 线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。
# 为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在初始情况下,Event对象中的信号标志被设置为False。
# 如果有线程等待一个Event对象, 而这个Event对象的标志为False,那么这个线程将会被一直阻塞直至该标志为True。一个线程如果将一个Event对象的信号标志设置为True,它将唤醒所有等待这个Event对象的线程。
# 如果一个线程等待一个已经被设置为True的Event对象,那么它将忽略这个事件, 继续执行。 # event.wait() 信号为False的时候就阻塞，直到事件内的信号变成True才不阻塞
# event.set() 把信号变成True
# event.clear() 把信号变成False
# event.is_set() 查看信号状态是否为True # 例如：检测数据库连接
import time
import random
from threading import Thread, Event def check(e):
print('开始检测数据库连接')
time.sleep(random.randint(1, 3)) # 检测数据库连接(这里只是模拟，应该是真正检测的)
e.set() # 成功了 def connect(e):
for i in range(3): # 尝试三次去连接数据库
e.wait(0.5) # 超时就向下继续执行
if e.is_set(): # 如果True才显示连接成功
print('数据库连接成功')
break
else:
print('尝试连接数据库%s次失败' % (i + 1))
else:
raise TimeoutError e = Event()
Thread(target=connect, args=(e,)).start()
Thread(target=check, args=(e,)).start() # 条件
# Python提供的Condition对象提供了对复杂线程同步问题的支持。Condition被称为条件变量，除了提供与Lock类似的acquire和release方法外，还提供了wait和notify方法。线程首先acquire一个条件变量，然后判断一些条件。
# 如果条件不满足则wait；如果条件满足，进行一些处理改变条件后，通过notify方法通知其他线程，其他处于wait状态的线程接到通知后会重新判断条件。不断的重复这一过程，从而解决复杂的同步问题。 # 就像排队进火车站一样
# notify 控制流量 通知有多少人可以通过了
# wait 在门口等待的所有人 # acquire
# wait 使用前后都需要加锁
# 做的事情
# release # acquire
# notify 使用前后都需要加锁
# release # 例子：
from threading import Thread, Condition def func(con, index):
print('%s在等待' % index)
con.acquire()
con.wait()
print('%s do something' % index)
con.release() con = Condition()
for i in range(10):
t = Thread(target=func, args=(con, i))
t.start() # con.acquire()
# con.notify_all() # 让所有的线程通过
# con.release() count = 10
while count > 0:
num = int(input('>>>:'))
con.acquire()
con.notify(num) # 选择让多少个线程通过
count -= num
con.release() # 定时器
# 定时器，指定n秒后执行某个操作 from threading import Timer def func():
print('执行我啦') t = Timer(5, func) # 起一个定时器任务，设置多少秒后执行
t.start()
print('主线程')
池
# concurrent.futures模块的进程池线程池 # 一、介绍
# concurrent.futures模块提供了高度封装的异步调用接口
# ThreadPoolExecutor：线程池，提供异步调用
# ProcessPoolExecutor: 进程池，提供异步调用
# 在这个模块中进程池和线程池的使用方法完全一样 # 这里就只介绍ThreadPoolExecutor的使用方法，顺便对比multiprocessing的Pool进程池 # 基本方法
# submit(fn, *args, **kwargs)：异步提交任务 # map(func, *iterables, timeout=None, chunksize=1) ：取代for循环submit的操作，iterables的每个元素都作为参数传给func # shutdown(wait=True) ：
# 相当于进程池的pool.close()+pool.join()操作
# wait=True，等待池内所有任务执行完毕回收完资源后才继续
# wait=False，立即返回，并不会等待池内的任务执行完毕
# 但不管wait参数为何值，整个程序都会等到所有任务执行完毕
# submit和map必须在shutdown之前 # result(timeout=None)：取得返回值 # add_done_callback(fn)：设置回调函数 # 二、例子
# 例1
import time
from concurrent.futures import ThreadPoolExecutor def func(i):
print('thread', i)
time.sleep(1)
print('thread %s end' % i) tp = ThreadPoolExecutor(5) # 相当于tp = Pool(5)
tp.submit(func, 1) # 相当于tp.apply_async(func,args=(1,))
tp.shutdown() # 相当于tp.close() + tp.join()
print('主线程') # 例2
import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread def func(i):
print('thread', i, currentThread().ident)
time.sleep(1)
print('thread %s end' % i) tp = ThreadPoolExecutor(5)
for i in range(20):
tp.submit(func, i)
tp.shutdown() # shutdown一次就够了，会自动把所有的线程都join()
print('主线程') # 例3 返回值
import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread def func(i):
print('thread', i, currentThread().ident)
time.sleep(1)
print('thread %s end' % i)
return i * '*' tp = ThreadPoolExecutor(5)
ret_lst = []
for i in range(20):
ret = tp.submit(func, i)
ret_lst.append(ret) for ret in ret_lst:
print(ret.result()) # 相当于ret.get() print('主线程') # 三、map
# map接收一个函数和一个可迭代对象
# 可迭代对象的每一个值就是函数接收的实参，可迭代对象的长度就是创建的线程数量
# map拿到的返回值是所有结果组成的迭代器(跟进程池Pool的map一样，就是返回值的类型不同，Pool的map拿到的返回值是列表) import time
from concurrent.futures import ThreadPoolExecutor def func(i):
print('thread', i)
time.sleep(1)
print('thread %s end' % i)
return i * '*' tp = ThreadPoolExecutor(5)
ret = tp.map(func, range(20))
for i in ret:
print(i) # 四、回调函数
# 回调函数在进程池是由主进程实现的
# 回调函数在线程池是由子线程实现的
import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread def func(i):
print('thread', i, currentThread().ident)
time.sleep(1)
print('thread %s end' % i)
return i def call_back(arg):
print('call back : ', currentThread().ident)
print('ret : ', arg.result()) # multiprocessing的Pool回调函数中的参数不需要get(),这里的参数需要result()才能取出值 tp = ThreadPoolExecutor(5)
ret_lst = []
for i in range(20):
tp.submit(func, i).add_done_callback(call_back) # 使用add_done_callback()方法实现回调函数
tp.shutdown()
print('主线程', currentThread().ident)
# 从结果可以看出：
# 子线程func执行完毕之后才去执行callback回调函数
# 子线程func的返回值会作为回调函数的参数
协程
# 一、介绍
# 进程 ：计算机中最小的资源分配单位 # 线程 ：计算机中能被cpu执行的最小单位 # 协程(纤程)：一条线程在多个任务之间来回切换就叫协程
# 切换这个动作是浪费时间的
# 对于CPU、操作系统来说协程是不存在的
# 他们只能看到线程 # 协程的本质就是一条线程在多个任务之间来回切换，所以完全不会产生数据安全的问题 # 二、greenlet
import time
from greenlet import greenlet def cooking():
print('cooking 1')
g2.switch() # 切换到g2，让g2的函数工作
time.sleep(1)
print('cooking 2') def watch():
print('watch TV 1')
time.sleep(1)
print('watch TV 2')
g1.switch() # 切换到g1，让g1的函数工作 g1 = greenlet(cooking)
g2 = greenlet(watch)
g1.switch() # 启动g1，让g1的函数工作 ######## 传参 ########
import time
from greenlet import greenlet def cooking(name):
print('cooking ', name)
g2.switch("cat") # 切换到g2，让g2的函数工作
time.sleep(1)
print('cooking 2') def watch(name):
print('watch TV', name)
time.sleep(1)
print('watch TV 2')
g1.switch() # 切换到g1，让g1的函数工作 g1 = greenlet(cooking)
g2 = greenlet(watch)
g1.switch(name="dog") # 切换到g1，让g1的函数工作 # greenlet的缺陷：很显然greenlet实现了协程的切换功能，可以自己设置什么时候切，在哪切，但是它遇到阻塞并没有自动切换，
# 因此并不能提高效率。所以一般我们都使用gevent模块实现协程 # 三、gevent
# 无返回值
from gevent import monkey monkey.patch_all()
import time
import gevent def cooking():
print('cooking 1')
time.sleep(1)
print('cooking 2') def watch():
print('watch TV 1')
time.sleep(1)
print('watch TV 2') g1 = gevent.spawn(cooking) # 自动检测阻塞事件，遇见阻塞了就会进行切换
g2 = gevent.spawn(watch) g1.join() # 阻塞直到g1结束
g2.join() # 阻塞直到g2结束 # 有返回值
import gevent def cooking(i):
print('%s号在煮饭' % i)
return i g_lst = []
for i in range(10):
g = gevent.spawn(cooking, i) # 函数名，参数
g_lst.append(g) # 把协程对象放入列表 for g in g_lst:
g.join()
print(g.value) # 打印返回值 # gevent.joinall(g_lst) # joinall一次性把全部对象都阻塞 # spawn_raw
import gevent def cooking(i):
print('%s号在煮饭' % i) for i in range(10):
gevent.spawn_raw(cooking, i) # 函数名，参数
gevent.wait() # 让主协程阻塞，自动切换到cooking协程 # 四、eventlet
# spawn
import eventlet def func(name):
print(name)
return ("I am %s" % name) e1 = eventlet.spawn(func, 'dog')
e2 = eventlet.spawn(func, 'cat')
print(e1.wait())
print(e2.wait()) # 使用wait获取这个协程的返回值
# 结果
# dog
# cat
# I am dog
# I am cat # spawn_n
import eventlet def func(name):
print(name)
return ("I am %s" % name) # 实际不会返回这个值 e1 = eventlet.spawn_n(func, "dog")
e2 = eventlet.spawn_n(func, "cat")
eventlet.sleep() # 让主协程阻塞，才会切换到其他协程
# 或者主动执行
# e1 = eventlet.spawn_n(func)
# e2 = eventlet.spawn_n(func)
# e1.run("dog")
# e2.run("cat") # 结果
# dog
# cat # spawn_after
import eventlet def func(name):
print(name)
return ("I am %s" % name) e1 = eventlet.spawn_after(2, func, 'dog')
e2 = eventlet.spawn_after(1, func, 'cat')
print(e1.wait())
print(e2.wait()) # 使用wait获取这个协程的返回值
# 结果
# cat
# dog
# I am dog
# I am cat # 五、eventlet.greenpool
eventlet.GreenPool = eventlet.greenpool.GreenPool
eventlet.GreenPile = eventlet.greenpool.GreenPile # spawn
from eventlet.greenpool import GreenPool def worker(work):
return work pool = GreenPool() # 实例化一个协程池
res1 = pool.spawn(worker, "打杂")
res2 = pool.spawn(worker, "打狗")
print(res1.wait())
print(res2.wait())
# 结果
# 打杂
# 打狗 # spawn_n
"""
spawn并不是去开了一个线程，而是greenthread
Spawn只是将这个方法放入到了需要调度的一个pool里面，只有调用wait了，eventlet才会用自己的调度器去对刚刚放入pool的协程（greenthread）进行调度执行 spawn_n 则需要waitall() GreenPool中创建的线程是绿色线程.这意味着它们都存在于操作系统级别的一个线程中,
并且Python解释器处理它们之间的切换.仅当一个线程产生(故意为其他线程提供运行机会)或正在等待I/O时,才会发生这种切换.
除了wait和waitall，还能主动sleep进行切换
import eventlet
eventlet.sleep()
"""
from eventlet.greenpool import GreenPool def worker(work):
print(work) # 没有返回值 pool = GreenPool() # 实例化一个协程池
pool.spawn_n(worker, "打杂")
pool.spawn_n(worker, "打狗")
pool.waitall() # 或者 eventlet.sleep()
# 结果
# 打狗
# 打杂 # imap
from eventlet.greenpool import GreenPool def worker(work):
print(work)
return work pool = GreenPool() # 实例化一个协程池
work_list = ["打杂", "打狗"]
res = pool.imap(worker, work_list)
for result in res:
print(result) # 结果
# 打杂
# 打狗