Py004-01-08生产者消费者模型

生产者消费者模型介绍

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

生产者指的是生产数据的任务，消费者指的是处理数据的任务，在并发编程中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

什么是生产者和消费者模式

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

这个阻塞队列就是用来给生产者和消费者解耦的

生产者消费者模型实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from multiprocessing import Process,Queue
import time,random,os
def consumer(q,name):
    while True:
        res=q.get()
        time.sleep(random.randint(1,3))
        print('\033[43m%s 吃 %s\033[0m' %(name,res))

def producer(q,name,food):
    for i in range(3):
        time.sleep(random.randint(1,3))
        res='%s%s' %(food,i)
        q.put(res)
        print('\033[45m%s 生产了 %s\033[0m' %(name,res))

if __name__ == '__main__':
    q=Queue()
    #生产者们:即厨师们
    p1=Process(target=producer,args=(q,'egon','包子'))

    #消费者们:即吃货们
    c1=Process(target=consumer,args=(q,'alex'))

    #开始
    p1.start()
    c1.start()
    print('主')

执行结果

1
2
3
4
5
6
7
主
egon 生产了 包子0
egon 生产了 包子1
alex 吃 包子0
alex 吃 包子1
egon 生产了 包子2
alex 吃 包子2

此时的问题是主进程永远不会结束，原因是：生产者p在生产完后就结束了，但是消费者c在取空了q之后，则一直处于死循环中且卡在q.get()这一步。

解决方式无非是让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就可以break出死循环

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from multiprocessing import Process,Queue
import time,random,os
def consumer(q,name):
    while True:
        res=q.get()
        if res is None:break
        time.sleep(random.randint(1,3))
        print('\033[43m%s 吃 %s\033[0m' %(name,res))

def producer(q,name,food):
    for i in range(3):
        time.sleep(random.randint(1,3))
        res='%s%s' %(food,i)
        q.put(res)
        print('\033[45m%s 生产了 %s\033[0m' %(name,res))

if __name__ == '__main__':
    q=Queue()
    #生产者们:即厨师们
    p1=Process(target=producer,args=(q,'egon','包子'))

    #消费者们:即吃货们
    c1=Process(target=consumer,args=(q,'alex'))

    #开始
    p1.start()
    c1.start()

    p1.join()
    q.put(None)
    print('主')

但上述解决方式，在有多个生产者和多个消费者时，我们则需要用一个很low的方式去解决,有几个消费者就需要发送几次结束信号：相当low,例如

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
from multiprocessing import Process,Queue
import time,random,os
def consumer(q,name):
    while True:
        res=q.get()
        if res is None:break
        time.sleep(random.randint(1,3))
        print('\033[43m%s 吃 %s\033[0m' %(name,res))

def producer(q,name,food):
    for i in range(3):
        time.sleep(random.randint(1,3))
        res='%s%s' %(food,i)
        q.put(res)
        print('\033[45m%s 生产了 %s\033[0m' %(name,res))

if __name__ == '__main__':
    q=Queue()
    #生产者们:即厨师们
    p1=Process(target=producer,args=(q,'egon1','包子'))
    p2=Process(target=producer,args=(q,'egon2','骨头'))
    p3=Process(target=producer,args=(q,'egon3','泔水'))

    #消费者们:即吃货们
    c1=Process(target=consumer,args=(q,'alex1'))
    c2=Process(target=consumer,args=(q,'alex2'))

    #开始
    p1.start()
    p2.start()
    p3.start()
    c1.start()
    c2.start()

    p1.join()
    p2.join()
    p3.join()
    q.put(None)
    q.put(None)
    q.put(None)
    print('主')

其实我们的思路无非是发送结束信号而已，有另外一种队列提供了这种机制

  • JoinableQueue([maxsize])
1
这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。

参数介绍

  • maxsize是队列中允许最大项数，省略则无大小限制。

方法介绍

JoinableQueue的实例p除了与Queue对象相同的方法之外还具：

  • q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常

  • q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

基于JoinableQueue实现生产者消费者模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
from multiprocessing import Process,JoinableQueue
import time,random,os
def consumer(q,name):
    while True:
        res=q.get()
        time.sleep(random.randint(1,3))
        print('\033[43m%s 吃 %s\033[0m' %(name,res))
        q.task_done() #发送信号给q.join()，说明已经从队列中取走一个数据并处理完毕了

def producer(q,name,food):
    for i in range(3):
        time.sleep(random.randint(1,3))
        res='%s%s' %(food,i)
        q.put(res)
        print('\033[45m%s 生产了 %s\033[0m' %(name,res))
    q.join() #等到消费者把自己放入队列中的所有的数据都取走之后，生产者才结束

if __name__ == '__main__':
    q=JoinableQueue() #使用JoinableQueue()

    #生产者们:即厨师们
    p1=Process(target=producer,args=(q,'egon1','包子'))
    p2=Process(target=producer,args=(q,'egon2','骨头'))
    p3=Process(target=producer,args=(q,'egon3','泔水'))

    #消费者们:即吃货们
    c1=Process(target=consumer,args=(q,'alex1'))
    c2=Process(target=consumer,args=(q,'alex2'))
    c1.daemon=True
    c2.daemon=True

    #开始
    p1.start()
    p2.start()
    p3.start()
    c1.start()
    c2.start()

    p1.join()
    p2.join()
    p3.join()
    #1、主进程等生产者p1、p2、p3结束
    #2、而p1、p2、p3是在消费者把所有数据都取干净之后才会结束
    #3、所以一旦p1、p2、p3结束了，证明消费者也没必要存在了，应该随着主进程一块死掉，因而需要将生产者们设置成守护进程
    print('主')

生产者消费者模型总结

  1. 程序中有两类角色
  • 一类负责生产数据（生产者）
  • 一类负责处理数据（消费者）
  1. 引入生产者消费者模型为了解决的问题是
  • 平衡生产者与消费者之间的速度差
  • 程序解开耦合
  1. 如何实现生产者消费者模型
  • 生产者<--->队列<--->消费者

Py004-01-07队列

队列

进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的

  • 队列操作的是内存空间

队列就是管道加锁实现

1
Queue([maxsize]):创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。

参数介绍：

maxsize是队列中允许最大项数，省略则无大小限制。

但需要明确：

  • 队列内存放的是消息而非大数据
  • 队列占用的是内存空间，因而maxsize即便是无大小限制也受限于内存大小

主要方法介绍：

1
2
q.put方法用以插入数据到队列中。
q.get方法可以从队列读取并且删除一个元素。

队列的使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from multiprocessing import Process,Queue

q=Queue(3) # 指定队列最大数量为3 就是能放三个数据进去

#put ,get ,put_nowait,get_nowait,full,empty
q.put(1)
q.put(2)
q.put(3)
print(q.full()) #满了  (队列是否满了)
# q.put(4) #再放就阻塞住了

print(q.get())
print(q.get())
print(q.get())
print(q.empty()) #空了
# print(q.get()) #再取就阻塞住了 队列已经没数据了

Py004-01-06互斥锁

互斥锁

多个进程直接虽然数据不共享，但是它们共享同一套文件系统(或者同一个打印终端)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from multiprocessing import Process
import time

def task(name):
    print('%s 1'%name)
    time.sleep(1)
    print('%s 2'%name)
    time.sleep(1)
    print('%s 3' % name)

if __name__ == '__main__':
    # args传递的参数必须是元组的形式
    for i in range(3):
        p = Process(target=task, args=('子进程%s'%i,))
        p.start()

    print('主进程。。。。')


'''
主进程。。。。
子进程0 1
子进程1 1
子进程2 1
子进程0 2
子进程1 2
子进程2 2
子进程0 3
子进程1 3
子进程2 3
此时数据发生错乱，三个进程都占据打印终端的输出，这样数据容易出现问题
'''

互斥锁(来把锁)

  • 从multiprocessing引入Lock
  • 初始化锁 mutex = Lock()
  • 给子进程的触发函数传入锁对象mutex
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from multiprocessing import Process,Lock
import time

def task(name,mutex):
    mutex.acquire()
    print('%s 1'%name)
    time.sleep(1)
    print('%s 2'%name)
    time.sleep(1)
    print('%s 3' % name)
    mutex.release()

if __name__ == '__main__':
    # args传递的参数必须是元组的形式
    for i in range(3):
        mutex=Lock()
        p = Process(target=task, args=('子进程%s'%i,mutex))
        p.start()

    print('主进程。。。。')

'''
主进程。。。。
子进程0 1
子进程1 1
子进程2 1
子进程0 2
子进程1 2
子进程2 2
子进程0 3
子进程1 3
子进程2 3
'''

模拟抢票

多个进程共享同一文件，我们可以把文件当数据库，用多个进程模拟多个人执行抢票任务

新建文件db.txt的内容为：{“count”:1}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
from multiprocessing import  Process
import json
import time

def search(name):
    time.sleep(1) # 模拟网络延迟
    dic = json.load(open('db.txt',encoding='utf-8'))
    print('<%s>查看剩余票数[%s]'%(name,dic['count']))

def get(name):
    time.sleep(1)
    dic = json.load(open('db.txt', encoding='utf-8'))
    if dic['count']>0:
        dic['count']-=1
        time.sleep(3)
        json.dump(dic,open('db.txt','w',encoding='utf-8'))
        print('<%s 购票成功>'%name)

def task(name):
    search(name)
    get(name)

if __name__ == '__main__':
    for i in range(10):
        p = Process(target=task,args=('路人%s'%i,))
        p.start()

'''
<路人0>查看剩余票数[1]
<路人2>查看剩余票数[1]
<路人1>查看剩余票数[1]
<路人3>查看剩余票数[1]
<路人4>查看剩余票数[1]
<路人5>查看剩余票数[1]
<路人6>查看剩余票数[1]
<路人7>查看剩余票数[1]
<路人8>查看剩余票数[1]
<路人9>查看剩余票数[1]
<路人0 购票成功>
<路人2 购票成功>
<路人1 购票成功>
<路人3 购票成功>
<路人4 购票成功>
<路人5 购票成功>
<路人6 购票成功>
<路人7 购票成功>
<路人8 购票成功>
<路人9 购票成功>

并发的执行查看同一个文件，于是都看到了剩余1张票，最后都购票成功于是 打起来了。。。。
'''

加锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
from multiprocessing import  Process,Lock
import json
import time

def search(name):
    time.sleep(1) # 模拟网络延迟
    dic = json.load(open('db.txt',encoding='utf-8'))
    print('<%s>查看剩余票数[%s]'%(name,dic['count']))

def get(name):
    time.sleep(1)
    dic = json.load(open('db.txt', encoding='utf-8'))
    if dic['count']>0:
        dic['count']-=1
        time.sleep(3)
        json.dump(dic,open('db.txt','w',encoding='utf-8'))

        print('<%s 购票成功>'%name)
def task(name,mutex):
    mutex.acquire()
    search(name)
    get(name)
    mutex.release()

if __name__ == '__main__':
    mutex = Lock()
    for i in range(10):
        p = Process(target=task,args=('路人%s'%i,mutex))
        p.start()
'''
<路人0>查看剩余票数[1]
<路人0 购票成功>
<路人2>查看剩余票数[0]
<路人1>查看剩余票数[0]
<路人3>查看剩余票数[0]
<路人4>查看剩余票数[0]
<路人5>查看剩余票数[0]
<路人6>查看剩余票数[0]
<路人7>查看剩余票数[0]
<路人8>查看剩余票数[0]
<路人9>查看剩余票数[0]
'''

上述有一个问题必须等第一个人购票了其他人才能查票

做如下修改

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def task(name,mutex):

    search(name)

    mutex.acquire()
    get(name)
    mutex.release()

'''
<路人0>查看剩余票数[1]
<路人1>查看剩余票数[1]
<路人2>查看剩余票数[1]
<路人3>查看剩余票数[1]
<路人4>查看剩余票数[1]
<路人5>查看剩余票数[1]
<路人6>查看剩余票数[1]
<路人7>查看剩余票数[1]
<路人8>查看剩余票数[1]
<路人9>查看剩余票数[1]
<路人0 购票成功>
'''

互斥锁和join的区别

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
from multiprocessing import  Process,Lock
import json
import time

def search(name):
    time.sleep(1) # 模拟网络延迟
    dic = json.load(open('db.txt',encoding='utf-8'))
    print('<%s>查看剩余票数[%s]'%(name,dic['count']))

def get(name):
    time.sleep(1)
    dic = json.load(open('db.txt', encoding='utf-8'))
    if dic['count']>0:
        dic['count']-=1
        time.sleep(3)
        json.dump(dic,open('db.txt','w',encoding='utf-8'))
        print('<%s 购票成功>'%name)
    else:
        print('<%s 购票失败>' % name)
def task(name):
    search(name)
    get(name)

if __name__ == '__main__':
    for i in range(10):
        p = Process(target=task,args=('路人%s'%i,))
        p.start()
        p.join()

'''
去掉锁后改用join虽然保证了串行，影响了整个的查询效率变慢了
<路人0>查看剩余票数[1]
<路人0 购票成功>
<路人1>查看剩余票数[0]
<路人1 购票失败>
<路人2>查看剩余票数[0]
<路人2 购票失败>
<路人3>查看剩余票数[0]
<路人3 购票失败>
<路人4>查看剩余票数[0]
<路人4 购票失败>
<路人5>查看剩余票数[0]
<路人5 购票失败>
<路人6>查看剩余票数[0]
<路人6 购票失败>
<路人7>查看剩余票数[0]
<路人7 购票失败>
<路人8>查看剩余票数[0]
<路人8 购票失败>
<路人9>查看剩余票数[0]
<路人9 购票失败>
'''

Py004-01-05守护进程

守护进程

主进程创建子进程，然后将该进程设置成守护自己的进程，守护进程就好比崇祯皇帝身边的老太监，崇祯皇帝已死老太监就跟着殉葬了。

关于守护进程需要强调两点：

  1. 守护进程会在主进程代码执行结束后就终止

  2. 守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children

如果我们有两个任务需要并发执行，那么开一个主进程和一个子进程分别去执行就ok了，如果子进程的任务在主进程任务结束后就没有存在的必要了，那么该子进程应该在开启前就被设置成守护进程。主进程代码运行结束，守护进程随即终止

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from multiprocessing import Process
import time
import random

def task(name):
    print('%s is piaoing' %name)
    time.sleep(random.randrange(1,3))
    print('%s is piao end' %name)


if __name__ == '__main__':
    p=Process(target=task,args=('egon',))
    p.daemon=True #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行
    p.start()
    print('主') #只要终端打印出这一行内容，那么守护进程p也就跟着结束掉了

守护进程要在start前设置daemon=True 否则报错

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
#守护进程一定是等到主进程结束然后终止运行
from multiprocessing import Process

import time
def foo():
    print(123)
    time.sleep(1)
    print("end123")

def bar():
    print(456)
    time.sleep(3)
    print("end456")

if __name__ == '__main__':
    p1=Process(target=foo)
    p2=Process(target=bar)

    p1.daemon=True
    p1.start()
    p2.start()
    print("main-------")

'''
永远看不到  end123
因为执行到  main------之后 主进程就结束了
'''

Py004-01-04Process其他属性和方法

Process其他属性和方法

需求一(父进程等待子进程结束在运行)

  • join
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from multiprocessing import Process
import time,os

def task(name):
    print('%s is running'%name)
    time.sleep(3)
    print('%s is done'%name)

if __name__ == '__main__':
    # args传递的参数必须是元组的形式
    p = Process(target=task,args=('子进程',))

    p.start()

    p.join()
    print('主进程。。。。1',os.getpid(),os.getppid())
    print(p.pid) # 主进程进行完后，子进程pid并不会回收，但是如果主进程结束的时候所有的子进程就会结束

不要以代码先后看执行结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from multiprocessing import Process
import time,os

def task(name):
    print('%s is running'%name)

if __name__ == '__main__':
    # args传递的参数必须是元组的形式
    p1 = Process(target=task,args=('子进程1',))
    p2 = Process(target=task,args=('子进程2',))
    p3 = Process(target=task,args=('子进程3',))

    p1.start()
    p2.start()
    p3.start()

    print('主进程开始了')

'''
代码执行结果：
p1.start()只是给操作系统发信号不一定会先执行
p2.start()只是给操作系统发信号不一定会先执行
p3.start()只是给操作系统发信号不一定会先执行
多执行几次就会看大顺序是不确定的
'''

偏要顺序执行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from multiprocessing import Process
import time,os

def task(name):
    print('%s is running'%name)

if __name__ == '__main__':
    # args传递的参数必须是元组的形式
    p1 = Process(target=task,args=('子进程1',))
    p2 = Process(target=task,args=('子进程2',))
    p3 = Process(target=task,args=('子进程3',))

    p1.start()
    p2.start()
    p3.start()

    p1.join()
    p2.join()
    p3.join()
    
    print('主进程开始了')

你已经踩坑了
你已经踩坑了
你已经踩坑了

　假设p1运行1个小时，p2运行半个小时,p3运行10分钟，请问主进程等待多长时间？

1
2
3
p1.join() # 让主进程等待，并不会让兄弟进程(其他子进程)等待

所以最后主进程会等待1小时

真正意义的串行执行

这样修改就行了

1
2
3
4
5
6
p1.start()
p1.join()
p2.start()
p2.join()
p3.start()
p3.join()

巧用for循环

1
2
3
4
arr_p = [p1,p2,p3]
for p in arr_p:
    p.start()
    p.join()

查看子进程是否活着

  • is_alive()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from multiprocessing import Process
import time,os

def task(name):
    print('%s is running'%name)
    time.sleep(3)
    print('%s is running'%name)

if __name__ == '__main__':
    # args传递的参数必须是元组的形式
    p = Process(target=task,args=('子进程',))

    p.start()
    print(p.is_alive())
    print('主进程')
    print(p.pid)
    print(p.is_alive())

杀死进程

  • terminate()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from multiprocessing import Process
import time,os

def task(name):
    print('%s is running'%name)

if __name__ == '__main__':
    # args传递的参数必须是元组的形式
    p = Process(target=task,args=('子进程',))

    p.start()
    
    p.terminate() # 只是给操作系统发信号让子进程终止，但不会立即执行 ，所以之后查看子进程is_alive还是True
    print(p.is_alive())
    
    print('主进程')

怎样真的死掉

1
2
3
4
5
p.start()
    
p.terminate() # 只是给操作系统发信号让子进程终止，但不会立即执行
time.sleep(3) #
print(p.is_alive())

给子进程起名字

  • name
1
2
3
4
# 不指定name参数就有一个默认的
p = Process(target=task,name='sub-Process')
p.start()
print(p.name)

父子进程是隔离的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from multiprocessing import Process
import time,os

def task():
    global n
    n = 0
    print('子进程内%s'%n)

if __name__ == '__main__':
    # args传递的参数必须是元组的形式
    p = Process(target=task)

    p.start()
    p.join()
    print('主进程',n)

Py004-01-03僵尸进程和孤儿进程

了解的知识点

例子

一个父进程，在运行中开启了10个子进程，这10个子进程运行跟父进程就没关系了，相互独立

但是：父进程毕竟是子进程的爹，有这么一个需求——查看子进程的状态

  • 假设子进程运行完后就把子进程的所有状态信息清空掉？可能吗？

此时作为父进程就看不到子进程的状态信息

僵尸进程

就算子进程死了，但是它相关的状态信息不能全部清除掉，要留着——尸体(方便父进程查看)

谁来给子进程收尸呢？

父进程结束的时候，将子进程的信息清除掉

僵尸进程有害吗？

如果父进程一直不死，子进程就一直无法释放。

有害，每个进程占据一个pid如果没有及时清除，积累过多的进程就会导致新进程无法开启。

孤儿进程

子进程没死，父进程先死了

谁收留孤儿

如果是linux系统，会有一个init进程(所有进程的爹)，如果子进程的父进程死了，就由init来接管。

由init进程发起系统调用回收子进程

孤儿进程有害吗？

无害，由init负责回收

Py004-01-02查看进程编号pid

进程编号pid和ppid

要引入os模块可以查看进程编号

  • pid 进程编号
  • ppid 父进程编号
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from multiprocessing import Process
import time,os

def task(name):
    print('%s is running 子进程编号为%s 父进程编号为%s' % (name,os.getpid(),os.getppid()))
    time.sleep(3)
    print('%s is running 子进程编号为%s 父进程编号为%s' % (name, os.getpid(), os.getppid()))

if __name__ == '__main__':
    # args传递的参数必须是元组的形式
    p = Process(target=task,args=('子进程',))

    p.start()

    print('主进程。。。。1   %s' % os.getpid())
    print('主进程。。。。2   %s' % os.getpid())
    print('主进程。。。。3   %s' % os.getpid())

查看运行的pycharm进程

if windows

1
2
打开cmd
tasklist | findstr pycharm

if mac

1
ps aus|grep pycharm

Py004-01-01进程的开启方式

进程

要引入进程模块

1
from multiprocessing import Process

开启进程方式一

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

from multiprocessing import Process
import time
# 方式一

def task(name):
    print('%s is running'%name)
    time.sleep(3)
    print('%s is done'%name)

if __name__ == '__main__':
    # args传递的参数必须是元组的形式
    p = Process(target=task,args=('子进程',))
    # 关键字参数方式
    p2 = Process(target=task,kwargs={'name':'子进程2'})

    p.start()
    p2.start()

    print('主进程。。。。1')
    print('主进程。。。。2')
    print('主进程。。。。3')

'''
执行结果
主进程。。。。1
主进程。。。。2
主进程。。。。3
子进程 is running
子进程2 is running
'''

开启进程方式二(自己定制)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from multiprocessing import Process
import time

class MyProcess(Process):
    def __init__(self,name):
        super().__init__()
        self.name = name
    # 自定制的进程  就不能叫task了  必须叫 run
    def run(self):
        print('%s is running' % self.name)
        time.sleep(3)
        print('%s is done' % self.name)

if __name__ == '__main__':
    p = MyProcess('子进程1')
    p.start()

    print('主进程。。。1')
    print('主进程。。。2')
    print('主进程。。。3')

Py003-04-02struct模块

struct模块

这里我们只用它的打包和解包

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

import struct
str = '0123456789'
str_size = len(str) # 10

# 打包
str_key = struct.pack('i', str_size)
print('打包',str_key)

# 解包
data_size = struct.unpack('i', str_key)
print('解包',data_size)

‘’‘
打包 b'\n\x00\x00\x00'
解包 (10,)
’‘’

他人博客链接

他人博客链接

Py003-04-01optparse模块

optparse模块

获取执行py文件时获取的参数

如test.py文件如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import optparse

parser = optparse.OptionParser()
parser.add_option("-s","--server", dest="server", help="ftp server ip_addr")
parser.add_option("-P","--port",type="int", dest="port", help="ftp server port")

a,b = parser.parse_args()

print(a,b)


# 如果执行时参数如下
# python3 test.py a b c -s localhost -P 9999
# 结果如下
# {'server': 'localhost', 'port': 9999} ['a','b','c']