线程,有时被称为轻量级进程,是程序执行流的最小单元
线程是程序中一个单一的顺序控制流程。进程内一个相对独立的、可调度的执行单元,是系统独立调度和分派CPU的基本单位指进行中的程序的调度单位。在单个程序中同时运行多个线程完成不同的工作,称为多线程。
python中多个cpu无法同时处理一个进程或其子进程,多个cpu可以同时处理多个线程
1 import time
def f1(arg):
time.sleep(1)
print(arg)
import threading
t = threading.Thread(target = f1,args = (123,))
#t.setDaemon(True)#表示主线程不等子线程
t.start()#不代表当前线程会立即被执行
t.join(2) #表示主线程到此等待。。。直到子线程执行完成#参数,表示住线程再次最多等待n秒 print('end')
#一秒后显示
>>>123
>>>end import time
def f1(arg):
time.sleep(1)
print(arg)
import threading
t = threading.Thread(target = f1,args = (123,))
t.setDaemon(True)#表示主线程不等子线程
t.start()#不代表当前线程会立即被执行
#t.join(2) #表示主线程到此等待。。。直到子线程执行完成#参数,表示住线程再次最多等待n秒 print('end')
#立即显示
>>>end
#我们可以写个类继承threading模块的Thead类并加入自定义的构造方法,
#用来添加新功能
class MyThread(threading.Thread):
def __init__(self,func,args):
self.func =func
self.args = args
#继承父类的构造方法
super(MyThread,self).__init__() def run(self):
self.func(self.args) def f2(arg):
print(arg) obj = MyThread(f2,123)
obj.start()
>>>123

队列

 #Python中,队列是线程间最常用的交换数据的形式。Queue模块是提供队列操作的模块,不同的队列应用在不同的场景中
#queue.Queue先进先出队列
#queue.LifoQueue,后进先出队列
#queue.PriorityQueue,优先级队列
#queue.deque,双向队列 import queue
q = queue.Queue(3)#参数为队列中最大个数
print(q.empty())#判断是否为空
>>>True
print(q.full())#判断队列是否已满
>>>False
q.put(11)
>>>传入元素
q.put(22)
>>>传入元素
#q.put(33,block=False)#最大值为2传入第三个的时候默认阻塞
print(q.qsize())#真实队列剩余个数
>>>2
print(q.maxsize)#最大个数
>>>3
print(q.get())#取值当队列中取完之后继续取得时候阻塞
>>>11
q.task_done()#任务完成
print(q.get())#取值
>>>22
q.task_done()#任务完成,用于释放队列
q.join()#不带这个的时候程序完成后释放队列,加上的时候阻塞
 #先进后出队列
q = queue.LifoQueue()
q.put(123)
q.put(456)
print(q.get())
#>>>456 #优先级队列
q = queue.PriorityQueue()
q.put((1,'alex1'))
q.put((1,'alex2'))
q.put((3,'alex3'))
print(q.get())
>>>(1, 'alex1') #双向队列
q= queue.deque()
q.append(123)
q.append(333)
q.appendleft(456)
#从左侧插入队列
print(q.pop())#从右侧取值
print(q.popleft())#从左侧取值
>>>333
>>>456

生产者消费者模型

python-线程进程与队列-LMLPHP

python-线程进程与队列-LMLPHP

#生产者消费者模型
import queue
import threading
import time #创建队列
q = queue.Queue(50) #定义消费者
def productor(arg):
'''
买票
:param arg:
:return:
'''
while True:
q.put(str(arg) + '号产生订单')#提交到队列 #创建300个线程发送请求
for i in range(300):#300个线程同时提交订单相当于300个人同时提交订单
t = threading.Thread(target= productor,args= (i,))
t.start() #定义生产者
def consumer(arg):
'''
服务器后台
:param arg:
:return:
'''
while True:
print(str(arg) + '处理了'+q.get())#进程从队列中取订单进行处理 #3个线程同时工作
for j in range(3):
t = threading.Thread(target=consumer,args=(j,))
t.start()

线程锁

 #线程锁
import threading
import time NUM = 10
#线程锁线程执行进程通过的接口,用来限制多个线程同时修改一个数据
def func(l):
global NUM
#上锁
l.acquire()
NUM -=1
time.sleep(2)
print(NUM)
#开锁
l.release()
#单层锁
lock = threading.Lock()
#多层锁
#lock = threading.RLock() for i in range(30):
t = threading.Thread(target=func,args = (lock,))
t.start()

信号量

 #设置可通过线程个数
import threading
import time
NUM =10
def func(i,l):
global NUM
#上锁
l.acquire()
NUM -=1
time.sleep(2)
print(NUM,i)
#开锁
l.release() #调用信号量设置每次多少个线程处理进程
lock = threading.BoundedSemaphore(5) for i in range(30):
t = threading.Thread(target= func,args=(i,lock,))
t.start()
 #event相当于红绿灯,通过一个标识来批量管理线程
import threading def func(i,e):
print(i)
e.wait()#检测时什么灯,如果是红灯,停,绿灯,行
print(i+100) event = threading.Event() for i in range(10):
t = threading.Thread(target= func,args = (i,event,))
t.start() event.clear()#设置成红灯
inp = input('>>>')
if inp == '':
event.set()#设置成绿灯
 #根据条件限定线程的执行
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#设置条件设置线程数第一种方式
import threading
def func(i,con):
print(i)
con.acquire()
con.wait()
print(i+100)
con.release() c = threading.Condition()
for i in range(10):
t = threading.Thread(target=func,args=(i,c))
t.start() while True:
inp = input('>>>')
if inp == 'q':
break
c.acquire()
c.notify(int(inp))#根据输入设置通过几个线程数
c.release() #第二种
import threading def condition():
ret =False
r = input('>>>')
if r == 'true':
ret =True
else:
ret = False
return ret def func(i,con):
print(i)
con.acquire()
# 设置condition函数为条件返回true继续运行,条件不成立则不执行此线程
con.wait_for(condition)
print(i+100)
con.release() c = threading.Condition()
for i in range(10):
t = threading.Thread(target=func,args=(i,c))
t.start()

Timer

 #定时器
form threading import Time def hello():
print('hello')
t = Timer(1,hello)#一秒后执行
t.start()

线程池

 import queue
import threading
import time class ThreadPool:
def __init__(self,maxsize):
self.maxsize = maxsize
self._q = queue.Queue(maxsize)#创建队列
for i in range(maxsize):
self._q.put(threading.Thread)#将创建线程的类放入队列 def get_thread(self):
return self._q.get()#获取队列的值 def add_thread(self):
self._q.put(threading.Thread) pool = ThreadPool(5)#设置线程池
def task(arg,p):
print(arg)
time.sleep(1)
p.add_thread()#添加新的线程 for i in range(100):
t = pool.get_thread()#当获取5次后,阻塞在此
obj = t(target = task,args = (i,pool,))#创建线程调用函数task
obj.start()
 #第二种创建线程池方式
#!/usr/bin/env python
# -*- coding:utf-8 -*- import queue
import threading
import contextlib
import time StopEvent = object()
class Pool:
def __init__(self,max_num,max_task_num=None):
if max_task_num:
self.q=queue.Queue(max_task_num)#创建队列并指定接受任务最大数
else:
self.q = queue.Queue()#不指定参数
self.max_num = max_num#最多有多少个线程
self.cancel = False
self.terminal = False
self.generate_list = []#已创建线程
self.free_list = []#空闲线程 def run(self,func,args,callback = None):#接收参数
if self.cancel:
return
if len(self.free_list) ==0 and len(self.generate_list) < self.max_num:#当没有空闲线程并且,已创建的线程没有达到最大值
self.generate_thread() #创建新线程调generte_thread函数
w = (func,args,callback,)#将参数传入队列中
self.q.put(w)#将参数作为元组传入队列中 def generate_thread(self):
'''
创建线程
:return:
'''
t = threading.Thread(target=self.call)#执行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#将传过来的参数赋值给event
try:
result = func(args)#执行action(i)
success = True#任务执行成功
except Exception as e:
success = False#action任务执行失败
if callback is not None:
try:
callback(success,result)
except Exception as c:
pass
#event.self.q.get()#继续去任务,当存在任务则执行action不存在则删除当前进程
with self.worker_state(self.free_list,current_thread):#任务执行完成后设置该线程为空闲
if self.terminal:#如果是空闲的
event = StopEvent
else:
event = self.q.get()#如果不是空闲的,则去取任务
else:
self.generate_list.remove(current_thread)#如果任务为空则删除当前线程 def close(self):
'''
执行完所有任务后,所有线程停止
:return:
'''
self.cancel = True
full_size = len(self.generate_list)#统计线程个数
while full_size:#根据线程个数传入对应个数的False标志
self.q.put(StopEvent)
full_size -=1 def terminate(self):
"""
无论是否还有任务,终止线程
"""
self.terminal = True while self.generate_list:
self.q.put(StopEvent)
self.q.empty() @contextlib.contextmanager
def worker_state(self, state_list, worker_thread):
"""
用于记录线程中正在等待的线程数
"""
state_list.append(worker_thread)
try:
yield
finally:
state_list.remove(worker_thread)
pool = Pool(5) def action():
pass
def callback(i):
print(i)
for i in range(300):
ret = pool.run(action,(i,),callback)#将函数i的值与callback函数传入类中
05-11 14:10
查看更多