什么是线程？
线程是CPU上的执行单位。
线程和进程的区别
1、进程是资源的集合，是一个资源单位。线程是CPU上是执行单位。所以开进程开销与远大于开线程
2、进程单独开辟内存空间。同一个进程内多线程共享同一个内存空间
a = 100
def task():
global a
a = 0
if __name__ == '__main__':
# p = Process(target=task, )
# p.start()
# p.join()
# print(a) # 100
t = Thread(target=task, )
t.start()
t.join()
print(a) # 0
3、开多个进程，每个进程有不同的pid。在主进程下开启多个线程，每个线程的pid和主进程的pid一样
Thread对象的其他用法
from threading import Thread, current_thread, active_count, enumerate
import time
def task():
print("子进程", current_thread().getName(), current_thread().name)
# current_thread().getName() == current_thread().name
time.sleep(2)
if __name__ == '__main__':
t = Thread(target=task, name="sb")
t.start()
print("子线程", t.getName())
t.setName("我是子线程") # 设置进程名
print("子线程", t.name) # t.name == t.getName() 查看进程名
# t.join()
print(t.is_alive()) # 查看进程是否活动
# t.is_alive() == t.isAlive
print(active_count()) # 返回正在运行的线程数量，
print(len(enumerate()))
# 与len(threading.enumerate())有相同的结果。
练习
from threading import Thread
import time
def foo():
print("123")
time.sleep(1)
print("end123")
def bar():
print(456)
time.sleep(3)
print("end 456")
if __name__ == '__main__':
t1 = Thread(target=foo,)
t1.daemon = True
t2 = Thread(target=bar)
t1.start()
t2.start()
print("zhu")
"""
123
456
zhu
end123
end 456
"""
线程互斥锁
不加锁
from threading import Thread
import time
n = 100
def task():
global n
temp = n
time.sleep(0.1)
n = temp - 1
if __name__ == '__main__':
t_l = []
for i in range(10):
t = Thread(target=task)
t_l.append(t)
t.start()
for t in t_l:
t.join()
print("zhu", n) # 99
加了锁
from threading import Thread, Lock
import time
n = 100
def task():
global n
mutex.acquire()
temp = n
time.sleep(0.1)
n = temp - 1
mutex.release()
if __name__ == '__main__':
mutex = Lock()
t_l = []
for i in range(10):
t = Thread(target=task)
t_l.append(t)
t.start()
for t in t_l:
t.join()
print("zhu", n) # 90
GIL全局解释器锁
注意:
GIL并不是Python的特性，Python完全可以不依赖于GIL。它是在实现Python解析器(CPython)时所引入的一个概念。
什么是GIL?
全局解释器锁，GIL本质就是一把互斥锁。解释器级别的一把互斥锁。
GIL和互斥锁的区别
GIL 与Lock是两把锁，保护的数据不一样，前者是解释器级别的（当然保护的就是解释器级别的数据，比如垃圾回收的数据），后者是保护用户自己开发的应用程序的数据。
验证python test.py只会产生一个进程
#test.py内容
import os,time
print(os.getpid())
time.sleep(1000)
#打开终端执行
python3 test.py
#在windows下查看
tasklist |findstr python
#在linux下下查看
ps aux |grep python
https://www.luffycity.com/python-book/73-duo-jin-cheng-shi-xian/746-gilquan-ju-jie-shi-qi-suo.html
信号量
互斥锁 同时只允许一个线程更改数据，而信号量（Semaphore）是同时允许一定数量的线程更改数据 任务拿到锁去执行。
from threading import Thread,Semaphore
import threading
import time
def func():
sm.acquire()
print('%s get sm' %threading.current_thread().getName())
time.sleep(3)
sm.release()
if __name__ == '__main__':
sm=Semaphore(5)
for i in range(23):
t=Thread(target=func)
t.start()
Event事件
对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。
from threading import Event
event.is_set()：返回event的状态值；
event.wait()：如果 event.isSet()==False将阻塞线程；
event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
event.clear()：恢复event的状态值为False。
代码实例
from threading import Thread,Event
import threading
import time,random
def conn_mysql():
count=1
while not event.is_set():
if count > 3:
raise TimeoutError('链接超时')
print('<%s>第%s次尝试链接' % (threading.current_thread().getName(), count))
event.wait(0.5)
count += 1
print('<%s>链接成功' % threading.current_thread().getName())
def check_mysql():
print(' 33[45m[%s]正在检查mysql 33[0m' % threading.current_thread().getName())
time.sleep(random.randint(2, 4))
event.set()
if __name__ == '__main__':
event=Event()
conn1=Thread(target=conn_mysql)
conn2=Thread(target=conn_mysql)
check=Thread(target=check_mysql)
conn1.start()
conn2.start()
check.start()
线程queue
定时器
from threading import Timer
def hello():
print("hello, world")
t = Timer(1, hello)
t.start()
多线程FTP
服务端
from threading import Thread
import socket
def communicate(conn):
while True:
try:
data = conn.recv(1024).decode()
if not data:
break
print(data)
conn.send(data.upper().encode())
except ConnectionRefusedError as e:
print(e)
break
def server(ip, port):
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind((ip, port))
server.listen(5)
while True:
conn, addr = server.accept()
t = Thread(target=communicate, args=(conn,))
t.start()
server.close()
if __name__ == '__main__':
server('127.0.0.1', 9999)
线程池&进程池
官网：https://docs.python.org/dev/library/concurrent.futures.html
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor：线程池，提供异步调用
ProcessPoolExecutor: 进程池，提供异步调用
基本方法
、submit(fn, *args, **kwargs)
异步提交任务
2、map(func, *iterables, timeout=None, chunksize=1)
取代for循环submit的操作
3、shutdown(wait=True)
相当于进程池的pool.close()+pool.join()操作
wait=True，等待池内所有任务执行完毕回收完资源后才继续
wait=False，立即返回，并不会等待池内的任务执行完毕
但不管wait参数为何值，整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前
4、result(timeout=None)
取得结果
5、add_done_callback(fn)
回调函数
用法
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import os,time,random
def task(n):
print('%s is runing' %os.getpid())
time.sleep(random.randint(1,3))
return n**2
if __name__ == '__main__':
executor=ProcessPoolExecutor(max_workers=3)
futures=[]
for i in range(11):
future=executor.submit(task,i)
futures.append(future)
executor.shutdown(True)
print('+++>')
for future in futures:
print(future.result())
爬虫小练习
from concurrent.futures import ThreadPoolExecutor
import requests
import time
def get(url):
print('GET %s' %url)
response=requests.get(url)
time.sleep(3)
return {'url':url,'content':response.text}
def parse(res):
res=res.result()
print('%s parse res is %s' %(res['url'],len(res['content'])))
if __name__ == '__main__':
urls=[
'http://www.cnblogs.com/linhaifeng',
'https://www.python.org',
'https://www.openstack.org',
]
pool=ThreadPoolExecutor(2)
for url in urls:
pool.submit(get,url).add_done_callback(parse)
基于进程池实现FTP
服务端
from socket import *
from concurrent.futures import ThreadPoolExecutor
def communicate(conn):
while True:
try:
data=conn.recv(1024)
if not data:break
conn.send(data.upper())
except ConnectionResetError:
break
conn.close()
def server(ip,port):
server = socket(AF_INET, SOCK_STREAM)
server.bind((ip,port))
server.listen(5)
while True:
conn, addr = server.accept()
pool.submit(communicate,conn)
server.close()
if __name__ == '__main__':
pool=ThreadPoolExecutor(2)
server('127.0.0.1', 8081)
同步与异步
同步调用:提交完任务后，就在原地等待任务执行完毕，拿到结果，再执行下一行代码,导致程序是串行执行
def la(name):
print('%s is laing' % name)
time.sleep(random.randint(3, 5))
res=random.randint(7, 13)*'#'
return {'name': name, 'res': res}
def weigh(shit):
name = shit['name']
size = len(shit['res'])
print('%s 拉了 《%s》kg' %(name,size))
if __name__ == '__main__':
pool = ThreadPoolExecutor(13)
shit1 = pool.submit(la, 'alex').result()
weigh(shit1)
shit2 = pool.submit(la, 'wupeiqi').result()
weigh(shit2)
shit3 = pool.submit(la, 'yuanhao').result()
weigh(shit3)
异步调用：提交完任务后，不地等待任务执行完毕。
from concurrent.futures import ThreadPoolExecutor
import time
import random
def la(name):
print('%s is laing' %name)
time.sleep(random.randint(3,5))
res=random.randint(7,13)*'#'
return {'name':name,'res':res}
def weigh(shit):
shit=shit.result()
name=shit['name']
size=len(shit['res'])
print('%s 拉了 《%s》kg' %(name,size))
if __name__ == '__main__':
pool=ThreadPoolExecutor(13)
pool.submit(la,'alex').add_done_callback(weigh)
pool.submit(la,'wupeiqi').add_done_callback(weigh)
pool.submit(la,'yuanhao').add_done_callback(weigh)