Python学习之==>线程&&进程

import threading,time
def run(): # 定义每个线程需要运行的函数
time.sleep(3)
print('呵呵呵') # 串行
for i in range(5): # 串行，需要运行15秒
run() # 多线程：
for j in range(5): # 并行：运行3秒
t = threading.Thread(target=run) # 实例化了一个线程
t.start()

import threading,time class MyThread(threading.Thread):
def __init__(self, num):
threading.Thread.__init__(self)
self.num = num def run(self): # 定义每个线程要运行的函数
print("running on number:%s" % self.num)
time.sleep(3) if __name__ == '__main__':
t1 = MyThread(1)
t2 = MyThread(2)
t1.start()
t2.start()

# threading 模块提供的常用方法：
# threading.currentThread(): 返回当前的线程变量。
# threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
# threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。
# 除了使用方法外，线程模块同样提供了Thread类来处理线程，Thread类提供了以下方法:
# run(): 用以表示线程活动的方法。
# start():启动线程活动。
# join([time]): 等待至线程中止。这阻塞调用线程直至线程的join() 方法被调用中止-正常退出或者抛出未处理的异常-或者是可选的超时发生。
# isAlive(): 返回线程是否活动的。
# getName(): 返回线程名。
# setName(): 设置线程名。

import requests,time,threading
# 定义需要下载的网页字典
urls = {
'besttest':'http://www.besttest.cn',
'niuniu':'http://www.nnzhp.cn',
'dsx':'http://www.imdsx.cn',
'cc':'http://www.cc-na.cn',
'alin':'http://www.limlhome.cn'
}
# 下载网页并保存成html文件
# 子线程运行的函数，如果里面有返回值的话，是不能获取到的
# 只能在函数外面定义一个list或者字典来存每次处理的结果
data = {}
def down_html(file_name,url):
start_time = time.time()
res = requests.get(url).content # content就是返回的二进制文件内容
open(file_name+'.html','wb').write(res)
end_time = time.time()
run_time = end_time - start_time
data[url] = run_time # 串行
start_time = time.time() # 记录开始执行时间
for k,v in urls.items():
down_html(k,v)
end_time = time.time() # 记录执行结束时间
run_time = end_time - start_time
print(data)
print('串行下载总共花了%s秒'%run_time) # 并行
start_time = time.time()
for k,v in urls.items():
t = threading.Thread(target=down_html,args=(k,v)) # 多线程的函数如果传参的话，必须得用args
t.start()
end_time = time.time()
run_time = end_time-start_time
print(data)
print('并行下载总共花了%s秒'%run_time)

import requests,time,threading
# 定义需要下载的网页字典
urls = {
'besttest':'http://www.besttest.cn',
'niuniu':'http://www.nnzhp.cn',
'dsx':'http://www.imdsx.cn',
'cc':'http://www.cc-na.cn',
'alin':'http://www.limlhome.cn'
}
# 下载网页并保存成html文件
# 子线程运行的函数，如果里面有返回值的话，是不能获取到的
# 只能在函数外面定义一个list或者字典来存每次处理的结果
data = {}
def down_html(file_name,url):
start_time = time.time()
res = requests.get(url).content # content就是返回的二进制文件内容
open(file_name+'.html','wb').write(res)
end_time = time.time()
run_time = end_time - start_time
data[url] = run_time # 串行
start_time = time.time() # 记录开始执行时间
for k,v in urls.items():
down_html(k,v)
end_time = time.time() # 记录执行结束时间
run_time = end_time - start_time
print(data)
print('串行下载总共花了%s秒'%run_time) # 多线程
start_time = time.time()
threads = [] # 存放启动的5个子线程
for k,v in urls.items():
# 多线程的函数如果传参的话，必须得用args
t = threading.Thread(target=down_html,args=(k,v))
t.start()
threads.append(t)
for t in threads: # 主线程循环等待5个子线程执行结束
t.join() # 循环等待
print(data) # 通过函数前面定义的data字典获取每个线程执行的时间
end_time = time.time()
run_time = end_time - start_time
print('并行下载总共花了%s秒'%run_time)

import threading
def run():
time.sleep(3)
print('哈哈哈') for i in range(50):
t = threading.Thread(target=run)
t.setDaemon(True) # 把子线程设置成为守护线程
t.start()
print('Done，运行完成')
time.sleep(3)

import threading def addNum():
global num #在每个线程中都获取这个全局变量
num-=1 num = 100 #设定一个全局变量
thread_list = []
for i in range(100):
t = threading.Thread(target=addNum)
t.start()
thread_list.append(t) for t in thread_list: #等待所有线程执行完毕
t.join() print('final num:', num ) # 运行结果为0

import threading,time def addNum():
global num #在每个线程中都获取这个全局变量 temp=num
# print('--get num:',num )
time.sleep(0.1)
num =temp-1 #对此公共变量进行减1操作 num = 100 #设定一个共享变量
thread_list = []
for i in range(100):
t = threading.Thread(target=addNum)
t.start()
thread_list.append(t) for t in thread_list: #等待所有线程执行完毕
t.join() print('final num:', num ) # 运行结果是99

import threading,time def addNum():
global num #在每个线程中都获取这个全局变量
lock.acquire() # 加锁
temp=num
print('--get num:',num )
time.sleep(0.1)
num =temp-1 #对此公共变量进行减1操作
lock.release() # 解锁 lock = threading.Lock() # 实例化一把锁
num = 100 #设定一个共享变量
thread_list = []
for i in range(100):
t = threading.Thread(target=addNum)
t.start()
thread_list.append(t) for t in thread_list: #等待所有线程执行完毕
t.join() print('final num:', num ) # 运行结果是0

import threading,time class myThread(threading.Thread):
def doA(self):
lockA.acquire()
print(self.name,"gotlockA",time.ctime())
time.sleep(3)
lockB.acquire()
print(self.name,"gotlockB",time.ctime())
lockB.release()
lockA.release() def doB(self):
lockB.acquire()
print(self.name,"gotlockB",time.ctime())
time.sleep(2)
lockA.acquire()
print(self.name,"gotlockA",time.ctime())
lockA.release()
lockB.release()
def run(self):
self.doA()
self.doB()
if __name__=="__main__": lockA=threading.Lock()
lockB=threading.Lock()
threads=[]
for i in range(5):
threads.append(myThread())
for t in threads:
t.start()
for t in threads:
t.join()

import threading,time class myThread(threading.Thread):
def doA(self):
lock.acquire()
print(self.name,"gotlockA",time.ctime())
time.sleep(3)
lock.acquire()
print(self.name,"gotlockB",time.ctime())
lock.release()
lock.release() def doB(self):
lock.acquire()
print(self.name,"gotlockB",time.ctime())
time.sleep(2)
lock.acquire()
print(self.name,"gotlockA",time.ctime())
lock.release()
lock.release()
def run(self):
self.doA()
self.doB()
if __name__=="__main__": lock = threading.RLock() # 递归锁
threads=[]
for i in range(5):
threads.append(myThread())
for t in threads:
t.start()
for t in threads:
t.join()

import threading,time
class myThread(threading.Thread):
def run(self):
if semaphore.acquire():
print(self.name)
time.sleep(1)
semaphore.release() if __name__=="__main__":
semaphore=threading.BoundedSemaphore(5)
thrs=[]
for i in range(100):
thrs.append(myThread())
for t in thrs:
t.start()

import threading,time
from random import randint
class Producer(threading.Thread):
def run(self):
global L
while True:
val=randint(0,100)
print('生产者',self.name,":Append"+str(val),L)
if lock_con.acquire():
L.append(val)
lock_con.notify()
lock_con.release()
time.sleep(3)class Consumer(threading.Thread):
def run(self):
global L
while True:
lock_con.acquire()
if len(L)==0:
lock_con.wait()
print('消费者',self.name,":Delete"+str(L[0]),L)
del L[0]
lock_con.release()
time.sleep(1)if __name__=="__main__":
L=[]
lock_con=threading.Condition()
threads=[]
for i in range(5):
threads.append(Producer())
threads.append(Consumer())
for t in threads:
t.start()
for t in threads:
t.join()

import threading,time
class Boss(threading.Thread):
def run(self):
print("BOSS：今晚大家都要加班到22:00。")
event.isSet() or event.set()
time.sleep(3)
print("BOSS：<22:00>可以下班了。")
event.isSet() or event.set() class Worker(threading.Thread):
def run(self):
event.wait()
print("Worker：哎……命苦啊！")
time.sleep(1)
event.clear()
event.wait()
print("Worker：Oh，Yeah!！") if __name__=="__main__":
event=threading.Event()
threads=[]
for i in range(5):
threads.append(Worker())
threads.append(Boss())
for t in threads:
t.start()
for t in threads:
t.join()

# 创建一个“队列”对象
import queue
q = queue.Queue(maxsize = 10)
# queue.Queue类即是一个队列的同步实现。队列长度可为无限或者有限。可通过Queue的构造函数的可选参数maxsize来设定队列长度。如果maxsize小于1就表示队列长度无限。 # 将一个值放入队列中
q.put(10)
# 调用队列对象的put()方法在队尾插入一个项目。put()有两个参数，第一个item为必需的，为插入项目的值；第二个block为可选参数，默认为:1。如果队列当前为空且block为1，put()方法就使调用线程暂停,直到空出一个数据单元。如果block为0，put方法将引发Full异常。 # 将一个值从队列中取出
q.get()
# 调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block，默认为True。如果队列为空且block为True，get()就使调用线程暂停，直至有项目可用。如果队列为空且block为False，队列将引发Empty异常。 # Python Queue模块有三种队列及构造函数:
queue.Queue(maxsize=10) # Python Queue模块的FIFO队列先进先出。
queue.LifoQueue(maxsize=10) # LIFO类似于堆，即先进后出。
queue.PriorityQueue(maxsize=10) # 还有一种是优先级队列级别越低越先出来。 # 此包中的常用方法(q = queue.Queue()):
# q.qsize() 返回队列的大小
# q.empty() 如果队列为空，返回True,反之False
# q.full() 如果队列满了，返回True,反之False
# q.full 与 maxsize 大小对应
# q.get([block[, timeout]]) 获取队列，timeout等待时间
# q.get_nowait() 相当q.get(False)
# 非阻塞 q.put(item) 写入队列，timeout等待时间
# q.put_nowait(item) 相当q.put(item, False)
# q.task_done() 在完成一项工作之后，q.task_done() 函数向任务已经完成的队列发送一个信号
# q.join() 实际上意味着等到队列为空，再执行别的操作

import threading,queue
from time import sleep
from random import randint
class Production(threading.Thread):
def run(self):
while True:
r=randint(0,100)
q.put(r)
print("生产出来%s号包子"%r)
sleep(1)
class Proces(threading.Thread):
def run(self):
while True:
re=q.get()
print("吃掉%s号包子"%re)
if __name__=="__main__":
q=queue.Queue(10)
threads=[Production(),Production(),Production(),Proces()]
for t in threads:
t.start()
for t in threads:
t.join()

import multiprocessing,time def f(name):
time.sleep(1)
print('Hello!!',name,time.ctime()) if __name__ == '__main__':
p_list = []
for i in range(3):
p = multiprocessing.Process(target=f,args=('Porcess',))
p_list.append(p)
p.start()
for i in p_list:
i.join() print('end')

import multiprocessing,threading
def my():
print('哈哈哈') def run(num):
for i in range(num):
t = threading.Thread(target=my)
t.start()
# 总共启动5个进程，每个进程下面启动6个线程，函数my()执行30次
if __name__ == '__main__':
process = []
for i in range(5):
# args只有一个参数一定后面要加逗号
p = multiprocessing.Process(target=run,args=(6,)) # 启动一个进程
p.start()
process.append(p)
[p.join() for p in process] # 与线程用法一致

class MyProcess(multiprocessing.Process):
def __init__(self):
super(MyProcess, self).__init__()
# self.name = name def run(self):
time.sleep(1)
print('Hello!!',self.name,time.ctime()) if __name__ == '__main__':
p_list = []
for i in range(3):
p = MyProcess()
p.start()
p_list.append(p)
for p in p_list:
p.join()
print('end')

from multiprocessing import Process,Queue def f(q,i):
q.put([42,i,'hello'])
print('subprocess q_id:',id(q)) if __name__ == '__main__':
q = Queue()
p_list=[]
print('main q_id:',id(q))
for i in range(3):
p = Process(target=f, args=(q,i))
p_list.append(p)
p.start()
print(q.get())
print(q.get())
print(q.get())
for p in p_list:
p.join()

from multiprocessing import Process,Pipe def f(child_conn):
child_conn.send('子进程')
child_conn.send([42, None, 'hello'])
print(child_conn.recv()) # 子进程接收父进程发送的信息,打印'数组'
child_conn.close() if __name__ == '__main__':
parent_conn,child_conn = Pipe()
p = Process(target=f, args=(child_conn,))
p.start()
# 父进程接收子进程发送的信息
print(parent_conn.recv()) # 打印 '子进程'
print(parent_conn.recv()) # 打印 '[42, None, 'hello']'
parent_conn.send('数组') # 父进程给子进程发送信息
p.join()

from multiprocessing import Process,Manager def f(dic,lis,n):
dic[n] = ''
dic[''] = 2
dic[0.25] = None
lis.append(n)
# print(lis) if __name__ == '__main__':
# with Manager() as manager:
manager = Manager()
dic = manager.dict()
lis = manager.list(range(5))
p_list = []
for i in range(10):
p = Process(target=f, args=(dic,lis,i))
p.start()
p_list.append(p)
for res in p_list:
res.join() print(dic)
print(lis)

from multiprocessing import Process,Lock def f(l,i):
l.acquire()
print('hello world',i)
l.release() if __name__ == '__main__':
lock = Lock()
p_list = []
for num in range(10):
p = Process(target=f, args=(lock,num))
p.start()
p_list.append(p)
for p in p_list:
p.join()
print('end')

apply(func [,args [,kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。
'''需要注意的是：此操作并不会在所有的池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数，必须从不同线程调用p.apply()函数或者使用apply_async()''' apply_async(func [,args [,kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。
'''此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变为可用时，将回调函数传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。''' close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成 join():等待所有工作进程退出。此方法只能在close（）或teminate()之后调用

import time,multiprocessing
def func(i):
i += 1 if __name__ == '__main__':
# 进程池
p = multiprocessing.Pool(5)
start = time.time()
p.map(func,range(100)) # 使用map函数循环调用func函数
p.close() # 不允许再向进程池中添加任务
p.join()
print(time.time()-start) # 0.21729111671447754
# 多进程
start = time.time()
p_list = []
for i in range(100):
# 多进程调用func函数
p = multiprocessing.Process(target=func,args=(i,))
p.start()
p_list.append(p)
for p in p_list:
p.join()
print(time.time()-start) # 3.7114222049713135

import multiprocessing,os,time
def func(i):
print('%s run'%os.getpid())
time.sleep(1)
i += 1 if __name__ == '__main__':
# 进程池
p = multiprocessing.Pool(5)
for i in range(20):
p.apply(func,args=(i,)) # 同步提交,串行执行

import time,multiprocessing
def func(i):
time.sleep(1)
i += 1
print(i) if __name__ == '__main__':
# 进程池
p = multiprocessing.Pool(5)
for i in range(20):
p.apply_async(func, args=(i,)) # 异步提交，并行执行
p.close() # 不允许再向进程池中添加任务,close()必须在join()前面
# 使用异步提交的任务,必须添加join()
p.join() # 等待子进程结束再往下执行，否则主进程结束了子进程还没执行完
print('end')

import time,multiprocessing def func(i):
time.sleep(1)
i += 1
return i if __name__ == '__main__':
# 进程池
p = multiprocessing.Pool(5)
res_l = []
for i in range(20):
res = p.apply_async(func,args=(i,)) # 异步提交
# print(res.get()) # 阻塞，等待任务结果
res_l.append(res) # 返回结果之后，将结果放入列表，归还进程，之后再执行新的任务
# 需要注意的是，进程池中的三个进程不会同时开启或者同时结束，而是执行完一个就释放一个进程，这个进程就去接收新的任务
p.close() # 不允许再向进程池中添加任务,close()必须在join()前面
p.join() # 等待子进程结束再往下执行，必须添加join()，否则主进程结束了子进程还没执行完
for i in res_l:
print(i.get()) # 使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get
print('end')

from multiprocessing import Process,Pool
import time def Foo(i):
time.sleep(2)
return i + 100 def Bar(arg): # 回调函数是在主进程中完成的，直接接收子进程中函数的返回值，不能传另外的参数
print('----->exec done:',arg) if __name__ == '__main__':
pool = Pool(5) # 允许进程池里同时放入5个进程
for i in range(10):
pool.apply_async(func=Foo, args=(i,),callback=Bar) # 并行执行，Bar函数接收Foo函数的返回结果，callback回调执行者为主进程
pool.close()
pool.join() # 进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭
print('end')

import multiprocessing,requests def get_url(url):
res = requests.get(url)
return {'url':url,
'status_code':res.status_code,
'content':res.text} def parser(dic):
print(dic['url'],dic['status_code'],len(dic['content'])) if __name__ == '__main__':
url_list = ['http://www.baidu.com',
'http://www.hao123.com',
'http://www.163.com',
'http://www.csdn.com']
p = multiprocessing.Pool(4)
res_l = []
for url in url_list:
p.apply_async(get_url,args=(url,),callback=parser)
p.close()
p.join()
print('END!!')