13518219792
这篇文章将为大家详细讲解有关python中怎么实现多线程和多进程，文章内容质量较高，因此小编分享给大家做个参考，希望大家阅读完这篇文章后对相关知识有一定的了解。
创新互联建站主营克拉玛依网站建设的网络公司,主营网站建设方案,app软件开发,克拉玛依h5小程序开发搭建,克拉玛依网站营销推广欢迎克拉玛依等地区企业咨询
1、GIL
名称:全局解释器锁。在cpython中python中一个线程对应c语言中的一个线程。GIL使得同一时刻只有一个线程运行在同一个cpu上，无法将多个线程映射到 多个cpu上，从而保证了线程在某种时刻是安全的。GIL并不会一直占有，会根据执行的字节码行数、时间片或io操作时释放。非常适用io操作。
例如：
import threading var_total = 0def add():global var_totalfor var_i in range(1000000): var_total += 1def desc():global var_totalfor var_i in range(1000000): var_total -= 1if __name__ == '__main__': var_thread1 = threading.Thread(target=add) var_thread2 = threading.Thread(target=desc) var_thread1.start() var_thread2.start() var_thread1.join() var_thread2.join()print(var_total)
2、多线程编程
操作系统所能操作和调度的最小单元为线程。线程的调度比进程更加轻量级。对于io操作来说多进程和多线程性能相当。
多线程实现-方法
import timeimport threadingdef getDetailHtml(var_url):print('get detail html started')
time.sleep(2)print('get detail html end')def getDetailUrl(var_url):print('get detail url started')
time.sleep(4)print('get detail url end')if __name__ == '__main__':
var_thread1 = threading.Thread(target=getDetailHtml,args=('http://www.baidu.com',))
var_thread2 = threading.Thread(target=getDetailUrl, args=('http://www.baidu.com',))
var_start_time = time.time()#设置当主线程退出,子线程终结,把线程设置成守护线程 #var_thread1.setDaemon(True) var_thread2.setDaemon(True)
var_thread1.start()
var_thread2.start()#线程阻塞,等待子线程执行完成,主线程退出 #var_thread1.join() #var_thread2.join() print('last time:{}'.format(time.time() - var_start_time))多线程实现-类
import threadingimport timeclass GetDetailHtml(threading.Thread):def __init__(self,var_name):super().__init__(name = var_name)def run(self):print('get detail html started')
time.sleep(2)print('get detail html end')class GetDetailUrl(threading.Thread):def run(self):print('get detail url started')
time.sleep(4)print('get detail url end')if __name__ == '__main__':
var_thread1 = GetDetailHtml('getDetailHtml')
var_thread2 = GetDetailUrl()
var_start_time = time.time()
var_thread1.start()
var_thread2.start()
var_thread1.join()
var_thread2.join()print('last time:{}'.format(time.time() - var_start_time))3、线程间的通信Queue
from queue import Queueimport threadingimport timedef setDetailUrl(var_detail_queue):while True:
time.sleep(2)for var_i in range(5):
var_detail_queue.put('https://www.{}.com'.format(var_i))def getDetailUrl(var_detail_queue,var_thread_name):while True:
var_url = var_detail_queue.get()
time.sleep(2)print(var_thread_name,':',var_url)if __name__ == '__main__':
var_detail_queue = Queue(maxsize=10000)
var_set_detail = threading.Thread(target=setDetailUrl,args=(var_detail_queue,))
var_get_thread_list= []for var_index in range(5):
var_get_thread = threading.Thread(target=getDetailUrl,args=(var_detail_queue,'thread'+str(var_index)))
var_get_thread_list.append(var_get_thread)
var_set_detail.start()for var_one in var_get_thread_list:
var_one.start()
var_set_detail.join()for var_one in var_get_thread_list:
var_one.join()4、线程间的锁Lock、Rlock
import threadingfrom threading import Lock var_total = 0var_lock = Lock()def add():global var_totalfor var_i in range(1000000):#使用锁会影响性能,使用锁会出现死锁(资源竞争也会出现死锁),lock不能连续acquire多次,否则出现死锁 var_lock.acquire() var_total += 1 var_lock.release()def reduce():global var_totalfor var_i in range(1000000): var_lock.acquire() var_total -= 1 var_lock.release()if __name__ == '__main__': var_thread_add = threading.Thread(target=add) var_thread_reduce = threading.Thread(target=reduce) var_thread_add.start() var_thread_reduce.start() var_thread_add.join() var_thread_reduce.join()print(var_total)
import threadingfrom threading import RLock var_total = 0var_lock = RLock()def add():global var_totalfor var_i in range(1000000):#RLock在同一个线程里面可以连续调用多次acquire,acquire的次数要和release的次数相等 var_lock.acquire() var_lock.acquire() var_total += 1 var_lock.release() var_lock.release()def reduce():global var_totalfor var_i in range(1000000): var_lock.acquire() var_total -= 1 var_lock.release()if __name__ == '__main__': var_thread_add = threading.Thread(target=add) var_thread_reduce = threading.Thread(target=reduce) var_thread_add.start() var_thread_reduce.start() var_thread_add.join() var_thread_reduce.join()print(var_total)
5、线程同步condition、Semaphore
from threading import Thread,Conditionclass XiaoAi(Thread):def __init__(self, var_con):super().__init__(name='小爱')self.var_con = var_condef run(self):with self.var_con:self.var_con.wait()print('{}:在'.format(self.name))self.var_con.notify()self.var_con.wait()print('{}:好啊'.format(self.name))class TianMao(Thread):def __init__(self, var_con):super().__init__(name='天猫精灵')self.var_con = var_condef run(self):with self.var_con:#notify，wait方法必须在with语句之后调用 print('{}:小爱同学'.format(self.name))#天猫先启动，notify已发出但是小爱未启动时未接收到 self.var_con.notify()self.var_con.wait()print('{}:我们来对古诗吧'.format(self.name))self.var_con.notify()if __name__ == '__main__':
var_con = Condition()
var_tianmao = TianMao(var_con)
var_xiaoai = XiaoAi(var_con)#注意启动顺序 var_xiaoai.start()
var_tianmao.start()#Semaphore用于控制进入数量的锁import threadingimport timeclass HtmlSpider(threading.Thread):def __init__(self, var_url, var_sem):super().__init__()self.var_url = var_urlself.var_sem = var_semdef run(self):
time.sleep(2)print(self.var_url)self.var_sem.release()class UrlProducer(threading.Thread):def __init__(self, var_sem):super().__init__()self.var_sem = var_semdef run(self):for var_i in range(20):self.var_sem.acquire()
var_thread = HtmlSpider('https://www.baidu.com{}'.format(var_i), self.var_sem)
var_thread.start()if __name__ == '__main__':
var_sem = threading.Semaphore(5)
var_url_producer = UrlProducer(var_sem)
var_url_producer.start()6、线程池
from concurrent.futures import ThreadPoolExecutor, as_completed, waitimport timefrom concurrent.futures import Future'''线程池，主线程中可以获取某一个线程的状态或者某一任务的状态，以及返回值当一个线程完成时主线程可以立即知道futures可以让多线程和多进程编码接口一致'''def getHtml(var_times):
time.sleep(var_times)print('get page {} success'.format(var_times))return var_timesif __name__ == '__main__':
var_execute = ThreadPoolExecutor(max_workers=1)# 通过submit函数提交执行的函数到线程池中,submit非阻塞的,立即返回 var_task1 = var_execute.submit(getHtml,(2))
var_task2 = var_execute.submit(getHtml,(3))#done方法用于判定任务是否执行成功 print(var_task1.done())print(var_task2.done())#在未执行的时候可以取消任务 print(var_task2.cancel())
time.sleep(3)print(var_task1.done())print(var_task2.done())#返回执行结果 print(var_task1.result())print('*'*20)#获取已经成功的task返回值 var_execute_01 = ThreadPoolExecutor(max_workers=2)
var_times = [3,2,5,6,4]#list推导式 var_tasks = [var_execute_01.submit(getHtml, var_one_time) for var_one_time in var_times ]#阻塞线程 wait(var_tasks)print('wait')#as_completed 是一个生成器,谁先完成就打印谁 for var_one_complete_task in as_completed(var_tasks):print(var_one_complete_task.result())print('*' * 20)#使用map函数，打印顺序和var_times一致 var_tasks_data = var_execute_01.map(getHtml, var_times)for var_one_tasks_data in var_tasks_data:print(var_one_tasks_data)7、多进程和多线程比较
#耗cpu的操作选用多进程编程，对于IO操作选用多线程编程，进程切换代价要高于线程import timefrom concurrent.futures import ThreadPoolExecutor, as_completedfrom concurrent.futures import ProcessPoolExecutordef fib(var_n):if var_n <= 2:return 1 return fib(var_n - 1) + fib(var_n - 2)def threadPool():with ThreadPoolExecutor(3) as var_executor:
var_all_tasks = [var_executor.submit(fib, (var_num)) for var_num in range(25,35)]
var_start_time = time.time()for var_future in as_completed(var_all_tasks):print('result: {}'.format(var_future.result()))print('thread run time is {} s'.format(time.time() - var_start_time))def processPool():with ProcessPoolExecutor(3) as var_executor:
var_all_tasks = [var_executor.submit(fib, (var_num)) for var_num in range(25,35)]
var_start_time = time.time()for var_future in as_completed(var_all_tasks):print('result: {}'.format(var_future.result()))print('process run time is {} s'.format(time.time() - var_start_time))def randomSleep(var_n):
time.sleep(var_n)return var_ndef threadIoPool():with ThreadPoolExecutor(3) as var_executor:
var_all_tasks = [var_executor.submit(randomSleep, (var_num)) for var_num in range(5,10)]
var_start_time = time.time()for var_future in as_completed(var_all_tasks):print('result: {}'.format(var_future.result()))print('thread io run time is {} s'.format(time.time() - var_start_time))def processIoPool():with ProcessPoolExecutor(3) as var_executor:
var_all_tasks = [var_executor.submit(randomSleep, (var_num)) for var_num in range(5,10)]
var_start_time = time.time()for var_future in as_completed(var_all_tasks):print('result: {}'.format(var_future.result()))print('process io run time is {} s'.format(time.time() - var_start_time))if __name__ == '__main__':
threadPool()
processPool()
threadIoPool()
processIoPool()8、fork案例
import osimport timeprint('bobby')#fork只能用于linux/unix中,用于创建子进程,子进程会把父进程中的数据原样拷贝至子进程中,子进程会运行fork之后的代码var_pid = os.fork()print('bobby1')if var_pid == 0:print('子进程:{},父进程是:{}'.format(os.getpid(), os.getppid()))else:print('我是父进程:{}'.format(var_pid))
time.sleep(2)'''bobbybobby1我是父进程:951bobby1子进程:951,父进程是:950'''9、多进程使用
import multiprocessingimport timedef getSleep(var_n):
time.sleep(var_n)print('The child process ran successfully')return var_nif __name__ == '__main__':
var_process = multiprocessing.Process(target=getSleep, args=(2,))print(var_process.pid)
var_process.start()print(var_process.pid)
var_process.join()print('The main process runs successfully')10、进程池
import multiprocessingimport timedef getSleep(var_n):
time.sleep(var_n)print('The child process ran successfully')return var_nif __name__ == '__main__':#使用进程池 var_pool = multiprocessing.Pool(multiprocessing.cpu_count())''' var_result = var_pool.apply_async(getSleep, args=(3,)) #关闭pool不在接收新的任务 var_pool.close() #等待所有的任务完成 var_pool.join() #打印结果 print(var_result.get()) ''' ''' #imap方法,完成顺序和添加顺序一致 for var_one_result in var_pool.imap(getSleep, [3,2,1]): print(var_one_result) ''' #imap_unordered方法,先执行完成先打印 for var_one_result in var_pool.imap_unordered(getSleep, [3, 2, 1]):print(var_one_result)11、进程间通信Queue、Manager
import timefrom multiprocessing import Process, Queue, Pool, Managerdef producer(var_queue):
var_queue.put('a')
time.sleep(2)def consumer(var_queue):
time.sleep(2)
var_data = var_queue.get()print(var_data)if __name__ == '__main__':''' var_queue = Queue(10) var_producer = Process(target=producer, args=(var_queue,)) var_consumer = Process(target=consumer, args=(var_queue,)) var_producer.start() var_consumer.start() var_producer.join() var_consumer.join() ''' #queue不能用于进程池 ''' var_queue = Queue(10) var_pool = Pool(2) var_pool.apply_async(producer, args=(var_queue,)) var_pool.apply_async(consumer, args=(var_queue,)) var_pool.close() var_pool.join() ''' #Manager可以用于进程之间的通信 var_queue = Manager().Queue(10)
var_pool = Pool(2)
var_pool.apply_async(producer, args=(var_queue,))
var_pool.apply_async(consumer, args=(var_queue,))
var_pool.close()
var_pool.join()from multiprocessing import Manager , Processdef addDict(var_dict, var_key, var_value): var_dict[var_key] = var_valueif __name__ == '__main__': var_dict = Manager().dict() first_process = Process(target=addDict, args=(var_dict, 'a', 'a')) second_process = Process(target=addDict, args=(var_dict, 'b', 'b')) first_process.start() second_process.start() first_process.join() second_process.join()print(var_dict)
12、进程间的通信Pipe
#pipe只能用于两个指定的进程之间的通信,pipe性能高于queueimport timefrom multiprocessing import Process, Pipedef producer(var_pipe):
var_pipe.send('a')
time.sleep(2)def consumer(var_pipe):
time.sleep(2)
var_data = var_pipe.recv()print(var_data)if __name__ == '__main__':
var_recv, var_send = Pipe()
var_producer = Process(target=producer, args=(var_send,))
var_consumer = Process(target=consumer, args=(var_recv,))
var_producer.start()
var_consumer.start()
var_producer.join()
var_consumer.join()关于python中怎么实现多线程和多进程就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。
