并发编程

一:进程:

1:进程的理论

1 何为进程:
    进程指的是一个正在运行/进行的程序,进程是用来描述程序执行过程的虚拟的概念
    进程VS程序
    程序:一堆代码
    进程:程序的执行过程

    进程的概念起源于操作系统,进程是操作系统最核心的概念,操作系统中其他所有的概念都是围绕进程来的

   2 操作系统理论:
        1 操作系统是什么: 操作系统是一个协调管理控制计算机硬件资源和应用软件资源的一段操作程序
        2两大功能:
            1 将复杂的硬件操作封装为简单的端口给用户或者应用程序qushiypng
            2 将多个进程对硬件的竞争变得有序

         3 操作系统的发展史
            1并发:多个任务看起来是同时运行的
            2 串行;一个任务完完整整的运行完毕,才能开始下一个任务

            3 多道技术:(复用=>共享/公用)
                1 空间上的复用:多个任务复用内存空间
                2 时间上的复用:多个任务复用CPU的时间
                注: 1 一个任务占用CPU时间过长也会被操作系统强行剥夺走CPU的执行权限,比起串行执行反而会降低效率

                2一个任务遇到io操作也会被操作系统强行剥夺走CPU的执行权限,比起串行执行可以提升效率

2:开启进程的方式:

from multiprocessing import Process
import time
def task(name):
    print('%s is running'%name)
    time.sleep(3)
    print('%s is done'%name)
# 在windows 中,开启子进程的操作必须放到if __name__ == '__main__':中

if __name__ == '__main__':
    p=Process(target=task,args=('andy',))
    p.start()  # 向操作系统发了一个开启子进程的信号
    print('hello')

from multiprocessing import Process
import time

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

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

if __name__ == '__main__':
    p=Myprocess('andy')
    p.start()  # 向操作系统发送开启子进程的信号
    print('哈哈哈哈哈')

3:join方法

# join:让主进程在原地等待,等待子进程运行完毕,不会影响子进程的执行
from multiprocessing import Process
import random
import time
def task(name,n):
    print('%s is running'%name)
    time.sleep(n)
    print('%s is done'%name)

if __name__ == '__main__':
    start=time.time()
    P_l=[]
    for i in range(1,4):
        p=Process(target=task,args=('子%s'%i,i))
        p.start()
        P_l.append(p)


    for i in P_l:
        i.join()

    print('哈哈哈',(time.time()-start))

    print('哈哈哈哈')

4 进程之间内存空间互相隔离

from multiprocessing import Process
x=100
def task():
    global x
    x=10

if __name__ == '__main__':
    p=Process(target=task)
    p.start()
    p.join()
    print(x)
# 在内存中,进程之间存在物理隔离

5 进程对象其他相关的属性或方法

current_process 实现

1. 进程pid:每一个进程在操作系统内都有一个唯一的id号,称之为pid
from multiprocessing import Process,current_process
import time

def task():
    print('%s is running' %current_process().pid)
    time.sleep(30)
    print('%s is done' %current_process().pid)

if __name__ == '__main__':
    p=Process(target=task)
    p.start()
    print('主',current_process().pid)

os模块实现

from multiprocessing import Process,current_process
import time,os

def task():
    print('%s is running 爹是:%s' %(os.getpid(),os.getppid()))
    time.sleep(30)
    print('%s is done 爹是:%s' %(os.getpid(),os.getppid()))


if __name__ == '__main__':
    p=Process(target=task)
    p.start()
    print('主:%s 主他爹:%s' %(os.getpid(),os.getppid()))

2. 进程对象其他相关的属性或方法

rom multiprocessing import Process,current_process
import time,os

def task():
    print('%s is running 爹是:%s' %(os.getpid(),os.getppid()))
    time.sleep(30)
    print('%s is done 爹是:%s' %(os.getpid(),os.getppid()))


if __name__ == '__main__':
    p=Process(target=task,name='子进程1')
    p.start()
    # print(p.name)
    p.terminate()
    # time.sleep(0.1)
    print(p.is_alive())
    print('主:%s 主他爹:%s' %(os.getpid(),os.getppid()))

6 僵尸进程和孤儿进程

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__':
    p=Process(target=task,args=('andy',))
    p.start()
    print('主进程')
    p.join()

7 守护进程

# 守护进程:本质就是一个'子进程',该"子进程"的声明周期<=被守护进程的生命周期
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__':
    start=time.time()
    p=Process(target=task,args=('alex',))
    p.start()
    print('朱进程已经狗带了',(time.time()-start))
    p.join()

8 互斥锁

import json
import time,random
from multiprocessing import Process,Lock
def search(name):
    with open('db.json','rt',encoding='utf-8')as f:
        dic=json.load(f)
    time.sleep(1)
    print('%s 查看到的余票为%s'%(name,dic['count']))

def get(name):
    with open('db.json','rt',encoding='utf-8')as f:
        dic=json.load(f)

    if dic['count']>0:
        dic['count']-=0
        time.sleep(random.randint(1,3))
        with open('db.json','wt',encoding='utf-8')as f:
            json.dump(dic,f)
            print('%s 购票成功' % name)
    else:
        print('%s 查看到没有票了' % name)

def task(name,mutex):
    search(name) #并发
    mutex.acquire()
    get(name) #串行
    mutex.release()

    # with mutex:


    #     get(name)

if __name__ == '__main__':
    mutex = Lock()
    for i in range(10):
        p=Process(target=task,args=('路人%s' %i,mutex))
        p.start()
        # p.join() # join只能将进程的任务整体变成串行

ps:补充egon博客()进程)

理论：http://www.cnblogs.com/linhaifeng/articles/7430066.html

链接：http://www.cnblogs.com/linhaifeng/articles/7428874.html

 

二:线程

1进程间通信(IPC机制)

实现方式一

from multiprocessing import Queue

q=Queue(3)
q.put(['first',],block=True,timeout=3)
q.put({'x':2},block=True,timeout=3)
q.put(3,block=True,timeout=3)
q.put(4,block=True,timeout=3)

q.put_nowait(1) #q.put(1,block=False)
q.put_nowait(2)
q.put_nowait(3)
#q.put_nowait(4)

实现方式二

from multiprocessing import Queue

q=Queue()
q.put(['first',])
q.put({'x':2})
q.put(3)
# q.put(4)

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

实现方式三

from multiprocessing import Queue
q=Queue(3)
q.put(['first',],block=True,timeout=3)
q.put({'x':2},block=True,timeout=3)
q.put(3,block=True,timeout=3)
q.put(4,block=True,timeout=3)


print(q.get_nowait()) #q.get(block=false)
print(q.get_nowait()) #q.get(block=false)
print(q.get_nowait()) #q.get(block=false)
# print(q.get_nowait()) #q.get(block=false)

2 生产者消费者模型

1. 什么是生产者消费者模型
    生产者:代指生产数据的任务
    消费者:代指处理数据的任务
    该模型的工作方式:
        生产生产数据传递消费者处理

        实现方式:
            生产者---->队列<------消费者

2. 为何要用
    当程序中出现明细的两类任务,一类负责生产数据,一类负责处理数据
    就可以引入生产者消费者模型来实现生产者与消费者的解耦合,平衡生产能力与消费能力,从提升效率

怎么用????

Queue版本的

是利用from mutiprocessing import Process,Queue实现的

import time,random
from multiprocessing import Process,Queue

def producer(name,food,q):
    for i in range(3):
        res='%s%s' %(food,i)
        time.sleep(random.randint(1,3)) #模拟生产数据的时间
        q.put(res)
        print('厨师[%s]生产了<%s>' %(name,res))

def consumer(name,q):
    while True:
        res=q.get()
        if res is None:break
        time.sleep(random.randint(1,3)) #模拟处理数据的时间
        print('吃货[%s]吃了<%s>' %(name,res))

if __name__ == '__main__':
    q=Queue()
    # 生产者们
    p1=Process(target=producer,args=('小Egon','泔水',q))
    p2=Process(target=producer,args=('中Egon','屎包子',q))
    p3=Process(target=producer,args=('大Egon','腰子汤',q))
    # 消费者们
    c1=Process(target=consumer,args=('刘清正',q))
    c2=Process(target=consumer,args=('吴三江',q))

    p1.start()
    p2.start()
    p3.start()
    c1.start()
    c2.start()

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

JoinableQueue版本的

是利用from mutiprocessing import Process,JoinableQueue实现的

import time,random
from multiprocessing import Process,JoinableQueue

def producer(name,food,q):
    for i in range(3):
        res='%s%s' %(food,i)
        time.sleep(random.randint(1,3)) #模拟生产数据的时间
        q.put(res)
        print('厨师[%s]生产了<%s>' %(name,res))

def consumer(name,q):
    while True:
        res=q.get()
        time.sleep(random.randint(1,3)) #模拟处理数据的时间
        print('吃货[%s]吃了<%s>' %(name,res))
        q.task_done()

if __name__ == '__main__':
    q=JoinableQueue()
    # 生产者们
    p1=Process(target=producer,args=('小Egon','泔水',q))
    p2=Process(target=producer,args=('中Egon','屎包子',q))
    p3=Process(target=producer,args=('大Egon','腰子汤',q))
    # 消费者们
    c1=Process(target=consumer,args=('刘清正',q))
    c2=Process(target=consumer,args=('吴三江',q))
    c1.daemon=True
    c2.daemon=True

    p1.start()
    p2.start()
    p3.start()
    c1.start()
    c2.start()

    p1.join()
    p2.join()
    p3.join()
    q.join() # 主进程等q结束,即q内数据被取干净了
    print('主')

3:线程理论

1 什么是线程
    进程其实一个资源单位,而进程内的线程才是cpu上的执行单位
    线程其实指的就是代码的执行过程

2 为何要用线程
    线程vs进程
        1. 同一进程下的多个线程共享该进程内的资源
        2. 创建线程的开销要远远小于进程

3 如何用线程

4 开启线程的两种方式

from threading import Thread
import time

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

if __name__ == '__main__':
    t=Thread(target=task,args=('线程1',))
    t.start()
    print('主')

from threading import Thread
import time

class Mythread(Thread):
    def run(self):
        print('%s is running' %self.name)
        time.sleep(2)
        print('%s is done' %self.name)

if __name__ == '__main__':
    t=Mythread()
    t.start()
    print('主')

5线程的特性

# 特性一
from threading import Thread
import time

n=100
def task():
    global n
    n=0

if __name__ == '__main__':
    t=Thread(target=task)
    t.start()
    t.join()
    print('主',n)

# 特性二
from threading import Thread
import time,os

def task():
    print('%s is running' %os.getpid())

if __name__ == '__main__':
    t=Thread(target=task)
    t.start()
    print('主',os.getpid())

# 特性三
from threading import Thread,active_count,current_thread
import time,os

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

if __name__ == '__main__':
    t=Thread(target=task,)
    t.start()
    # t.join()
    # print('主',active_count())
    print('主',current_thread().name)

6 守护线程

简单守护

from threading import Thread
import time

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

if __name__ == '__main__':
    t=Thread(target=task,args=('线程1',))
    t.daemon=True
    t.start()
    print('主')

辛苦守护

from threading import Thread
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__':
    # t1=Thread(target=foo)
    # t2=Thread(target=bar)

    t1=Process(target=foo)
    t2=Process(target=bar)
    t1.daemon=True
    t1.start()
    t2.start()
    print("main-------")
#看一看执行的先后

7 线程的互斥锁

from threading import Thread,Lock
import time

mutex=Lock()
n=100
def task():
    global n
    mutex.acquire()
    temp=n
    time.sleep(0.1)
    n=temp-1
    mutex.release()

if __name__ == '__main__':
    t_l=[]
    for i in range(100):
        t=Thread(target=task)
        t_l.append(t)
        t.start()

    for t in t_l:
        t.join()
    print(n)

 8作业小练习

1 实现多线程开发socket

from threading import Thread
from socket import *
s=socket(AF_INET,SOCK_STREAM)
s.bind(('127.0.0.1',8001))
s.listen(5)

def action(conn):
    while True:
        data=conn.recv(1024)
        print('传过来的数是%s'%data)
        conn.send(data.upper())

if __name__ == '__main__':
    while True:
        conn,client_addr=s.accept()
        t=Thread(target=action,args=(conn,))
        t.start()

from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(("127.0.0.1",8001))
while True:
    msg=input(">>>").strip()
    if len(msg)==0:break
    client.send(msg.encode('utf-8'))
    data=client.recv(1024)
    print('回来的是%s'%data.decode('gbk'))

2 练习二：三个任务，一个接收用户输入，一个将用户输入的内容格式化成大写，一个将格式化后的结果存入文件

ps:献上egon多线程

理论：http://www.cnblogs.com/linhaifeng/articles/7430082.html

链接：http://www.cnblogs.com/linhaifeng/articles/7428877.html