线程：可在计算机中使用cpu调度最小单元(真正在工作)。 流程：计算机资源分配的最小单元（流程为线程提供资源）。 #如容器，资源隔离使用的过程级别隔离。  一个过程中可以有多个线程，同一过程中的线程可以共享这个过程中的资源。 

• 一个程序，创建 三个进程，每个进程 一个线程(共3人)，并行处理任务。

  import time import requests import multiprocessing  # 创建过程后，在过程中创建一个线程。 # t = multiprocessing.Process(target=函数名, args=(name, url)) # t.start()            url_list = [     ("东北F4模仿秀.mp4", "https://aweme.snssdk.com/aweme/v1/playwm/?video_id=v0300f570000bvbmace0gvchlo53oog"),
      ("卡特扣篮.mp4", "https://aweme.snssdk.com/aweme/v1/playwm/?video_id=v0200f3e0000bv52fpn5t6p007e34q1g"),
      ("罗斯mvp.mp4", "https://aweme.snssdk.com/aweme/v1/playwm/?video_id=v0200f240000buuer5aa4tij4gv6ajqg")
  ]
  
  
  def task(file_name, video_url):
      res = requests.get(video_url)
      with open(file_name, mode='wb') as f:
          f.write(res.content)
      print(time.time())
  
  
  if __name__ == '__main__':
      print(time.time())
      for name, url in url_list:
          t = multiprocessing.Process(target=task, args=(name, url))
          t.start()
  

  多进程 的开销比 多线程 的开销大

GIL， 全局解释器锁（Global Interpreter Lock），是CPython解释器特有一个玩意，让一个进程中同一个时刻只能有一个线程可以被CPU调用。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-HAoziBlw-1641646024730)(assets/image-20220108111638498.png)]

如果程序想利用计算机多核优势,让CPU同时处理一些任务,适合用多进程开发(即使资源开销大)(计算操作多)

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-PrPRKwxg-1641646024732)(assets/image-20220108112656464.png)]

如果程序不利用 计算机的多核优势,适合用多线程开发(IO操作较多,计算操作少)

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-1PbHljDT-1641646024733)(assets/image-20220108112750341.png)]

在Python程序开发中,计算操作需要使用CPU多核优势,IO操作不需要利用CPU多核优势。

• 计算密集型，用多进程，例如：大量的数据计算【累加计算示例】。
• IO密集型，用多线程，例如：文件读写、网络数据传输【下载抖音视频示例】。

累加计算实例(计算密集型):

• 串行处理

import time

start = time.time()

result = 0
for i in range(100000000):
    result += i
print(result)

end = time.time()

print("耗时：", end - start) # 耗时： 9.40793490409851

• 多进程处理

import time
import multiprocessing

def task(start,end,queue):
    result = 0
    for i in range(start,end):
        result += 1
    queue.put(result)

if __name__ == '__main__':
    queue = multiprocessing.Queue()

    start_time = time.time()

    p1 = multiprocessing.Process(target=task,args=(0,50000000,queue))
    p1.start()

    p2 = multiprocessing.Process(target=task,args=(50000000, 100000000, queue))
    p2.start()

    v1 = queue.get(block=True)  #block=True 不会超时,会一直阻塞等待结果返回
    v2 = queue.get(block=True)

    print(v1+v2)

    end_time = time.time()

    print("耗时",end_time - start_time) #耗时 2.5452682971954346

多进程、多线程结合使用: 创建2个进程(建议于CPU个数相同),每个进程中创建3个线程。

import multiprocessing
import threading


def thread_task():
    pass


def task(start, end):
    t1 = threading.Thread(target=thread_task)
    t1.start()

    t2 = threading.Thread(target=thread_task)
    t2.start()

    t3 = threading.Thread(target=thread_task)
    t3.start()


if __name__ == '__main__':
    p1 = multiprocessing.Process(target=task, args=(0, 50000000))
    p1.start()

    p2 = multiprocessing.Process(target=task, args=(50000000, 100000000))
    p2.start()

import threading

def task(arg):
	print(arg)

# 创建一个Thread对象（线程），并封装线程被CPU调度时应该执行的任务和相关参数。
t = threading.Thread(target=task,args=('xxx',))
# 线程准备就绪（等待CPU调度），代码继续向下执行。
t.start()

print("继续执行...") # 默认主线程执行完所有代码，不结束（等待子线程）

线程常见的方法

• t.start() ,当前线程准备就绪(等待CPU调度,具体调度时间是由CPU来决定)。

import threading

loop = 10000000
number = 0

def _add(count):
    global number
    for i in range(count):
        number += 1

t = threading.Thread(target=_add,args=(loop,))
t.start()

print(number) #207703

• t.join(),等待当前线程的任务执行完毕后再向下继续执行。

import threading

number = 0

def _add():
    global number
    for i in range(10000000):
        number += 1

t = threading.Thread(target=_add)
t.start()

t.join() # 主线程等待中...

print(number)

import threading
number = 0
def _add():
    global number
    for i in range(10000000):
        number += 1
def _sub():
    global number
    for i in range(10000000):
        number -= 1

t1 = threading.Thread(target=_add)
t2 = threading.Thread(target=_sub)
t1.start()
t1.join()  # t1线程执行完毕,才继续往后走
t2.start()
t2.join()  # t2线程执行完毕,才继续往后走

print(number) #0

import threading

loop = 10000000
number = 0


def _add(count):
    global number
    for i in range(count):
        number += 1


def _sub(count):
    global number
    for i in range(count):
        number -= 1


t1 = threading.Thread(target=_add, args=(loop,))
t2 = threading.Thread(target=_sub, args=(loop,))
t1.start()
t2.start()

t1.join()  # t1线程执行完毕,才继续往后走
t2.join()  # t2线程执行完毕,才继续往后走

print(number)  

• t.setDaemon(布尔值), 守护线程(必须放在start之前)
  • t.setDaemon(True), 设置为守护线程,主线程执行完毕后,子线程也自动关闭(不管有没有执行完子线程任务)
  • t.setDaemon(False),设置为非守护线程,主线程等待子线程,子线程执行完毕后,主线程才结束。(默认)
  • python3.10中setDaemon已经启用,使用 t.daemon = True 来进行设置。

import threading
import time

def task(arg):
    time.sleep(5)
    print('任务')

t = threading.Thread(target=task, args=(11,))
t.daemon = True # True/False
t.start()

print('END')

• 线程名称的设置和获取

  • 获取线程名称:

    • 3.10 之前版本 threading.current_thread().getName()

    • 3.10 threading.current_thread().name

  • 设置线程名称:

    • 3.10 之前版本 t.setName(‘日魔-{}’.format(i))
    • 3.10 t.name = “测试线程{}”.format(i)

import threading

def task(arg):
    # 获取当前执行此代码的线程
    name = threading.current_thread().name
    print(name)

for i in range(10):
    t = threading.Thread(target=task, args=(11,))
    t.name = "测试线程{}".format(i)
    t.start()

一个进程中可以有多个线程,且线程共享所有进程中的资源。

多个线程同时去操作一个“东西”,可能 会存在数据混乱的情况,例如:

• 示例1

import threading

loop = 10000000
number = 0


def _add(count):
    global number
    for i in range(count):
        number += 1


def _sub(count):
    global number
    for i in range(count):
        number -= 1


t1 = threading.Thread(target=_add, args=(loop,))
t2 = threading.Thread(target=_sub, args=(loop,))
t1.start()
t2.start()

t1.join()  # t1线程执行完毕,才继续往后走
t2.join()  # t2线程执行完毕,才继续往后走

print(number)

import threading

lock_object = threading.RLock()

loop = 10000000
number = 0

def _add(count):
    lock_object.acquire() # 加锁
    global number
    for i in range(count):
        number += 1
    lock_object.release() # 释放锁


def _sub(count):
    lock_object.acquire() # 申请锁（等待）
    global number
    for i in range(count):
        number -= 1
    lock_object.release() # 释放锁


t1 = threading.Thread(target=_add, args=(loop,))
t2 = threading.Thread(target=_sub, args=(loop,))
t1.start()
t2.start()

t1.join()  # t1线程执行完毕,才继续往后走
t2.join()  # t2线程执行完毕,才继续往后走

print(number) #0

• 示例2:

import threading

num = 0

def task():
    global num
    for i in range(1000000):
        num += 1
    print(num)


for i in range(2):
    t = threading.Thread(target=task)
    t.start()   # 1469983
                # 2000000

import threading

num = 0
lock_object = threading.RLock()


def task():
    print("开始")
    lock_object.acquire()  # 第1个抵达的线程进入并上锁，其他线程就需要再此等待。
    global num
    for i in range(1000000):
        num += 1
    lock_object.release()  # 线程出去，并解开锁，其他线程就可以进入并执行了
    print(num)


for i in range(2):
    t = threading.Thread(target=task)
    t.start()   #1000000 2000000

import threading

num = 0
lock_object = threading.RLock()


def task():
    print("开始")
    with lock_object: # 基于上下文管理，内部自动执行 acquire 和 release
        global num
        for i in range(1000000):
            num += 1
    print(num)


for i in range(2):
    t = threading.Thread(target=task)
    t.start() #1000000 2000000

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

import threading

data_list = []
th_list = []

lock_object = threading.RLock()

def task():
    print("开始")
    for i in range(1000000):
        data_list.append(i)


for i in range(2):
    t = threading.Thread(target=task)
    t.start()
    th_list.append(t)

for i in th_list:
    i.join()
print(len(data_list))

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-eG5OeI1Z-1641646024734)(assets/image-20220108124727407.png)]

所以,要多注意看一些开发文档中是否标明线程安全。

在程序中如果想要自己手动加锁,一般有两种: Lock和RLock。

• Lock,同步锁

import threading

num = 0
lock_object = threading.Lock()


def task():
    print("开始")
    lock_object.acquire()  # 第1个抵达的线程进入并上锁，其他线程就需要再此等待。
    global num
    for i in range(1000000):
        num += 1
    lock_object.release()  # 线程出去，并解开锁，其他线程就可以进入并执行了
    
    print(num)


for i in range(2):
    t = threading.Thread(target=task)
    t.start()

• Rlock,递归锁

import threading

num = 0
lock_object = threading.RLock()


def task():
    print("开始")
    lock_object.acquire()  # 第1个抵达的线程进入并上锁，其他线程就需要再此等待。
    global num
    for i in range(1000000):
        num += 1
    lock_object.release()  # 线程出去，并解开锁，其他线程就可以进入并执行了
    print(num)


for i in range(2):
    t = threading.Thread(target=task)
    t.start()

RLock支持多次申请锁和多次释放,Lock不支持,例如:

import threading
import time

lock_object = threading.Lock()


def task():
    print("开始")
    lock_object.acquire()
    lock_object.acquire()  #会一直卡在这里,等待第一把锁释放
    print(123)
    lock_object.release()
    lock_object.release()


for i in range(3):
    t = threading.Thread(target=task)
    t.start()  

import threading

lock = threading.RLock()


# 程序员A开发了一个函数，函数可以被其他开发者调用，内部需要基于锁保证数据安全。
def func():
    with lock:
        pass


# 程序员B开发了一个函数，可以直接调用这个函数。
def run():
    print("其他功能")
    func()  # 调用程序员A写的func函数，内部用到了锁。
    print("其他功能")


# 程序员C开发了一个函数，自己需要加锁，同时也需要调用func函数。
def process():
    with lock:
        print("其他功能")
    func()  # ----------------> 此时就会出现多次锁的情况，只有RLock支持（Lock不支持）。
    print("其他功能")

死锁，由于竞争资源或者由于彼此通信而造成的一种阻塞的现象。

import threading

num = 0
lock_object = threading.Lock()


def task():
    print("开始")
    lock_object.acquire()  # 第1个抵达的线程进入并上锁，其他线程就需要再此等待。
    lock_object.acquire()  # 第1个抵达的线程进入并上锁，其他线程就需要再此等待。
    global num
    for i in range(1000000):
        num += 1
    lock_object.release()  # 线程出去，并解开锁，其他线程就可以进入并执行了
    lock_object.release()  # 线程出去，并解开锁，其他线程就可以进入并执行了

    print(num)


for i in range(2):
    t = threading.Thread(target=task)
    t.start()