并发编程

提升代码的执行效率
串行 并发和并行
串行 多个人去排队按照先后顺序逐一去执行
并发 同时运行多个任务，比如有多个任务，只有一个cpu，那么在同一时刻只能处理一个任务，为了避免串行，可以让任务切换运行（每个任务执行一点再切换），达到并发的效果（看似再同时运行）如协程 多线程
并行 假设有多个任务，有多个cpu，那么同一时刻每个cpu都执行一个任务，真正的同时运行任务，如多进程

线程

线程 是计算机中可以被cpu调度的最小单元 （真正工作的单位 ）
进程 是计算机资源分配的最小单元为线程提供资源 （进程默认会创建一个线程）  进程只是一个资源单元或者说是资源集合

import threading #线程模块
t = threading.Thread(target = '函数',args = (11,22,33)#传如函数的参数 )
t.start() #开始线程

import time
import request
import threading
t = threading.Thread(target = '函数'，args=(1,2,3))
url_list = [
    (XXX,url),
    (XXX,url),
]
def task(filename,video_url)
    res = requests.get(video_url)
    with open(file_name,mode='wb') as f:
        f.write(res.content)
    print(time.time())
for name,url in url_list:
    #创建线程，让每个线程都去执行task函数（参数不同）
    t = threading.Thread(traget = task,args = (name,url))
    t.start()
#执行程序的线程为主线程，在下载的时候不会等待继续往下执行代码，执行完之后等待子线程，子线程完成后程序才完成

多线程开发

import threading
def task(arg)
    pass
#创建一个thread对象（线程）并封装线程被cpu调用时应该执行的任务和相关参数
t = threading.Thread(targ=task,args=('XXX',))
#线程准备就绪等待cpu调度，代码继续象下执行
t.start()
print('继续执行') #主线程执行完所有代码，不结束等待子线程

线程的常见方法

t.start() 当前线程准备就绪，等待cpu调度，具体的时间由cpu决定

import threading
loop = 100000
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)

t.join() 让主线程等待当前子线程的任务执行完毕后再继续往下执行

import threading
loop = 100000
number = 0
def _add (count):
    global number
    for i in range(count):
        number +=1
t = threading.Thread(target=_add,args=(loop,))
t.start()
t.join() # 主线程等待t线程执行完后才继续往下走 （主线程从上往下执行代码）
print(number)

import threading
number = 0
def _add ():
    global number
    for i in range(10000):
        number +=1
def _sub():
    global number
    for i in range(10000):
        number -=1
t = threading.Thread(target=_add)
t1 = threading.Thread(target=_sub)
t.start()
t.join() # t线程执行完毕，才继续往下走
         #主线程卡在这里，当前只有t线程可以被cpu调度，因为t线程执行了start(),当t线程执行完了，主线程才继续往下走
t1.start()
t1.join() #t1线程执行完毕，才继续往下走
print(number)

import threading
loop = 10000
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()  #cpu在执行线程的时候不会一直让一个线程运行，而是切换着运行，这样就导致数据可能不准确
t1.join() #t1线程执行完毕之后，才继续望后走
t2.join() #t1线程执行完毕之后，才继续望后走
print(number)

t.setDaemon(布尔值)，守护线程 必须放在start之前

t.setDaemon(True),设置为守护线程，主线程执行完毕后，子线程也自动关闭

t.setDaemon(False),设置为非守护线程，主线程等待子线程，字线程执行完毕后，主线程才结束（默认）

import threading
import time
def task(arg):
    time.sleep(5)
    print("任务")
t = threading.Thread(target= task,args = (11,))
t.setDaemon(True)
t.start()
print('end')

线程名称的设置和获取

import threading
def task(arg):
    #获取当前执行此代码的线程的名称
    name = threading.current_thread().getName()
    print(name)
for i in range(10):
    t = threading.Thread(target=task.args=(11,))
    t.setName('name_{}'.formait(i)) #设置线程的名字需要在start()之前
    t.start()

自定义线程类，直接将线程需要做的事情写到run方法中

import threading
class Mythread(threading.Thread)：
    def run(slef):
        print('执行此线程'，self._args)
t = Mythread(args= (100,))
t.start()
# 执行线程其实就是执行线程中的run方法

import requests
import threading
class DouyintThread(thread.Thread):
    def run(self):
        file_name,video_url=self._args
        res = request.get(video_url)
        with open(file_name,mode='wb') as f:
            f.write(res.content)
url_list = [(文件名，网址)，(文件名，网址)，(文件名，网址)]
for item in url_list :
    t = DouyinThread(args=(item[0],item[1]))
    t.start()

线程安全

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

多个线程同时去操作一个东西，可能会存在数据混乱的情况。这种情况叫做线程不安全

可以通过锁来解决不安全，用锁来控制其他线程的运行，只有申请到锁的才可以运行，没有申请到的等待

#不安全情况
import threading
loop = 10000
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
##cpu在执行线程的时候不会一直让一个线程运行，而是切换着运行，这样就导致数据可能不准确,
#如t1线程再相加的时候可能还没完成相加操作就切换到t2线程了,再切回t1线程的时候就继续往下相加了
t1= reading.Thread(target=_add,args=(loop,))
t2= reading.Thread(target=_sub,args=(loop,)) 
t1.start()
t2.start()

import threading
lock_object = threading.Rlock() #创建锁
loop = 10000
number = 0
def _add(count):
    look_object.acquire() #加锁或者叫申请锁  加锁之前先申请锁，谁先来的谁先申请，申请后就加锁 继续往下执行
    global = number
    for i in range(count):
        number+=1
    look.object.release()#释放锁
def _sub(count):
    look_object.acquire() #此时在等等着申请锁，只有前一个释放了才可以申请到，两个线程用用一把锁才有意义  没有申请到锁的就等待者，只有前一把锁释放了才可以申请到锁
    global number
    for i in range(count):
        number-=1
    look.object.release()#解锁 
t1= reading.Thread(target=_add,args=(loop,))
t2= reading.Thread(target=_sub,args=(loop,)) 
t1.start()
t2.start()

#数据混乱情况
import threading
num = 0
def task():
    global = num
    for i in range(10000):
        num+=1
    print(num)
for i in range(2)
    t = threading.Thread(target=task)
    t.start()
================================================================================
#解决方法
import threading
num = 0
lock_object= threading.Rlock()
def task():
    lock_object.aceqire() #第一个到达的线程先来到申请并加锁，其他线程就需要等到申请锁的过程
    global = num
    for i in range(10000):
        num+=1
    lock_object.release() #线程出去，并释放锁，其他线程就可以进入并执行了
    print(num)
for i in range(2)
    t = threading.Thread(target=task)
    t.start()

加锁释放锁支持上下文管理 with
import threading
num = 0
lock_object = threading.RLOCK()
def task():
    print('开始')
    with lock_object: #基于上下文管理，内部自动执行acquire 和 release
        global = num
        for i in range(10000):
            num+=1
        print('结束')
for i in range(2):
    t = threading.Thread(target=task)
    t.start()

在开发过程中有些操作默认都是线程安全的（内部集成了锁的机制），使用的时候不用再使用锁的进制

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

线程锁

程序中自己手动加锁 有两种Lock和Rlock

Lock 同步锁 不支持嵌套

import threading
num = 0
lock_object = threading.Lock()
def task():
    print('开始')
    lock_object.acquire() #第一个抵达的线程进入并上锁，其他线程就需要等待
    global num
    for i in range(1000):
        num+=1
    lock_object.release() #线程出去，并解开锁，其他线程就可以进入并执行了
    print(num)
for i in range(2):
    t = threading.Thread(target=task)
    t.start()

Rlock 递归锁 支持多次申请锁和多次解锁支持嵌套

内部有计数器每acquire一次计数器+1,只要计数不为0就不能被其他线程acquire

import threading
num = 0
lock_object = threading.Rlock()
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_object = threading.Rlock()
#A开发了一个函数，函数可以被其他人调用，内部需要基于锁保证线程安全
def func():
    with lock:
        pass
#B开发了一个函数，以可直接调用这个函数
def run():
    print('其他')
    func()  # 调用A开发的函数，内部用到了锁
    print('outher')
#c开发了函数需要自己加锁，同事也需求要调用func函数
def process():
    with lock:
        print('other')
        func()  # 此时就会出现多次锁的情况只有Rlock支持。Lock不支持
        print('other')

死锁

由于竞争资源或者彼此通信做成的一种阻塞现象

import threading
num = 0
lock_object = threading.Lock()
def task():
    print('开始')
    lock_object.acquire() #第一个线程申请并加锁
    print('00001')
    lock_object.acquire()#此时第一个线程卡在此处 
    global num 
    for i in range(1000):
        num +=1
    lock_object.release()
    lock_object.release()
for i in range(3):
    t = threading.Thread(target=task)
    t.start()
=============================================================
锁的资源互相竞争
import threading
import time
lock1 = threading.Lock()
lock2 = threading.Lock()
def task1():
    lock1.acquire() # t1申请锁
    time.sleep(1)
    lock2.acquire() # 此时t1申请lock2的锁因为lock2锁已经被t2申请走 卡住
    print(11)
    lock2.release()
     print(111)
    lock1.release()
     print(1111)
def task2():
    lock2.acquire() # t2申请锁
    time.sleep(1)
    lock1.acquire() # 此时t2申请lock1的锁因为lock1锁已经被t1申请走 卡住
    print(22)
    lock1.release()
     print(222)
    lock2.release()
     print(2222)
t1 = threading.Thread(target=task1)
t1.start()
t2 = threading.Thread(target=task2)
t2.start()

线程池

python3中才正式提供线程池

线程不是开的越多越好，开的多了可能会导致系统的性能更低了

不建议无限制的创建线程

import conucurrent.futures import ThreadPoolExectuor
def task(video_url,num):
    print('开始'，video_url)
    time.sleep(5)
#创建一个线程池，最多维护10个线程
pool = ThreadPoolExecutor(10)
url_list = ['www.xxx{}.com'.format(i) for i in range(100)]
for url in url_list:  #向线程池中提交100个任务，线程池去执行
    #在线程池中提交一个任务，线程池中有空闲的线程则分配一个去执行，执行完毕后将线程交还给线程池，如果没有空闲线程则等待
    pool.submit(task,url,2)# 异步调用,把任务全部提交
#主线程等待线程池中的任务执行完毕 然后结束

import conucurrent.futures import ThreadPoolExectuor
def task(video_url):
    print('开始'，video_url)
    time.sleep(5)
#创建线程池，最多维护10个线程
pool = ThreadPoolExecutor(10)
url_list = ['www.xxx{}.com'.format(i) for i in range(100)]
for url in url_list:  #向线程池中提交100个任务，线程池去执行
    #在线程池中提交一个任务，线程池中有空闲的线程则分配一个去执行，执行完毕后将线程交还给线程池，如果没有空闲线程则等待
    pool.submit(task,url)#函数  参数
print('执行中')
pool.shutdowm(wait=True) #等待线程池中的所有任务执行完毕后，再继续执行
print('继续往下走')

回调函数add_done_callback()

#该线程执行完任务后再干点别的事情
import time
import random
import concurrent.futures import ThreadPoolExectuor,tuture
def task(video_url):
    print('开始'，video_url)
    time.sleep(5)
    return random.randit(0,10)
def done(reaponse):
    print('任务执行完后的返回值'，response.result())#result可以拿到
#创建线程池，最多维护10个线程
pool = ThreadPoolExecutor(10)
url_list = ['www.xxx{}.com'.format(i) for i in range(100)]
for url in url_list:  
    future=pool.submit(task,url)
    future.add_done_callback(done)# 传入函数名 #是子主线程执行
    # pool.submit(task,url)任务执行完毕,return返回值后就会执行回调用函数
    #把任务交给线程池后会返回一个特殊的对象,这个对象不是结果，通过该对象可以调用add_done_callback函数,是线程中的任务执行完后,再执行一下done函数
#把任务交给线程池后会返回一个特殊的对象,通过对象.result()可以拿到返回值
#可以用task专门下载，done专门做下载的数据写入本地
#第一个线程执行task后会有一个返回值，这个线程再去执行done的时候会把返回值封装进response，通过result()可以拿到最终的结果

#统一获取结果
import time
import random
import concurrent.futures import ThreadPoolExectuor,tuture
def task(video_url):
    print('开始'，video_url)
    time.sleep(5)
    return random.randit(0,10)
pool = ThreadPoolExecutor(10)
future_list = []
for url in url_list: 
    future=pool.submit(task,url)
    future_list.append(future)
pool.shutdown(True)
for i in future_list:
    print(i.result())

线程queue

队列：先进先出

import queue
q=queue.Queue()
q.put('first')
q.put('second')
q.put('third')
q.put(4,block=True,timout=3)# 队列阻塞的时候阻塞3秒,如果3秒后还是阻塞则抛出异常
print(q.get())
print(q.get())
print(q.get())
'''
结果(先进先出):
first
second
third
'''

堆栈：后进先出

import queue
q=queue.LifoQueue()
q.put('first')
q.put('second')
q.put('third')
print(q.get())
print(q.get())
print(q.get())
'''
结果(后进先出):
third
second
first
'''

优先级队列：存储数据时可设置优先级的队列

import queue
q=queue.PriorityQueue()
#put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高,第二个元素是存入的数据
q.put((20,'a'))
q.put((10,'b'))
q.put((30,'c'))
print(q.get())
print(q.get())
print(q.get())
'''
结果(数字越小优先级越高,优先级高的优先出队):
(10, 'b')
(20, 'a')
(30, 'c')
'''

event事件

线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行
例如，有多个工作线程尝试链接MySQL，我们想要在链接前确保MySQL服务正常才让那些工作线程去连接MySQL服务器，如果连接不成功，都会去尝试重新连接。那么我们就可以采用threading.Event机制来协调各个工作线程的连接操作

from threading import Thread,Event
def student(name):
    print('学生在听课')
    event.wait()
    print('学生课间活动')
def teacher(name):
    print('老师在授课')
    event.set()
if __name__ == '__main__':
    stu1=Thread(target=student,args=('l1'))
    stu2=Thread(target=student,args=('l2'))
    stu3=Thread(target=student,args=('l3'))
    t1=Thread(target=teacher,args=('l4'))
    stu1.start()
    stu2.start()
    stu3.start()
    t1.start()

event.isSet()：返回event的状态值；
event.wait(参数time)：如果 event.isSet()==False将阻塞线程,超过参数就继续往下运行   等待 time 时间后，执行下一步。或者在调用 event.set() 后立即执行下一步。
event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
event.clear()：恢复event的状态值为False。

信号量

信号量也是一把锁，可以指定信号量为5，对比互斥锁同一时间只能有一个任务抢到锁去执行，信号量同一时间可以有5个任务拿到锁去执行，如果说互斥锁是合租房屋的人去抢一个厕所，那么信号量就相当于一群路人争抢公共厕所，公共厕所有多个坑位，这意味着同一时间可以有多个人上公共厕所，但公共厕所容纳的人数是一定的，这便是信号量的大小.
Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。

from threading import Thread,Semaphore
import threading
import time
def func():
    sm.acquire()
    print('%s get sm' %threading.current_thread().getName())
    time.sleep(3)
    sm.release()
if __name__ == '__main__':
    sm=Semaphore(5)
    for i in range(23):
        t=Thread(target=func)
        t.start()

定时器

定时器，指定n秒后执行某操作

from threading import Timer
def hello(n):
    print("hello, world")
t = Timer(1, hello,args=(12)) # 1秒之后执行hello
t.start()  # after 1 seconds, "hello, world" will be printed

进程

进程中默认会创建一个线程（主线程）进程与进程之间的资源是隔离的

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

python中通过多进程可以利用cpu多核的优势，计算密集型操作适用与多进程

t =  multiprocessing.Process(target = 函数名，args = (1,2,3))#创建进程
t.start()
#使用多进程的时候需要放在 if__name__ = '__main__' 下面
#linux基于fork windows基于spawn  mac基于fork和spawn (python3.8之后是spawn)
#multiprocessing.set_start_method('fork') 模式修改为fork就可以不用放在 if __name__ = '__main__'下面了
os.getpid() # 查看当前进程的id
os.getppid() # 查看父进程的id

多进程开发

进程是计算机中资源分配的最小单元，一个进程中可以有多个线程，同一个进程中的线程共享进程的资源

进程与进程之间则是相互隔离的

from multoprocessing import Process
def task():
    pss
def run():
    t =  multiprocessing.Process(target = 函数名，args = (1,2,3))#创建进程
    t.start() # 告诉操作系统开通一个子进程,具体什么时候开通这个进程是操作系统决定的
if__name__ = '__main__' :
    run() #不是说把进程的代码放在if__name__ = '__main__' ，而是main启动就好了

进程模式

通过multiprocessing操作进程有3中模式

fork 拷贝几乎所有资源 支持文件对象、线程锁等传参 unix系统 进程代码任意位置

#通过一个进程创建一个子进程的时候，子进程会拷贝当前进程的资源
#特点，通过父进程创建的子进程几乎会把父进程中的资源拷贝一份
import multiprocessing
def task():
    print(name)
if __name__ ='__main__':
    multiprocessing.set_start_method('fork') #创建进程的模式改为fork
    name = []
    p1 = multiprocessing.Process(target=task) #创建子进程去运行task函数，子进程拷贝父进程的数据以及父进程中文件对象和锁的当前状态 ，子进程和主进程的数据是两份数据各自维护自己的
    p1.start() 
=================================================================
import multiprocessing
def task():
    print(name)
if __name__ ='__main__':
    multiprocessing.set_start_method('fork') #创建进程的模式改为fork
    name = []
    name.append(123)
    p1 = multiprocessing.Process(target=task) #创建子进程去运行task函数，子进程拷贝父进程的数据 ，子进程和主进程的数据是两份数据各自维护自己的
    p1.start() 
    print(name)
=================================================================
import multiprocessing
def task():
    print(name)
if __name__ ='__main__':
    multiprocessing.set_start_method('fork') #创建进程的模式改为fork
    name = []   
    p1 = multiprocessing.Process(target=task) #创建子进程去运行task函数，子进程拷贝父进程的数据 ，子进程和主进程的数据是两份数据各自维护自己的
    p1.start() 
    name.append(123)
    print(name)
==============================================================================
#文件对象和线程锁可以直接拷贝也可以传参拷贝
import multiprocessing
def task():
    print(name)
    file_object.write('kill\n')
    file_object.flush()
if __name__ = '__main__':
    multiprocessing.set_start_method('fork')
    name = []
    file_object = open('x1.txt',mode='a+',encoding='utf-8')
    file_object.write('刘小刘\n')#写到内存中 （此时在进程中还没有写到文件里面）
    p1= multiprocessing.Process(target=task) #fork模式会拷贝主线程的资源，包括内存中的
    p1.start() 
    #主进程在终止的时候会把内存中的资源刷新到硬盘
"""文件里面的内容是
刘小刘
kill
刘小刘
"""
==============================================================================
import multiprocessing
def task():
    print(name)
    file_object.write('kill\n')
    file_object.flush()
if __name__ = '__main__':
    multiprocessing.set_start_method('fork')
    name = []
    file_object = open('x1.txt',mode='a+',encoding='utf-8')
    file_object.write('刘小刘\n')
    file_object.flush() #内存中的数据已经被刷新到硬盘
    p1= multiprocessing.Process(target=task) #此时进程拷贝的时候会有文件对象，但是文件对象没有值
    p1.start() 
"""
刘小刘
kill
"""
===============================================================================
import multiprocessing
import threading
def task():
    print(lock)
    with lock:
        print('执行中')
if __name__ = '__main__':
    multiprocessing.set_start_method('fork')
    name = []
    lock = threading.Rlock() #主进程创建锁
    lock.acquire() #主进程申请并加锁
    p1 = multiprocessing.Process(target = task)#子进程在拷贝资源的时候，锁的状态已经是被主进程的主线程申请走的状态，所以锁在子进程的状态也是被申请走的状态，在子进程中 是被子进程中的主线程申请走了
    p1.start()
"""
print(lock)
<unlock _thread.Rlock object owner =0 count=0 at 0x7fbc800cefc0 >  未申请走的状态
<locked _thread.Rlock object owner =4709898909 count=1 at 0x7fbc800cefc0 >申请走的状态
"""
========================================================================
import multiprocessing
import threading
def task():
    print(lock) #Rlock支持锁多次
    lock.acquire()  
    print(666)
if __name__ = '__main__':
    multiprocessing.set_start_method('fork')
    name = []
    lock = threading.Rlock() #主进程创建锁
    lock.acquire() #主进程申请并加锁
    p1 = multiprocessing.Process(target = task)
    p1.start()
===========================================================================
import multiprocessing
import threading
def func():
    print('来了')
    with lock:  #创建的10个线程都开在这里，因为锁被子进程的主线程申请去了
        print(666)
def task():
    for i in range(10):
        t = threading.Thread(target=func)
        t.start()
    lock.release() #在此处解锁，就不会卡住了
if __name__ = '__main__':
    multiprocessing.set_start_method('fork')
    name = []
    lock = threading.Rlock() #主进程创建锁
    lock.acquire() #主进程申请并加锁
    p1 = multiprocessing.Process(target = task)
    p1.start()

spawn run参数传必备的资源 不支持文件对象、线程锁等传参 unix win系统 进程代码main位置

#不会拷贝资源，需要手动传参
#创建子进程的时候会在内部创建一个python解释器，让python解释器去运行代码
import multiprocessing
def task():
    print(name)
if __name__ ='__main__':
    multiprocessing.set_start_method('spawn') #创建进程的模式改为spawn
    name = []
    p1 = multiprocessing.Process(target=task) #cpu在运行的时候是运行的Process中的run方法
    p1.start()  
# 会报错，因为子进程中没有name
================================================================
import multiprocessing
def task(data):
    print(data)
if __name__ ='__main__':
    multiprocessing.set_start_method('spawn') #创建进程的模式改为fork
    name = []
    p1 = multiprocessing.Process(target=task,args=(name)) #子进程拿到name后会拷贝一份传给data，他们之间各自维护自己的数据
    p1.start()  
=====================================================================
#文件对象和线程锁不支持传参需要的话在子进程中创建

forkservre run参数传必备的资源 不支持文件对象、线程锁等传参 部分unix main代码块

#正常代码是由主进程中的主线程来运行的，如果是forkserver模式，在代码运行之前，python会先创建一个进程，在进程中创建一个线程，把他当作模板放进去，在代码中要是创建子进程，则把模板拷贝一份当作子进程（模板相当于一个什么也没有的进程）

进程的常见方法

p.start() 当前线程准备就绪，等待cpu调度

p.join() 等待当前子进程的任务执行完毕后主进程再向下继续执行

import time
from multipricessing import Process
def task(arg):
    print('执行中')
if __name__ =="__main__":
    multiprocessing.set_start_method('spawn')
    p = Process(target=task,args=('XXX',))
    p.start()
    p.join() # 让主进程等待子进程
    print('继续执行')

p.daemon = 布尔值，守护进程必须放在start之前

p.daemon = True 设置为守护进程，主进程执行 完毕后，子进程也自动关闭
p.daemon = False 设置为非守护进程，主进程等子进程。子进程执行完毕后，主进程才结束

进程设置名称和获取名称

import time
import os
import threading
import multiprocessing
def task(arg):
    threading.enumerate()  #获取当前进程有多少个线程 列表的形式
    os.getpid() #当前进程的id
    os.getppid() #当前进程的父进程的id
    print('当前进程的名字',multiprocess.current_prcess().name) #获取名字
if __name__ = '__main__':
    multiprocessing.set_start_method('spawn')
    p = multiprocessing.Process(target=task,args=('XXX'),name=('进程名'))
    p.name = '90'#设置名字
    p.start()
    print('继续执行')

自定义进程类，直接将线程需要做的事情写到run方法

import multipocessing
class MyProcess(multiprocess.Process)
    def run(self):
        print('执行此线程'，self._args)
if __name__=='__main__':
    multiprocess.set_start_method('spawn')
    p = MyProcess(args=('XXX',))
    p.start()

cpu个数

import multiprocessing
multiprocessing.cpu_count()

p.is_alive() 查看进程是否存活 True 或 False

p.terminate()

告诉操作系统结束进程,具体结束时间由操作系统决定

进程之间数据共享

进程是资源分配的最小单元，默认进程与进程之间是互相隔离的，不共享

基于value array实现进程数据共享

from multiprocessing import Process ,Value,Array
def func(n,m1,m2)
    n.Value = 888
    m1.Value = 'a'.encode('utf-8')
    m2.value= '武'
if __name__ =='__main__':
    num = value('i',666) #创建666类型为int
    v1 = Value('c')#代表一个字符，abc
    v2 = Value('u')#代表一个中文
    p = Process(target=func,args=(num,v1,v2))# 传给子进程，子进程可以通过.value来进行修改
    p.start()
    p.join()
    print(num.value) #888  主进程的数值是子进程中修改后的
    print(v1.value)#a
    print(v2.value)#武
=====================================================================
from multiprocessing import Process ,Value,Array
def f(fata_array):
    data_array[0] = 666
if __name__ =="__main__":
    arr = Array('i',[11,22,33,44])#数组，元素类型必须是int,只能是4个元素
    p = Process(target = f,args=(arr,))
    p=start()
    p.join()
    print(arr[:])

基于Manager（）

from multiprocessing import Process, Manager
def f(d, i):
    d[i] = '1'
    d['2'] = '2'
    d[0.25] = None
    i.append(666)
if __name__ == "__main__":
    with Manager() as manager:
        d = manager.dict() # 创建一个字典
        i = manager.list() # 创建一个列表
        p = Process(target=f, args=(d, i)) # 将创建的字典列表传递给子进程
        p.start()
        p.join()
        print(d)
        print(i)

基于Queues 队列

队列一个放一个取，先进先出
对列是用的内存的空间

import multiprocessing
def task(q):
    for i in range(10):
        q.put(i)  #put往队列里面放值
        q.full()  #查看队列里的数据是否到达最大个数
if __name__ == '__main__':
    queue = multiprocessing.Queue(4)  #主进程创建一个队列,队列里最多放4个数据 省略则不限制个数
    p =multiprocessing.Process(target=task,args=(queue,)) # 将队列传递给子进程
    p.start()
    p.join()
    print('主进程')
    print(queue.get())#get从队列里面取值

基于Pipes 管道

管道是双向的，a可以发数据也可以取数据b也是

import time
import multiprocessing
def task(conn):
    time.sleep(1)
    comm.send([11,22,33,44]) #子进程发送数据
    data = conn.recv() #阻塞
    print('子进程接收'，data)
    time.sleep(2)
if __name__ =='__main__':
    parent_coonn,child_conn = multiprocess.Pipe() #创建一个管道 parent_coonn,child_conn 相当于管道两端
    p = multiprocess.Process(target = task,args=(child_conn,)) # 将管道传递给子进程
    p.start()
    info = parent_conn.recv() #阻塞 # 主进程等待接收
    print('主进程接收'，info)
    parent_comm.send(666) # 主进程发送数据

进程锁

多个进程共享一个资源的时候会用到进程锁

# 数据混乱情况
import time
from multiprocessing import Process ,Value,Array
def func(n,):
    n.value = n.value+1
if __name__ == '__main__':
    num = value('i',0)
    for i in range(20):
        p = process(target=func,args=(num,)) #20个进程都去运行func
        p.start()
    time.sleep(3)
    print(num.value)

# 数据混乱情况
import time
from multiprocessing import Process,manager
def f(d,):
    d[1]+=1
if __name__ =='__main__':
    with Manager() as manager:
        d = manager.dict()
        d[1]=0
        for i in range(20):
            p = Process(target=f,args=(d,))
            p.start()
        time.sleep(3)
        print(d)

# 数据混乱情况
import time
import multiprocessing
def task():
    with open('f1.txt',mode='r',encoding='utf-8') as f:
        current_num = int(f.read())
    print('排队抢票')
    time.sleep(1)
    current_num-=1
    with open('f1.txt',mode='w',encoding='utf-8') as f:
        f.write(str(current_num))
if __name__ =='__main__':
    for i in range(20):
        p = multiprocessing.Process(target=task)
        p.start()

以上实例在进行操作的时候都会出现数据混乱的情况，就需要锁来介入

import time
import multiprocessing
def task(lock):
    print('开始')
    lock.acquire() #加锁
    with open('f1.txt',mode='r',encoding='utf-8') as f:
        current_num = int(f.read())
    print('排队抢票')
    time.sleep(1)
    current_num-=1
    with open('f1.txt',mode='w',encoding='utf-8') as f:
        f.write(str(current_num))
    lock.release()  # 释放锁
if __name__ =='__main__':
    multiprocessing.set_start_method('spawn')
    lock = multiprocessing.Rlock() #创建进程锁
    for i in range(10):
        p = multiprocessing.Process(target=task，args=(lock,))#将锁传递给子进程,共享一把锁
        p.start()
    time.sleep(7)#spawn模式需要特殊处理等子进程运行完毕再继续往下执行
==============================================================================
import time
import multiprocessing
def task(lock):
    print('开始')
    lock.acquire()
    with open('f1.txt',mode='r',encoding='utf-8') as f:
        current_num = int(f.read())
    print('排队抢票')
    time.sleep(1)
    current_num-=1
    with open('f1.txt',mode='w',encoding='utf-8') as f:
        f.write(str(current_num))
    lock.release()
if __name__ =='__main__':
    multiprocessing.set_start_method('spawn')
    lock = multiprocessing.Rlock() #进程锁
    process_list= []
    for i in range(10):
        p = multiprocessing.Process(target=task，args=(lock,))#子进程共享一把锁
        p.start()
        process_list.append(p)
    for item in process_list:
        item.join() # 等每个子进程运行完才往下走

进程池

import time
from concurrent.futures import ProcessPoolExecutor ,ThreadPoolExecutor
def task(num):
    print('执行'，num)
    timr.sleep(2)
if __name__ = '__main__':
    pool = ProcessPoolExecutor(4) #创建进程池最多创建4个进程
    for i in range(10): #创建10个任务
        pool.submit(task,i) #等待过程是在进程池内部执行的

import time
from concurrent.futures import ProcessPoolExecutor ,ThreadPoolExecutor
def task(num):
    print('执行'，num)
    timr.sleep(2)
if __name__ = '__main__':
    pool = ProcessPoolExecutor(4) #创建进程池最多创建4个进程
    for i in range(10): #创建10个任务
        pool.submit(task,i)
    #等待进程池中的任务都执行完了再继续往下执行
    #默认是主线程执行完，子进程同步这进行
    pool.shutdown(True)

import time
import multiprocessing
from concurrent.futures import ProcessPoolExecutor ,ThreadPoolExecutor
def task(num):
    print('执行'，num)
    time.sleep(2)
    return num
def done(res):
    print(multiprocessing.current_process()) #查看当前进程
    time.sleep(1)
    print(res.result())
    time.sleep(1)
if __name__ == '__main__':
    pool = ProcessPoolExecutor(4) #创建进程池最多创建4个进程
    for i in range(10): #创建10个任务
        fur = pool.submit(task,i) #fur为对象
        #done的调用由主进程处理（与线程池不同）
        fur.add_done_callback(done) #done函数中的参数res接收的就是线程执行函数的返回值,此处就是task函数的返回值
    print(multiprocessing.current_process())
    pool.shutdown(True)

注意，在进程池中要使用进程锁，需要基于Manager中的lock和Rlock来实现

import time
import multiprocessing
from concurrent.futures.process import processPoolExecutor
def task(lock):
    with lock:
        # 假设文件中就保存了一个10
        with open('f1.txt',mode = 'r',encoding='utf-8') as f:
            current_num = int(f.read())
        print('抢票')
        time.sleep(1)
        current_num-=1
        with open('f1.txt',mode ='w',encoding='utf-8') as f:
            f.write(str(current_num))
if __name__ =='__main__':
    pool = ProcessPoolExecutor()
    #lock_object = multiprocessing.Rlock()  是不能使用的
    manager = multiprocessing.Manager()
    lock_object = manager.Rlock()
    for i in range(10):
        pool.submit(task,lock_object)

统计用户次数

import os
import time
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Manager
def task(file_name, count_dict):
    ip_set = set()
    total_count = 0
    ip_count = 0
    file_path = os.path.join("files", file_name)
    file_object = open(file_path, mode='r', encoding='utf-8')
    for line in file_object:
        if not line.strip():
            continue
        user_ip = line.split(" - -", maxsplit=1)[0].split(",")[0]
        total_count += 1
        if user_ip in ip_set:
            continue
        ip_count += 1
        ip_set.add(user_ip)
    count_dict[file_name] = {"total": total_count, 'ip': ip_count}
    time.sleep(1)
def run():
    # 根据目录读取文件并初始化字典
    """
        1.读取目录下所有的文件，每个进程处理一个文件。
    """
    pool = ProcessPoolExecutor(4)
    with Manager() as manager:
        """
        count_dict={
            "20210322.log":{"total":10000,'ip':800},
        }
        """
        count_dict = manager.dict()
        for file_name in os.listdir("files"):
            pool.submit(task, file_name, count_dict)
        pool.shutdown(True)
        for k, v in count_dict.items():
            print(k, v)
if __name__ == '__main__':
    run()

import os
import time
from concurrent.futures import ProcessPoolExecutor
def task(file_name):
    ip_set = set()
    total_count = 0
    ip_count = 0
    file_path = os.path.join("files", file_name)
    file_object = open(file_path, mode='r', encoding='utf-8')
    for line in file_object:
        if not line.strip():
            continue
        user_ip = line.split(" - -", maxsplit=1)[0].split(",")[0]
        total_count += 1
        if user_ip in ip_set:
            continue
        ip_count += 1
        ip_set.add(user_ip)
    time.sleep(1)
    return {"total": total_count, 'ip': ip_count}
def outer(info, file_name):
    def done(res, *args, **kwargs):
        info[file_name] = res.result()
    return done
def run():
    # 根据目录读取文件并初始化字典
    """
        1.读取目录下所有的文件，每个进程处理一个文件。
    """
    info = {}
    pool = ProcessPoolExecutor(4)
    for file_name in os.listdir("files"):
        fur = pool.submit(task, file_name)
        fur.add_done_callback(outer(info, file_name))  # 回调函数：主进程
    pool.shutdown(True)
    for k, v in info.items():
        print(k, v)
if __name__ == '__main__':
    run()