python多进程

发布时间：2019-08-09 10:34:13编辑：auto阅读（1682）

第一种开启进程方式

#!/usr/bin/python
# -*- coding:utf-8 -*-
from multiprocessing import Process
import time, random, os

# print(os.cpu_count())
#
#
# # 定义一个任务
# def piao(name):
#     print('%s is piaoing' % name)
#     time.sleep(3)  # cpu阻塞，切换到主进程
#     print('%s is piao end' % name)
#
#
# if __name__ == '__main__':
#     # target指定执行目标，args指定位置参数，指定为元组，kwargs指定字典
#     p1 = Process(target=piao, args=('lh',)，name='<p1>')  # 产生对象
#     # 创建子进程，赋值主进程地址空间，开启一个子进程,创建进程需要时间，
#     # 发送系统调用，只是一个系统调用，指挥操作系统启动子进
#     # 程，主进程并不等待子进程
#     p1.start()  # 开启子进程
#     # time.sleep(1)  #
#     print('主进程')
#     # 谁先执行完谁先打印

第二种，自定义进程类

class Piao(Process):
    def __init__(self, name):
        super().__init__()  # 重用父类方法,
        self.name = name

    def run(self):  # 方法名必须为run
        print('%s is piaoing...' % self.name)
        time.sleep(3)
        print('%s is piao end' % self.name)


if __name__ == '__main__':
    p1 = Piao('lh')
    p1.start()  # 相当于p1.run()
    print('主进程')

锁

#!/usr/bin/python
# -*- coding:utf-8 -*-
# import threading, time
#
# v = 10
#
# lock = threading.Lock() #只能有一个使用锁
# lock = threading.RLock()  # 递归锁
#
#
# def task(arg):
#     time.sleep(2)
#     lock.accquire()  # 申请使用锁，其他线程等待
#     global v
#     v -= 1
#     print(v)
#     lock.release()  # 交还释放




# for i in range(10):
#     t = threading.Thread(target=task, args=(i,))
#     t.start()


# 多个人同时使用锁，一批批操作
# import threading, time
#
# v = 10
#
# lock = threading.BoundedSemaphore(3) #同时三个使用该锁
# # lock = threading.RLock()  # 递归锁
#
#
# def task(arg):
#
#     lock.acquire()  # 申请使用锁，其他线程等待
#     time.sleep(1)
#     global v
#     v -= 1
#     print(v)
#     lock.release()  # 交还释放
#
# for i in range(10):
#     t = threading.Thread(target=task, args=(i,))
#     t.start()

# 时间锁.解脱锁的限制
# import threading, time
#
# v = 10
#
# lock = threading.Event()  # 同时三个使用该锁
#
#
# # lock = threading.RLock()  # 递归锁
#
#
# def task(arg):
#     time.sleep(1)
#     lock.wait()  # 锁住多有线程
#     print(arg)
#
#
# for i in range(10):
#     t = threading.Thread(target=task, args=(i,))
#     t.start()
#
# while True:
#     value = input('>>>')
#     if value == '1':
#         lock.set()
#         lock.clear()

#想怎么锁就怎么锁
import threading, time

v = 10

lock = threading.Condition()  # 同时三个使用该锁


# lock = threading.RLock()  # 递归锁


def task(arg):
    time.sleep(1)
    lock.acquire()
    lock.wait()# 锁住多有线程
    print(arg)
    lock.release()


for i in range(10):
    t = threading.Thread(target=task, args=(i,))
    t.start()

while True:
    value = input('>>>')
    lock.acquire()
    lock.notify(int(value))
    lock.release()

进程对象的方法和属性

from multiprocessing import Process
import time, 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=('egon',))
#     p2 = Process(target=piao, args=('lh',))
#     p3 = Process(target=piao, args=('apla',))
#     p4 = Process(target=piao, args=('tom',))
#
#     # 开启顺序不一定，谁先执行完谁打印
#     p1.start()  # 不是阻塞操作，在发系统调用
#     p2.start()  # 不是阻塞操作，在发系统调用
#     p3.start()  # 不是阻塞操作，在发系统调用
#     p4.start()  # 不是阻塞操作，在发系统调用
#
#     # p1.join()  # 主进程等待p1执行完毕，hold
#     # p1.join()  # 主进程等待
#     # p2.join()
#     # p3.join()
#     # p4.join()
#     p_l = [p1, p2, p3, p4]
#     for p in p_l:
#         p.start()
#
#     for p in p_l:
#         p.join()
#     print('主进程')  # 每次都是主进程先打印，原因是cpu性能低


# 守护进程
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=('egon',))
    p1.daemon = True  # 主进程运行完成后子进程执行完也会被回收
    p1.start()
    print(p1.pid)
    print(p1.name)
    print(p1.is_alive())  # True
    p1.terminate()  # 把p1干掉，僵尸进程
    time.sleep(0.5)
    print(p1.is_alive())  # False 不能立即干掉进程
    print('主进程')

进程同步之模拟抢票

from multiprocessing import Process, Lock
import json, time, random


def work(dbfile, name, lock):
    # lock.acquire()  # 加锁，谁先拿到，其他进程无法使用，加锁是并发，join是排队
    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('%s抢票成功！！！' % name)
            else:
                print('%s抢票失败！！！' % name)
                # lock.release()  # 锁要释放，不然其他的进程无法使用


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('主进程')

生产者消费者模型

from multiprocessing import Process, JoinableQueue
import random, time


def consumer(q, name):
    while True:
        # time.sleep(random.randint(1, 3))
        res = q.get()  # 一直在取状态

        q.task_done()  # 执行完成
        # if res is None:
        #     break
        print('\033[41m消费者%s拿到%s\033[0m' % (name, res))


def producer(seq, q, name):
    for item in seq:
        # time.sleep(random.randint(1, 3))
        q.put(item)

        print('\033[42m生产者%s拿到%s\033[0m' % (name, item))
    # q.put(None)
    q.join()
    print('**************>>')  # 本行执行说明生产者执行完成，消费者把所有任务执行


if __name__ == '__main__':
    q = JoinableQueue()
    c = Process(target=consumer, args=(q, 'lh'))
    c.daemon=True #设置守护进程，朱金城结束，c就结束
    c.start()

    seq = ['包子%s' % i for i in range(10)]
    p = Process(target=producer, args=(seq, q, '厨师1'))
    p.start()  # 生产者进程

    p.join()  # 主进程等待p结束，p等待c把数据取完，c一旦取完数据，p.join就不在阻塞，进而追进程结束
    # ，主进程结束会回收守护进程c，而且此时c也没有存在的必要只要join生产者就行，生产者在等待消费者取走
    # c.join()
    print('主进程')

# [i for i in range(10)] 列表生成式
# [i for i in range(10) if i > 5] 列表生成式

关键字：