Python中的多线程

什么是多线程:

  进程:正在运行的程序,QQ 360 ......

线程:就是进程中一条执行程序的执行路径,一个程序至少有一条执行路径。(360中的杀毒 电脑体检 电脑清理 同时运行的话就需要开启多条路径)

  每个线程都有自己需要运行的内容,而这些内容可以称为线程要执行的任务。

  开启多线程是为了同时运行多部分代码。

  好处:解决了多部分需要同时运行的问题

  弊端:如果线程过多,会导致效率很低(因为程序的执行都是CPU做着随机 快速切换来完成的)

  • 线程与进程的区别
  • 线程共享内存,进程独立内存
  • 线程启动速度块,进程启动速度慢,运行时速度没有可比性
  • 同一个进程的线程间可以直接交流,两个进程想通信,必须通过一个中间代理来实现
  • 创建新线程很简单,创建新进程需要对其父进程进行一次克隆
  • 一个线程可以控制和操作同一线程里的其他线程,但是进程只能操作子进程

threading模块

多线程的使用方式一:直接使用

# -*- coding:utf-8 -*-
# 线程使用的方式一
import threading
import time
# 需要多线程运行的函数
def fun(args):
    print("我是线程%s" % args)
    time.sleep(2)
    print("线程%s运行结束" % args)
# 创建线程
t1 = threading.Thread(target=fun, args=(1,))
t2 = threading.Thread(target=fun, args=(2,))
start_time = time.time()
t1.start()
t2.start()
end_time = time.time()
print("两个线程一共的运行时间为:", end_time-start_time)
print("主线程结束")
"""
运行结果:
我是线程1
我是线程2两个线程一共的运行时间为: 0.0010077953338623047
主线程结束
线程1运行结束
线程2运行结束
"""

线程的第二种使用方式:继承式调用

# 继承式调用
import threading
import time
class MyThreading(threading.Thread):
    def __init__(self, name):
        super(MyThreading, self).__init__()
        self.name = name
    # 线程要运行的代码
    def run(self):
        print("我是线程%s" % self.name)
        time.sleep(2)
        print("线程%s运行结束" % self.name)
t1 = MyThreading(1)
t2 = MyThreading(2)
start_time = time.time()
t1.start()
t2.start()
end_time = time.time()
print("两个线程一共的运行时间为:", end_time-start_time)
print("主线程结束")
"""
运行结果:
我是线程1
我是线程2
两个线程一共的运行时间为: 0.0010724067687988281
主线程结束
线程2运行结束
线程1运行结束
"""

守护线程与join方法

  • 在Python多线程编程中,join方法的作用式线程同步。
  • 守护线程,是为守护别人而存在的,当设置为守护线程后,被守护的主线程不存在后,守护线程也自然不存在。
  • 第一种:python多线程默认情况
  • Python多线程默认情况(设置线程setDaemon(False)),主线程执行完自己的任务后,就退出了,此时子线程会继续执行自己的任务,直到子线程任务结束
  • 代码演示:threading中的两个创建多线成的例子都是。
  • 第二种:开启守护线程
  • 开启线程的setDaemon(True)),设置子线程为守护线程,实现主程序结束,子程序立马全部结束功能
  • 代码演示:
# 守护线程
import threading
import time
class MyThreading(threading.Thread):
    def __init__(self, name):
        super(MyThreading, self).__init__()
        self.name = name
    # 线程要运行的代码
    def run(self):
        print("我是线程%s" % self.name)
        time.sleep(2)
        print("线程%s运行结束" % self.name)
t1 = MyThreading(1)
t2 = MyThreading(2)
start_time = time.time()
t1.setDaemon(True)
t1.start()
t2.setDaemon(True)
t2.start()
end_time = time.time()
print("两个线程一共的运行时间为:", end_time-start_time)
print("主线程结束")
  • 注意:如果要设置为守护线程,一定要在开启线程之前,将该线程设置为守护线程
  • 结论:主线程结束后,无论子线程1,2是否运行完成,都结束线程,不在继续向下运行
  • 第三种:加入join方法设置同步
  • 当不给程序设置守护进程时,主程序将一直等待子程序全部运行完成才结束
  • 代码演示:
# join:线程同步
import threading
import time
class MyThreading(threading.Thread):
    def __init__(self, name):
        super(MyThreading, self).__init__()
        self.name = name
    # 线程要运行的代码
    def run(self):
        print("我是线程%s" % self.name)
        time.sleep(3)
        print("线程%s运行结束" % self.name)
threading_list = []
start_time = time.time()
for x in range(50):
    t = MyThreading(x)
    t.start()
    threading_list.append(t)
for x in threading_list:
    x.join()    # 为线程开启同步
end_time = time.time()
print("50个线程一共的运行时间为:", end_time-start_time)
print("主线程结束")

结论:主线程等待50个子线程全部执行完成才结束。

线程锁(互斥锁Mutex)

  • 一个进程下可以启用多个线程,多个线程共享父进程的内存空间,也就意味着每个线程可以访问同一份数据,此时如果多个线程同时要修改一份数据,会出现什么状况?
  • 代码演示:
# -*- coding:utf8  -*-
import threading
import time
num = 100
threading_list = []
def fun():
    global num
    print("get num:", num)
    num += 1
    time.sleep(1)
for x in range(200):
    t = threading.Thread(target=fun)
    t.start()
    threading_list.append(t)
for x in threading_list:
    x.join()
print("nun:", num)
  • 结论:运行结果可能会出现num<300的情况
  • 正常来讲,这个num结果应该是300, 但在python 2.7上多运行几次,会发现,最后打印出来的num结果不总是300,为什么每次运行的结果不一样呢? 哈,很简单,假设你有A,B两个线程,此时都 要对num 进行加1操作, 由于2个线程是并发同时运行的,所以2个线程很有可能同时拿走了num=100这个初始变量交给cpu去运算,当A线程去处完的结果是101,但此时B线程运算完的结果也是101,两个线程同时CPU运算的结果再赋值给num变量后,结果就都是101。那怎么办呢? 很简单,每个线程在要修改公共数据时,为了避免自己在还没改完的时候别人也来修改此数据,可以给这个数据加一把锁, 这样其它线程想修改此数据时就必须等待你修改完毕并把锁释放掉后才能再访问此数据。

*注:不要在3.x上运行,不知为什么,3.x上的结果总是正确的,可能是自动加了锁

加锁版本:

import random
import threading
import time
num = 100
threading_list = []
def fun():
    global num
    time.sleep(random.random())
    lock.acquire() # 加锁
    print("get num:", num, threading.current_thread())
    num += 1
    lock.release()  # 释放锁
# 实例化锁对象
lock = threading.Lock()
for x in range(200):
    t = threading.Thread(target=fun)
    t.start()
    threading_list.append(t)
for x in threading_list:
    x.join()
print("num:", num)

GIL VS Lock

机智的同学可能会问到这个问题,就是既然你之前说过了,Python已经有一个GIL来保证同一时间只能有一个线程来执行了,为什么这里还需要lock? 注意啦,这里的lock是用户级的lock,跟那个GIL没关系 ,具体我们通过下图来看一下+配合我现场讲给大家,就明白了。

那你又问了, 既然用户程序已经自己有锁了,那为什么C python还需要GIL呢?加入GIL主要的原因是为了降低程序的开发的复杂度,比如现在的你写python不需要关心内存回收的问题,因为Python解释器帮你自动定期进行内存回收,你可以理解为python解释器里有一个独立的线程,每过一段时间它起wake up做一次全局轮询看看哪些内存数据是可以被清空的,此时你自己的程序 里的线程和 py解释器自己的线程是并发运行的,假设你的线程删除了一个变量,py解释器的垃圾回收线程在清空这个变量的过程中的clearing时刻,可能一个其它线程正好又重新给这个还没来及得清空的内存空间赋值了,结果就有可能新赋值的数据被删除了,为了解决类似的问题,python解释器简单粗暴的加了锁,即当一个线程运行时,其它人都不能动,这样就解决了上述的问题, 这可以说是Python早期版本的遗留问题。

RLock(递归锁)

说白了就是在一个大锁中还要再包含子锁

import threading, time
def run1():
    lock.acquire()
    print("grab the first part data")
    global num
    num += 1
    lock.release()
    return num
def run2():
    lock.acquire()
    print("grab the second part data")
    global num2
    num2 += 1
    lock.release()
    return num2
def run3():
    lock.acquire()
    res = run1()
    print('--------between run1 and run2-----')
    res2 = run2()
    lock.release()
    print(res, res2)
if __name__ == '__main__':
    num, num2 = 0, 0
    lock = threading.RLock()
    for i in range(3):
        t = threading.Thread(target=run3)
        t.start()
while threading.active_count() != 1:
    print(threading.active_count())
else:
    print('----all threads done---')
    print(num, num2)

在开发的过程中要注意有些操作默认都是 线程安全的(内部集成了锁的机制),我们在使用的时无需再通过锁再处理,例如:

import threading
data_list = []
lock_object = threading.RLock()
def task():
    print("开始")
    for i in range(1000000):
        data_list.append(i)
    print(len(data_list))
for i in range(2):
    t = threading.Thread(target=task)
    t.start()

Semaphore(信号量)

  • 互斥锁同时只允许一个线程修改数据,而Semaphore是同时允许一定数量的线程修改数据,比如厕所有三个坑,那最多只允许三个人上厕所,后面的人只能等前面的人出来才能进去。
  • 代码演示:
# -*- coding:GBK -*-
import threading
import time

sum_1 = 0
def run(i):
    global sum_1
    time.sleep(1)
    # lock.acquire()
    semaphore.acquire()
    sum_1 += 1
    print("线程%s来了,并修改了sum_1的值为:%s" % (i, sum_1))
    semaphore.release()
    # lock.release()

# lock = threading.Lock()
semaphore = threading.BoundedSemaphore(5)

for x in range(10):
    t = threading.Thread(target=run, args=(x,))
    t.start()

while threading.active_count() != 1:
    pass

print("程序结束")

Event(事件)

  • 通过Event来实现两个或多个线程间的交互,下面是一个红绿灯的例子,即起动一个线程做交通指挥灯,生成几个线程做车辆,车辆行驶按红灯停,绿灯行的规则。
  • 四个常用方法

set()  # 设置标志位为 True
clear()   # 清空标志位(将标志位改为false)
is_set()  # 检测标志位,如果标志位被设置,返回True,否则返回False
wait()   # 等待标志位被设置位True程序才继续往下运行

代码演示:

# -*- coding:utf-8 -*-
import threading
import time
def light():
    count = 1
    event.set()  # 设置标志位 True
    while True:
        if count <= 10:
            print("现在是绿灯")
            time.sleep(1)
        elif count <= 15:
            print("现在是红灯")
            event.clear()   # 清空标志位(将标志位改为false)
            time.sleep(1)
        else:
            count = 0
            event.set()
        count += 1
def car(name):
    while True:
        if event.is_set():
            print("----------%s在起飞-------------" % name)
            time.sleep(1)
        else:
            print("---------%s在等红灯---------------" % name)
            event.wait()   # 等待标志位被设置位True程序才继续往下运行
event = threading.Event()
light_1 = threading.Thread(target=light)
light_1.start()
for x in range(5):
    car_1 = threading.Thread(target=car, args=("马自达"+str(x),))
    car_1.start()

红绿灯案例

Queue(队列)

queue.Queue(maxsize=0)
#队列:先进先出  maxsize:设置队列的大小
queue.LifoQueue(maxsize=0)
##last in fisrt out  maxsize:设置队列的大小
queue.PriorityQueue(maxsize=0)
#存储数据时可设置优先级的队列,按优先级顺序(最低的先)  maxsize:设置队列的大小  

exceptionqueue.Empty

Exception raised when non-blockingget()(orget_nowait()) is called on aQueueobject which is empty.

当在一个空的队列对象上调用非阻塞的get()(或get_nowait())时,会产生异常。

exceptionqueue.Full

Exception raised when non-blockingput()(orput_nowait()) is called on aQueueobject which is full.

当非阻塞的put()(或put_nowait())被调用到一个已满的队列对象上时引发的异常。

import queue
# 实例化队列对象
q = queue.Queue(3)
print(q.qsize())    # 获取队列内数据的长度
print(q.empty())    # 如果队列是空的,返回True,否则返回False(不可靠!)
print(q.full())     # 如果队列已满,返回True,否则返回False(不可靠!)。
"""
Queue.put(item, block=True, timeout=None)
可以简写:Queue.put(item, True, 5)
将项目放入队列。
如果可选的args block为true(默认值),并且timeout为None(默认值),必要时进行阻塞,直到有空闲的槽。
如果timeout是一个正数,它最多阻断timeout秒,如果在这段时间内没有空闲槽,则引发Full异常。
否则(block为false),如果有空闲的槽立即可用,就在队列上放一个项目,否则就引发Full异常(在这种情况下忽略超时)。
"""
q.put(1)            # 数据“1”进入队列
q.put("nihao")      # 数据"nihao"进入队列
q.put("456ni", block=True, timeout=5)
'''
将一个项目放入队列中,不进行阻断。
只有在有空闲位置的情况下才排队。
否则会引发Full异常。
'''
# q.put_nowait(123)

'''
Queue.get(block=True, timeout=None)
可以简写:Queue.get(True, 3)
从队列中删除并返回一个项目。
如果可选的args'block'为True(默认),'timeout'为无(默认)。
    就会在必要时阻塞,直到有一个项目可用。
    如果'timeout'是非负数,它最多阻断'timeout'秒,如果在这段时间内没有项目可用,则引发Empty异常。
否则('block'为False),如果有一个项目立即可用,则返回一个项目。
    否则引发Empty异常('timeout'被忽略了在这种情况下)。
'''
print(q.get())
print(q.get())
print(q.get())
print(q.get(block=True, timeout=2))
'''
从队列中移除并返回一个项目,而不阻塞。
只有当一个项目立即可用时,才会得到一个项目。
否则引发Empty异常。
'''
# print(q.get_nowait())

生产者消费者模型

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。

为什么要使用生产者和消费者模式

在线程世界里,生产者就是生产数据的线程,消费者就是消费数据的线程。在多线程开发当中,如果生产者处理速度很快,而消费者处理速度很慢,那么生产者就必须等待消费者处理完,才能继续生产数据。同样的道理,如果消费者的处理能力大于生产者,那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

什么是生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯,而通过阻塞队列来进行通讯,所以生产者生产完数据之后不用等待消费者处理,直接扔给阻塞队列,消费者不找生产者要数据,而是直接从阻塞队列里取,阻塞队列就相当于一个缓冲区,平衡了生产者和消费者的处理能力。

# 生产者/消费者
import threading
import queue
import time
# 生产者
def producer(name):
    count = 1
    while True:
        p.put("{}骨头{}".format(name, count))
        print("骨头{}被{}生产".format(count, name).center(60, "*"))
        count += 1
        time.sleep(0.1)
# 消费者
def consumer(name):
    while True:
        print("{}被{}吃掉了".format(p.get(), name))
# 实例化队列对象
p = queue.Queue(10)
# 创建生产者线程
producer_threading1 = threading.Thread(target=producer, args=("飞某人",))
producer_threading2 = threading.Thread(target=producer, args=("Alex",))
# 创建消费者线程
consumer_threading1 = threading.Thread(target=consumer, args=("张三",))
consumer_threading2 = threading.Thread(target=consumer, args=("李四",))
producer_threading1.start()
producer_threading2.start()
consumer_threading1.start()
consumer_threading2.start()

线程池

Python3中官方才正式提供线程池。

线程不是开的越多越好,开的多了可能会导致系统的性能更低了,例如:如下的代码是不推荐在项目开发中编写。

import threading
def task(video_url):
    pass
url_list = ["www.xxxx-{}.com".format(i) for i in range(30000)]
for url in url_list:
    t = threading.Thread(target=task, args=(url,))
    t.start()
# 这种每次都创建一个线程去操作,创建任务的太多,线程就会特别多,可能效率反倒降低了。

建议:使用线程池

import time
from concurrent.futures import ThreadPoolExecutor  # 并行期货,线程池执行者
"""
pool = ThreadPoolExecutor(100)
pool.submit(函数名,参数1,参数2,参数...)
"""
def task(video_url, num):
    print("开始执行任务", video_url, num)     # 开始执行任务 www.xxxx-299.com 3
    time.sleep(1)
# 创建线程池,最多维护10个线程
threadpool = ThreadPoolExecutor(10)
# 生成300网址,并放入列表
url_list = ["www.xxxx-{}.com".format(i) for i in range(300)]
for url in url_list:
    """
    在线程池中提交一个任务,线程池如果有空闲线程,则分配一个线程去执行,执行完毕后在将线程交还给线程池,
    如果没有空闲线程,则等待。注意在等待时,与主线程无关,主线程依然在继续执行。
    """
    threadpool.submit(task, url, 3)
print("等待线程池中的任务执行完毕中······")
threadpool.shutdown(True)   # 等待线程池中的任务执行完毕后,在继续执行
print("END")

任务执行完任务,再干点其他事:

"""线程池的回调"""
import time
import random
from concurrent.futures import ThreadPoolExecutor
def task(video_url):
    print("开始执行任务", video_url)
    time.sleep(1)
    return random.randint(0, 10)    # 将结果封装成一个Futuer对象,返回给线程池
def done(response):     # response就是futuer对象,也就是task的返回值分装的一个Futuer对象
    print("任务执行完后,回调的函数", response.result())    # 即Futuer.result():取出task的返回值
# 创建线程池
threadpool = ThreadPoolExecutor(10)
url_list = ["www.xxxx-{}.com".format(i) for i in range(5)]
for url in url_list:
    futuer = threadpool.submit(task, url)    # futuer是由task返回的一个Future对象,里面有记录task的返回值
    futuer.add_done_callback(done)           # 回调done函数,执行者依然是子线程
# 优点:可以做分工,例如:task专门下载,done专门将下载的数据写入本地文件。
(0)

相关推荐

  • python中的多线程实例教程

    本文以实例形式较为详细的讲述了Python中多线程的用法,在Python程序设计中有着比较广泛的应用.分享给大家供大家参考之用.具体分析如下: python中关于多线程的操作可以使用thread和threading模块来实现,其中thread模块在Py3中已经改名为_thread,不再推荐使用.而threading模块是在thread之上进行了封装,也是推荐使用的多线程模块,本文主要基于threading模块进行介绍.在某些版本中thread模块可能不存在,要使用dump_threading来代

  • Python中的多线程实例(简单易懂)

    目录 1.python中显示当前线程信息的属性和方法 2.添加一个线程 3.线程中的join函数 4.使用Queue存储线程的结果 5.线程锁lock 前言: 多线程简单理解就是:一个CPU,也就是单核,将时间切成一片一片的,CPU轮转着去处理一件一件的事情,到了规定的时间片就处理下一件事情. 1.python中显示当前线程信息的属性和方法 # coding:utf-8 # 导入threading包 import threading if __name__ == "__main__":

  • 详解Python中的多线程编程

    一.简介 多线程编程技术可以实现代码并行性,优化处理能力,同时功能的更小划分可以使代码的可重用性更好.Python中threading和Queue模块可以用来实现多线程编程. 二.详解 1.线程和进程        进程(有时被称为重量级进程)是程序的一次执行.每个进程都有自己的地址空间.内存.数据栈以及其它记录其运行轨迹的辅助数据.操作系统管理在其上运行的所有进程,并为这些进程公平地分配时间.进程也可以通过fork和spawn操作来完成其它的任务,不过各个进程有自己的内存空间.数据栈等,所以只

  • Python中尝试多线程编程的一个简明例子

    综述     多线程是程序设计中的一个重要方面,尤其是在服务器Deamon程序方面.无论何种系统,线程调度的开销都比传统的进程要快得多.   Python可以方便地支持多线程.可以快速创建线程.互斥锁.信号量等等元素,支持线程读写同步互斥.美中不足的是,Python的运行在Python 虚拟机上,创建的多线程可能是虚拟的线程,需要由Python虚拟机来轮询调度,这大大降低了Python多线程的可用性.希望高版本的Python可以 解决这个问题,发挥多CPU的最大效率.   网上有些朋友说要获得真

  • python中的多线程锁lock=threading.Lock()使用方式

    目录 多线程锁lock=threading.Lock()使用 疑问 解决方法 例子 python多线程中锁的概念 锁可以独立提取出来 概念 线程不安全 线程锁 多线程锁lock=threading.Lock()使用 疑问 多线程任务是同时执行的,如果我们需要先执行线程a,再执行线程b,需要怎么办呢? 解决方法 使用python的多线程锁lock. 例子 未使用多线程锁lock: def a():     for i in range(3):         print('a%d' % (i +

  • Python中的多线程

    什么是多线程: 进程:正在运行的程序,QQ 360 ...... 线程:就是进程中一条执行程序的执行路径,一个程序至少有一条执行路径.(360中的杀毒 电脑体检 电脑清理 同时运行的话就需要开启多条路径) 每个线程都有自己需要运行的内容,而这些内容可以称为线程要执行的任务. 开启多线程是为了同时运行多部分代码. 好处:解决了多部分需要同时运行的问题 弊端:如果线程过多,会导致效率很低(因为程序的执行都是CPU做着随机 快速切换来完成的) 线程与进程的区别 线程共享内存,进程独立内存 线程启动速度

  • 深入理解 Python 中的多线程 新手必看

    示例1 我们将要请求五个不同的url: 单线程 import time import urllib2 defget_responses(): urls=[ 'http://www.baidu.com', 'http://www.amazon.com', 'http://www.ebay.com', 'http://www.alibaba.com', 'http://www.jb51.net' ] start=time.time() forurlinurls: printurl resp=urll

  • python中使用多线程改进flask案例

    目录 1.线程和进程关系? 2.多线程 3.多进程 前言: 线程是指进程内的一个执行单元,也是进程内的可调度实体. 与进程的区别: (1) 地址空间:进程内的一个执行单元;进程至少有一个线程;它们共享进程的地址空间;而进程有自己独立的地址空间; (2) 资源拥有:进程是资源分配和拥有的单位,同一个进程内的线程共享进程的资源 (3) 线程是处理器调度的基本单位,但进程不是. (4) 二者均可并发执行. 简而言之,一个程序至少有一个进程,一个进程至少有一个线程. 线程的划分尺度小于进程,使得多线程程

  • Python中多线程的创建及基本调用方法

    1. 多线程的作用 简而言之,多线程是并行处理相互独立的子任务,从而大幅度提高整个任务的效率. 2. Python中的多线程相关模块和方法 Python中提供几个用于多线程编程的模块,包括thread,threading和Queue等 thread模块提供了基本的线程和锁的支持,除产生线程外,也提供基本的同步数据结构锁对象,其中包括: start_new_thread(function, args kwargs=None)  产生一个新的线程来运行给定函数 allocate_lock()  分配

  • Python中实现输入超时及如何通过变量获取变量名

    背景介绍 开发中遇到了一个需求:程序运行到某处时需要用户确认, 但不能一直傻等, 后面的程序不能被一直阻塞, 需要有个超时限制, 也就是这个程序如果在一段时间后还没有得到用户输入就执行默认操作. 解决思路 – 多线程法 我就想到了用多线程的方式, 开启一个子线程用stdin(比如python的input函数)获取用户输入, 主线程里设置线程启动和超时. 创建线程 Python中使用多线程很方便, threading.Threaded(函数, 参数表)然后thread.start就好了. 只是有一

随机推荐