Python 多进/线程 协程 的实现:二

Python 多进/线程 协程 的实现:二

• 1.实现进程与线程的模块:multiprocessing/threading
• 2.I/O密集型/计算密集型

1.实现进程与线程的模块:multiprocessing/threading

multiprocessing.Process

Process的语法如下：(https://blog.csdn.net/u014745194/article/details/70860986)
Process([group [, target [, name [, args [, kwargs]]]]])
target：表示这个进程实例所调用对象；
args：表示调用对象的位置参数元组；
kwargs：表示调用对象的关键字参数字典；
name：为当前进程实例的别名；
group：大多数情况下用不到，表示在哪个组

Process类常用方法：
is_alive()：判断进程实例是否还在执行；
join([timeout])：是否等待进程实例执行结束，或等待多少秒；
start()：启动进程实例（创建子进程）；
run()：如果没有给定target参数，对这个对象调用start()方法时，就将执行对象中的run()方法；
terminate()：不管任务是否完成，立即终止；
Process类常用属性：
name：当前进程实例别名，默认为Process-N，N为从1开始递增的整数；
pid：当前进程实例的PID值；

multiprocessing.Pool常用函数解析:
apply_async(func[, args[, kwds]]) ：使用非阻塞方式调用func
（并行执行，堵塞方式必须等待上一个进程退出才能执行下一个进程），args为传递给func的参数列表，
kwds为传递给func的关键字参数列表；
apply(func[, args[, kwds]])：使用阻塞方式调用func
close()：关闭Pool，使其不再接受新的任务；
terminate()：不管任务是否完成，立即终止；
join()：主进程阻塞，等待子进程的退出， 必须在close或terminate之后使用；

threading.Thread

Thread的语法如下：
Thread([group [, target [, name [, args [, kwargs]]]]])
target：表示这个线程实例所调用对象；
args：表示调用对象的位置参数元组；
kwargs：表示调用对象的关键字参数字典；
name：为当前线程实例的别名；
group：大多数情况下用不到，表示在哪个组；

Thread类常用方法：
is_alive()：判断线程实例是否还在执行；
join([timeout])：是否等待线程实例执行结束，或等待多少秒；
start()：启动线程实例（创建子线程）；
run()：如果没有给定target参数，对这个对象调用start()方法时，就将执行对象中的run()方法；
setDaemon()	主线程执行过程中，守护线程也在进行，主线程执行完毕后，等待守护线程执行完成后，程序才停止。
start()方法调用之前设置，如果不设置为守护线程，程序会被无限挂起。

Thread类常用属性：
name：当前线程实例别名，默认为Thread-N，N为从1开始递增的整数；
current_thread()	返回当前线程
active_count()	返回当前活跃的线程数，1个主线程+n个子线程
get_ident()	返回当前线程
enumerater()	返回当前活动 Thread 对象列表
main_thread()	返回主 Thread 对象

#-*- coding:utf-8 -*-
from time import ctime,sleep
import threading #多线程
#多进程:multiprocessing

class MyThread(threading.Thread):
def __init__(self,func,args,name=''):
threading.Thread.__init__(self)
self.func = func
self.args = args
self.name = name

def run(self):
threadLock.acquire()
self.func(*self.args)
threadLock.release()

def super_player(file,stime):
for i  in range(2):
print('Start Playing:%s %s'%(file,ctime()))
test1(file)
test2(file)
sleep(stime)

def test1(a):
s1 = a * a
print('s1:%s %s'%(s1,ctime()))
def test2(a):
s2 = a * a
print('s2:%s %s'%(s2,ctime()))

threadLock = threading.RLock()
list = {2:3,5:4}

threads = []
for (v,k) in list.items():
t = MyThread(super_player,(v,k),super_player.__name__)
threads.append(t)

if __name__=='__main__':
for i in range(len(list)):
threads[i].start()
for i in range(len(list)):
threads[i].join()

print('the end:%s'%ctime())

2.I/O密集型/计算密集型

计算密集型/CPU利用率的比较:

from threading import Thread
from multiprocessing import Process,Pool
from time import *

#3次是为了更好显示出进/线程的不同
#线程:一个cpu上执行3个线程
#进程:三个cpu上分别执行一个进程

def getValue():#普通python计算
result = 0
start_time = time()
for i in range(100000):
result += (1+2**i+i*33)

stop_time = time()
sum_time = stop_time - start_time
print("result:%s and time:%s"%(result,sum_time*3))
print("length of result:",len(str(result)))

def getValueByThread(name):#线程:一个cpu上执行3个线程
result = 0

for i in range(100000):
result += (1+2**i+i*33)

print("the Name:%s result:%s"%(name,result))

def getValueByProcess(name):#进程:三个cpu上分别执行一个进程
result = 0

for i in range(100000):
result += (1 + 2 ** i + i * 33)

print("the Name:%s result:%s"%(name,result))

if __name__ == '__main__':
getValue() #time:28.422567129135132
pool = Pool(3)
start_time = time()
for i in range(3):
pool.apply_async(getValueByProcess,args=("p1",))

print("等待所有子进程完成...")
pool.close()
pool.join()
stop_time = time()
sum_time = stop_time - start_time

print("进程总共耗时:%s"%sum_time)#time:11.65796971321106

start_time1 = time()
thread_list = []
for i in range(3):
t = Thread(target=getValueByThread,args=("t"+str(i),))
thread_list.append(t)

for j in thread_list:
j.start()

for j in thread_list:
j.join()

stop_time1 = time()
sum_time1 = stop_time1 - start_time1

print("线程总共耗时:%s" % sum_time1)#time:27.50909924507141

结论:

Time:3*(线程=普通)=进程(约等于)

CPU:3*(线程=普通)=进程(约等于)

I/O密集型的比较:

#!/usr/bin/env python2
# -*- coding=utf-8 -*-

from threading import Thread
from queue import  Queue
import time
from lxml import etree
import requests

class DouBanSpider(Thread):
def __init__(self, url, q):
# 重写写父类的__init__方法
super(DouBanSpider, self).__init__()
self.url = url
self.q = q

self.headers = {
'Cookie': '***',#豆瓣的登录信息我就不放了
'Host': 'movie.douban.com',
'Referer': 'https://movie.douban.com/top250?start=225&filter=',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.104 Safari/537.36',
}

self.proxy = {"***"}#设置ip代理

def run(self):
self.parse_page()

def send_request(self, url):
'''
用来发送请求的方法
:return: 返回网页源码
'''
# 请求出错时，重复请求３次,
i = 0
while i <= 3:
try:
print(u"[INFO]请求url:" + url)
html = requests.get(url=url, headers=self.headers,proxies = self.proxy).content
except Exception as e:
print(u'[INFO] %s%s' % (e, url))
i += 1
else:
return html

def parse_page(self):
'''
解析网站源码，并采用xpath提取　电影名称和平分放到队列中
:return:
'''
response = self.send_request(self.url)
html = etree.HTML(response)
# 　获取到一页的电影数据
node_list = html.xpath("//div[@class='info']")
for move in node_list:
# 电影名称
title = move.xpath('.//a/span/text()')[0]
# 评分
score = move.xpath('.//div[@class="bd"]//span[@class="rating_num"]/text()')[0]

# 将每一部电影的名称跟评分加入到队列
self.q.put(score + "\t" + title)

def main():
# 创建一个队列用来保存进程获取到的数据
q = Queue()
base_url = 'https://movie.douban.com/top250?start='
# 构造所有url
url_list = [base_url + str(num) for num in range(0, 225 + 1, 25)]

# 保存线程
Thread_list = []
# 创建并启动线程
for url in url_list:
p = DouBanSpider(url, q)
p.start()
Thread_list.append(p)

# 让主线程等待子线程执行完成
for i in Thread_list:
i.join()

while not q.empty():

2b612
print(q.get())

if __name__ == "__main__":
start = time.time()
main()
print('[info]耗时：%s' % (time.time() - start))

结论

Time:线程<<普通=进程(约等于)

参考博客
进程，线程以及协程 爬虫