后续很快就要进入Spark的学习,在学习之前也是要铺垫一个知识点:进程是什么?线程是什么?两者有什么区别?又有什么关系?
import time
import os
import multiprocessing
# os.getpid():process id,获取进程的编号
# os.getppid():parent process id,获取父进程编号
# 跳舞函数
def dance():
print(f'dance进程的编号:{os.getpid()},父进程编号:{os.getppid()}')
for i in range(5):
print('跳舞中...')
# 休眠1秒钟
time.sleep(1)
# 唱歌函数
def sing():
print(f'sing进程的编号:{os.getpid()},父进程编号:{os.getppid()}')
for i in range(5):
print('唱歌中...')
# 休眠1秒钟
time.sleep(1)
if __name__ == '__main__':
# 最原始的父进程是由操作系统提供的
print(f'主进程的编号为:{os.getpid()},父进程编号:{os.getppid()}')
# 创建两个进程,一个执行 dance 函数,另一个执行 sing 函数
dance_process = multiprocessing.Process(target=dance)
sing_process = multiprocessing.Process(target=sing)
# 启动进程
dance_process.start()
sing_process.start()
import multiprocessing
import time
# 带有参数的任务(函数)
def task(count):
for i in range(count):
print('任务执行中...')
time.sleep(0.2)
else:
print('任务执行完成')
if __name__ == '__main__':
# 传参方式1 args=(参数值1, ...)
# task_process = multiprocessing.Process(target=task, args=(3,))
# 传参方式2 kwargs={'形参字符串1': 值1, '形参字符串2': 值2, ...}
task_process = multiprocessing.Process(target=task, kwargs={'count': 5})
# 启动进程
task_process.start()
task_process.join()
print('这行代码一定等task_process进程执行完毕之后才会运行,打印')
import multiprocessing
import time
# 定义全局变量
g_list = []
# 添加数据的函数
def add_data():
for i in range(5):
g_list.append(i)
print('add:', i)
time.sleep(0.2)
print('add:', g_list)
# 读取数据的函数
def read_data():
print('read:', g_list)
if __name__ == '__main__':
# 创建添加数据的子进程
add_data_process = multiprocessing.Process(target=add_data)
# 创建读取数据的子进程
read_data_process = multiprocessing.Process(target=read_data)
# 启动添加数据子进程
add_data_process.start()
# 阻塞等待:主进程等待 add_data_process 执行完成,再向下继续执行
add_data_process.join()
# 启动读取数据子进程
read_data_process.start()
# 主进程读取数据
print('main:', g_list)
# 主进程延时 1s
time.sleep(1)
print('主进程结束!')
主进程结束,设为守护进程的子进程主动结束
import multiprocessing
import time
def task():
for i in range(10):
print('任务执行中...')
time.sleep(0.5)
if __name__ == '__main__':
# 创建子进程并启动
sub_process = multiprocessing.Process(target=task)
# TODO:设置子进程为守护进程
sub_process.daemon = True
sub_process.start()
# 主进程延时 1s
time.sleep(1)
print('主进程结束!')
子进程被动结束
import multiprocessing
import time
# 任务函数
def task():
for i in range(10):
print('任务执行中...')
time.sleep(0.5)
if __name__ == '__main__':
# 创建子进程并启动
sub_process = multiprocessing.Process(target=task)
sub_process.start()
# 主进程延时 1s
time.sleep(1)
print('主进程结束!')
# TODO: 终止子进程
sub_process.terminate()
import time
import threading
# 跳舞函数
def dance(num):
for i in range(num):
print('跳舞中...')
time.sleep(1)
# 唱歌函数
def sing(num):
for i in range(num):
print('唱歌中...')
time.sleep(1)
if __name__ == '__main__':
# 创建两个线程,分别执行 dance 和 sing
dance_thread = threading.Thread(target=dance, args=(5,))
sing_thread = threading.Thread(target=sing, kwargs={'num':5})
# 启动线程
dance_thread.start()
sing_thread.start()
import threading
import time
# 定义全局变量
g_list = []
# 添加数据的函数
def add_data():
for i in range(5):
g_list.append(i)
print('add:', i)
time.sleep(0.2)
print('add:', g_list)
# 读取数据的函数
def read_data():
print('read:', g_list)
if __name__ == '__main__':
# 创建添加数据的子线程
add_data_thread = threading.Thread(target=add_data)
# 创建读取数据的子线程
read_data_thread = threading.Thread(target=read_data)
# 启动添加数据子线程
add_data_thread.start()
# 阻塞等待:主线程等待 add_data_thread 执行完成,再向下继续执行
add_data_thread.join()
# 启动读取数据子线程
read_data_thread.start()
# 阻塞等待:主线程等待 read_data_thread 执行完成,再向下继续执行
read_data_thread.join()
print('main:', g_list)
join()使用总结:
1- 作用:线程等待子线程执行结束
2- 使用场景:主线程需要使用多个子线程运行的最终结果,才能够继续往下运行。
注释lock锁相关代码,并使用 低版本的Python解释器(python3.6.5) 才能看到线程资源安全问题的效果!Python 3.10 已经解决了该问题!
# 互斥锁:多个线程去抢同一把"锁",抢到锁的线程执行,没抢到锁的线程会阻塞等待
import threading
# 定义全局变量
g_num = 0
# 创建一个多线程互斥锁
lock = threading.Lock()
def sum_num1():
global g_num
# 循环一次给全局变量加1
for i in range(1000000):
# 抢到锁,代码可以继续向下执行,否则就会阻塞等待
lock.acquire() # 抢锁
g_num += 1
lock.release() # 释放锁
print('sum1:', g_num)
def sum_num2():
global g_num
# 循环一次给全局变量加1
for i in range(1000000):
# 抢到锁,代码可以继续向下执行,否则就会阻塞等待
lock.acquire() # 抢锁
g_num += 1
lock.release() # 释放锁
print('sum2:', g_num)
if __name__ == '__main__':
# 创建两个线程
first_thread = threading.Thread(target=sum_num1)
second_thread = threading.Thread(target=sum_num2)
# 启动两个线程
first_thread.start()
second_thread.start()
# 阻塞等待:主线程等待子线程结束再向下运行
first_thread.join()
second_thread.join()
print(g_num)
线程的资源共享问题(只存在于低版本的Python解释器比如3.6.5,3.10已解决):当多个线程同时操作同一个共享的全局变量时,可能会造成错误的结果,解决办法如下
thread.join()
: