python学习之进程线程学习一

Posted 2020-09-25

一、概念
进程: 未完成任务而执行一堆代码的过程，进程是任务，真正执行进程的是cpu
并行: 多个cpu同时运行
并发: 单个cpu分时操作，利用单cpu 的多道技术，看起来像是程序同时运行，其实是分时段运行，
只不过cpu切换速度比较快，并行也属于并发。

多道技术：内存中同时存入多道（多个）程序，cpu从一个进程快速切换到另外一个，使每个进程各
自运行几十或几百毫秒，这样，虽然在某一个瞬间，一个cpu只能执行一个任务，但在1秒内，cpu却
可以运行多个进程，这就给人产生了并行的错觉，即伪并发，以此来区分多处理器操作系统的真正硬
件并行（多个cpu共享同一个物理内存）
同步: 同步就是指一个进程在执行某个请求的时候，若该请求需要一段时间才能返回信息，那么这个
进程将会一直等待下去，直到收到返回信息才继续执行下去
异步: 异步是指进程不需要一直等下去，而是继续执行下面的操作，不管其他进程的状态。当有消息
返回时系统会通知进程进行处理，这样可以提高执行的效率
二、创建进程模块与方法
例子:
from multiprocessing import Process
import time
import random

def piao(name):
    print(‘%s is piaoing‘ %name)
    time.sleep((random.randint(1,3)))
    print(‘%s is piao end‘ %name)

if __name__ == ‘__main__‘:
    p1 = Process(target=piao, args=(‘p1‘,))
    p2 = Process(target=piao, args=(‘p2‘,))
    p3 = Process(target=piao, args=(‘p3‘,))
    p4 = Process(target=piao, args=(‘p4‘,))

    p_l = [p1, p2, p3, p4]
    for p in p_l:
        p.start()
    for p in p_l:
        p.join()
    print(‘主进程‘)
multiprocessing: multiprocessing模块用来开启子进程，并在子进程中执行我们定制的任务（比如
函数），该模块与多线程模块threading的编程接口类似
Process: Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，
表示一个子进程中的任务（尚未启动）
group: 未使用，默认是None
target: 子进程执行的任务，一般为函数
name: 子进程名字
args: target任务对象的参数，由元组组成
kwargs: target任务对象的字典参数,

进程方法介绍:
p.start(): 启动进程，并调用该子进程中的p.run()
p.run(): 进程启动时运行的方法，正是它去调用target指定的函数，自定义类的类中一定要实现该方法
p.terminate(): 强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进
程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
p.is_alive(): 如果p仍然运行，返回True
p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。
timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run
开启的进程
进程属性介绍:
p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，
p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置
p.name: 进程的名称
p.pid：进程的pid
创建进程的两种方式:
例子:
from multiprocessing import Process
import time
import random

def piao(name):
    print(‘%s is piaoing‘ %name)
    time.sleep(3)
    print(‘%s is piao end‘ %name)

if __name__ == ‘__main__‘:
    p1 = Process(target=piao, args=(‘hyh‘,), name=‘<p1>‘)
    p1.start()
    print(‘p1 name is %s ‘ %p1.name)
    print(‘父进程‘)

from multiprocessing import Process
import time
import random

class piao(Process):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        print(‘%s is piaoing‘ %self.name)
        time.sleep(3)
        print(‘%s is piao end‘ %self.name)

if __name__ == ‘__main__‘:
    p1 = piao(‘hyh‘)
    p1.start()  #p1.run
    print(‘父进程‘)
并发实现例子:
server:
#!/usr/bin/python
# --*-- coding: utf-8 --*--
from multiprocessing import Process
from socket import *

server = socket(AF_INET, SOCK_STREAM)
server.setsockopt(SOL_SOCKET, SO_REUSEADDR,1)
server.bind((‘127.0.0.1‘, 8080))
server.listen(5)

def talk(conn, addr):
    while True:
        try:
            msg = conn.recv(1024)
            if not msg: break
            conn.send(msg.upper())
        except Exception:
            break

if __name__ == ‘__main__‘:
    while True:
        conn, addr = server.accept()
        p = Process(target=talk, args=(conn, addr))
        p.start()
        
多个client
#!/usr/bin/python
# --*-- coding: utf-8 --*--
from socket import *

client = socket(AF_INET, SOCK_STREAM)
client.connect((‘127.0.0.1‘, 8080))

while True:
    msg = input(">>: ").strip()
    if not msg: continue
    client.send(msg.encode(‘utf-8‘))
    msg = client.recv(1024)
    print(msg.decode(‘utf-8‘))
    
多少个client链接到服务端，就需要服务端开启多少个进程，大并发下，服务器会崩溃，怎么处理，
需要借助进程池pool（后面有pool类使用例子）

       
守护进程daemon,主进程执行完毕，子进程就跟着结束
例子:
from multiprocessing import Process
import time
import random

class Piao(Process):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        print(‘%s is piaoing‘ %self.name)
        time.sleep(random.randint(1,3))
        print(‘%s is piao end‘)
p = Piao(‘test‘)
p.daemon = True
p.start()
#p.join(0.0001)
print(‘开始‘)

三、进程间通信
进程之间是彼此隔离的，需要借助队列和管道实现通信
队列类Queue
Queue([maxsize]):创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间
的数据传递
maxsize表示队列中存放最多进程数

Queue对象q的方法:
q.put(): 插入数据到队列，put方法还有两个可选参数：blocked和timeout。如果blocked为True
（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。
如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常

q.get(): 从队列中去除数据
q.empty():调用此方法时q为空则返回True，该结果不可靠，比如在返回True的过程中，如果队列中又
加入了项目
q.full(): q已满，返回True
q.qsize(): 返回队列中目前项目的正确数量

Lock锁类，通过锁定对象，使一个进程操作该对象时，其他进程等待
抢票例子:
from multiprocessing import Process,Lock
import json
import time
import random

def work(dbfile, name, lock):
    with lock:
        with open(dbfile, encoding=‘utf-8‘) as f:
            dic = json.loads(f.read())
        if dic[‘count‘] > 0:
            dic[‘count‘] -= 1
            time.sleep(random.randint(1,3))
            with open(dbfile, ‘w‘, encoding=‘utf-8‘) as f:
                f.write(json.dumps(dic))
            print(‘\033[34m%s 抢票成功\033[0m‘ %name)
        else:
            print(‘\033[37m%s 抢票失败\033[0m‘ %name)

if __name__ == ‘__main__‘:
    lock = Lock()
    p_l = []
    for i in range(100):
        p = Process(target=work, args=(‘a.txt‘, ‘用户%s‘ %i, lock))
        p_l.append(p)
        p.start()

    for p in p_l:
        p.join()
    print(‘主进程‘)
    
    
Queue队列存放取出对象例子:
from multiprocessing import Process, Queue
import time
q = Queue(3)
q.put(3)
q.put(4)
q.put(5)
print(q.full())

print(q.get())
print(q.get())
print(q.get())
print(q.empty())

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

def consumer(q):
    while True:
        time.sleep(random.randint(1,3))
        res = q.get()
        if res is None: break
        print(‘\033[34m消费者拿到了: %s\033[0m‘ %res)

def producer(seq, q):
    for item in seq:
        time.sleep(random.randint(1,3))
        print(‘\033[45m生产者生产了: %s\033[0m‘ %item)
        q.put(item)

if __name__ == ‘__main__‘:
    q = Queue()
    c = Process(target=consumer, args=(q,))
    c.start()

    producer((‘P-%s‘ %i for i in range(10)), q)
    q.put(None)
    c.join()
    print(‘主线程‘)
    
JoinableQueue 的实例p除了与Queue对象相同的方法之外还具有
q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常
q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止  
from multiprocessing import Process, JoinableQueue
import time, random
def consumer(q):
    while True:
        res = q.get()
        print(‘消费者拿到了 %s‘ %res)
        q.task_done()

def producer(seq, q):
    for item in seq:
        q.put(item)
        print(‘生产者做好了 %s‘ %item)
    q.join()

if __name__ == ‘__main__‘:
    q = JoinableQueue()
    seq = (‘P-%s‘ %i for i in range(10))
    p = Process(target=consumer, args=(q,))
    p.daemon=True
    p.start()

    producer(seq, q)
    print(‘主线程‘)
    
多消费者模式
例子:
from multiprocessing import Process, JoinableQueue
import time,random

def consumer(name, q):
    while True:
        time.sleep(random.randint(1,3))
        res = q.get()
        print(‘\033[34m%s拿到了 %s\033[0m‘ %(name, res))
        q.task_done()

def producer(seq, q):
    for item in seq:
        time.sleep(random.randrange(1,2))
        q.put(item)
        print(‘\033[42m生产者做好了 %s\033[0m‘ %item)
    q.join()

if __name__ == ‘__main__‘:
    q = JoinableQueue()
    seq = (‘P-%s‘ %i for i in range(10))
    p1 = Process(target=consumer, args=(‘C-1‘, q,))
    p2 = Process(target=consumer, args=(‘C-2‘, q,))
    p3 = Process(target=consumer, args=(‘C-3‘, q,))
    p1.daemon = True
    p2.daemon = True
    p3.daemon = True
    p1.start()
    p2.start()
    p3.start()

    producer(seq, q)
    print(‘主线程‘)

Semaphore类允许一定数量的线程修改数据，，如果指定信号量为3，那么来一个人获得一把锁，计数加1，
当计数等于3时，后面的人均需要等待。一旦释放，就有人可以获得一把锁
信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念
例子:
from multiprocessing import Process, Semaphore
import time, random

def go_wc(sem, user):
    sem.acquire()
    print(‘%s 占用一个位置‘ %user)
    time.sleep(random.randint(0,3))
    sem.release()

if __name__ == ‘__main__‘:
    sem = Semaphore(5)
    p_l = []
    for i in range(13):
        p = Process(target=go_wc, args=(sem,‘user%s‘ %i,))
        p.start()
        p_l.append(p)

    for i in p_l:
        i.join()
    print(‘=========>‘)

    
四、进程池pool
进程开启越多占用系统资源越多，不能无限开启进程，需要用有控制进程数量的手段，pool类可以控制
创建进程的数量.

Pool可以提供指定数量的进程，供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就
会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，
直到池中有进程结束，就重用进程池中的进程

Pool([numprocess  [,initializer [, initargs]]]): 创建进程池
numprocess: 要创建的进程数，如果省略，将默认使用cpu_count()的值
initializer：是每个工作进程启动时要执行的可调用对象，默认为None
initargs：是要传给initializer的参数组

p.apply(func [, args [, kwargs]]): 在一个池工作进程中执行func(*args,**kwargs),然后返回结果。
需要强调的是：此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地
执行func函数，必须从不同线程调用p.apply()函数或者使用p.apply_async()

p.apply_async(func [, args [, kwargs]]): 在一个池工作进程中执行func(*args,**kwargs),然后返回
结果。此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变
为可用时，将理解传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果

p.close(): 关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成
p.terminate()：立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数
p.join():等待所有工作进程退出。此方法只能在close（）或teminate()之后调用

使用进程池pool的例子:
异步例子:    
#!/usr/bin/python
# --*-- coding: utf-8 --*--
from multiprocessing import Process, Pool
from socket import *
import os

server = socket(AF_INET, SOCK_STREAM)
server.setsockopt(SOL_SOCKET, SO_REUSEADDR,1)
server.bind((‘127.0.0.1‘, 8080))
server.listen(5)

def talk(conn, addr):
    print(os.getpid())
    while True:
        try:
            msg = conn.recv(1024)
            if not msg: break
            conn.send(msg.upper())
        except Exception:
            break

if __name__ == ‘__main__‘:
    pool = Pool()   #默认保留4个进程,最多同时4个client连过来
    res_l = []
    while True:
        conn, addr = server.accept()
        res = pool.apply_async(talk, args=(conn, addr))     #异步方法apply_async
        res_l.append(res)
        #print(res_l)
同步例子:
#!/usr/bin/python
# --*-- coding: utf-8 --*--
from multiprocessing import Process, Pool
from socket import *
import os

server = socket(AF_INET, SOCK_STREAM)
server.setsockopt(SOL_SOCKET, SO_REUSEADDR,1)
server.bind((‘127.0.0.1‘, 8080))
server.listen(5)

def talk(conn, addr):
    print(os.getpid())
    while True:
        try:
            msg = conn.recv(1024)
            if not msg: break
            conn.send(msg.upper())
        except Exception:
            break

if __name__ == ‘__main__‘:
    pool = Pool()   #默认保留4个进程
    #res_l = []
    while True:
        conn, addr = server.accept()
        pool.apply(talk, args=(conn,addr))    一次只能一个进程通信，其他进程等待io

回掉函数
例子:
from multiprocessing import Pool
import time, random

def get_page(url):
    time.sleep(random.randint(1,3))
    print(‘下载页面: %s‘ %url)
    return {‘url‘: url}

def parse_page(page_content):
    time.sleep(1)
    print(‘解析页面: %s‘ %page_content)

if __name__ == ‘__main__‘:
    urls = [
        ‘http://maoyan.com/board/7‘,
        ‘http://maoyan.com/board/1‘,
        ‘http://maoyan.com/board/2‘
    ]
    p = Pool()
    res_l = []
    for url in urls:
        res = p.apply_async(get_page, args=(url, ), callback=parse_page)    #调用parse_page
        res_l.append(res)

    for i in res_l:
        i.get()

本文出自 “linux技术” 博客，请务必保留此出处http://haoyonghui.blog.51cto.com/4278020/1942464