码迷,mamicode.com
首页 > 其他好文 > 详细

第十五章 并发编程

时间:2018-06-23 01:37:56      阅读:178      评论:0      收藏:0      [点我收藏+]

标签:color   inter   eve   ali   进程 线程   完全   导致   英文   ffffff   

1.操作系统的发展史                                            

参考:http://www.cnblogs.com/Eva-J/articles/8253521.html

知识点

输入输出 -- 大部分时间都不会占用CPU,但会降低你程序的效率

操作系统的三种基本类型:多道批处理系统、分时系统、实时系统。

 现在操作系统

基于多道批处理系统和分时系统

多个程序、作业在遇到IO操作的时候,操作系统会帮助你进行切换

让CPU的利用率得到最大的提高

2.进程                                                        

初识进程

进程: 运行中的程序

操作系统 只负责管理调度进程

进程是操作系统中资源分配的最小单位

每一个运行中的程序都需要有自己的内存、资源

都分配给进程 记录执行的状态 管理自己的内存资源

在Python中,每一个运行的程序 都是一个进程

如何使用python来开启一个进程--我们使用multiprocessing模块,process-进程,multi多元的

并发编程例:

import os
import time
from multiprocessing import Process
def func(num):
    print(num,os.getpid())
    time.sleep(0.5)
    print(num,os.getpid())
    time.sleep(0.5)
    print(num, os.getpid())
    time.sleep(0.5)
    print(num, os.getpid())

# 创建一个进程
if __name__ == __main__:  # windows,开启子进程的时候,会执行import
    print(os.getpid())
    p = Process(target=func, args=(10,))
    p.start() # 开启进程
    print(os.getpid())
    time.sleep(1)
    print(os.getpid(),1)
    time.sleep(1)
    print(os.getpid(),2)

# 运行结果
#                  时间轴
# 17500              0
# 17500              0
# 10 12656           0+
# 10 12656           0.5
# 17500 1            1
# 10 12656           1+
# 10 12656           1.5+
# 17500 2            2

几个概念:

同步:一个时刻只能做一件事

异步:同时做多件事

异步可以有效地提高程序的效率

进程:主进程中开启的新的进程

主进程:运行的这个程序

父进程:创造出子进程的进程

注意:进程与进程之间都是异步的,而且开启一个进程是有时间开销的

进程的进阶

什么是进程:运行中的程序,计算机中最小的资源分配单位

程序开始执行就会产生一个主进程

python中可以主进程中用代码启动一个进程 -- 子进程

同时主进程也被称为父进程父子进程之间的代码执行是异步的,各自执行自己的

父子进程之间的数据不可以共享

例:

import time
from multiprocessing import Process

n = 100
def func():
    global n
    n = 0
    print(n)
    print(----------)

if __name__ == __main__:
    Process(target=func).start()
    time.sleep(1)
    print(n)
# 0
# ----------
# 100
# 说明:父进程中n的值并没有随着子进程的运行而改变

 

进程什么时候结束

主进程什么时候结束:主进程会等待子进程结束之后再结束,父进程会等待回收子进程所占用的资源

例:直接开启多个子进程

import time
from multiprocessing import Process

n = 100
def func(n):
    time.sleep(1)
    print(-*n)

if __name__ == __main__:
    Process(target=func, args=(1,)).start()
    Process(target=func, args=(2,)).start()
    Process(target=func, args=(3,)).start()
    Process(target=func, args=(4,)).start()

# 说明:主要看时间片的轮转,是一个随机事件
# -
# ----
# ---
# --

例:使用for循环开启多个子进程

import time
from multiprocessing import Process

n = 100
def func(n):
    time.sleep(1)
    print(-*n)

if __name__ == __main__:
    for i in range(10):
        Process(target=func, args=(i,)).start()
# -
# --
# --------
# ----
# ---
# -----
# ---------
# ------
# -------

阻塞

例:使用阻塞(join)开启子进程

import time
from multiprocessing import Process

n = 100
def func(n):
    time.sleep(1)
    print(-*n)

if __name__ == __main__:
    p = Process(target=func, args=(1,))
    p.start()
    print(子进程开始了)
    p.join() # 阻塞 直到子进程都执行完毕
    print(十条信息已经都发送完了)
# 子进程开始了
# -
# 十条信息已经都发送完了

例:使用阻塞(join)循环开启多个子进程

import time
from multiprocessing import Process

n = 100
def func(n):
    time.sleep(1)
    print(-*n)

if __name__ == __main__:
    lst = []
    for i in range(10):
        p = Process(target=func, args=(i,))
        p.start()
        lst.append(p)
    for p in lst:p.join()

    print(十条信息已经都发送完了)

# -----
# -
# ---
# -------
# --
# ---------
# ----
# --------
# ------
# 
# 十条信息已经都发送完了

守护进程

守护进程也是一个子进程

定义:当主进程的代码执行完毕之后自动结束的子进程叫做守护进程

例:

import time
from multiprocessing import Process
def deamon_func():
    while True:
        print(我还活着)
        time.sleep(0.5)


if __name__ == __main__:
    p = Process(target=deamon_func)
    # 开启守护进程
    p.daemon = True
    p.start()
    for i in range(3):
        print(i**)
        time.sleep(0.4)

# 我还活着
# *
# 我还活着
# **
# 我还活着
# 说明:正常情况下,父进程会等待子进程结束才结束
# 而设置守护进程后,父进程结束了,子进程的死循环即可停止

练习:分析程序请问会打印多少个我还活着

import time
from multiprocessing import Process
def deamon_func():
    while True:
        print(我还活着)
        time.sleep(0.5)

def wahaha():
    for i in range(10):
        time.sleep(1)
        print(i*#)

if __name__ == __main__:
    Process(target=wahaha).start()
    p = Process(target=deamon_func)
    p.daemon = True
    p.start()
    for i in range(3):
        print(i**)
        time.sleep(1)


# 我还活着
# 我还活着
# *
#
# 我还活着
# 我还活着
# **
# #
# 我还活着
# 我还活着
# ##
# ###
# ####
# #####
# ######
# #######
# ########
# #########
# 说明:因为‘我还活着‘是守护进程打印的,所以当父进程结束后,停止打印,3/0.5=6次
# 但程序不会停止,需要等待子进程wahaha结束

例:主进程添加阻塞

import time
from multiprocessing import Process
def deamon_func():
    while True:
        print(我还活着)
        time.sleep(0.5)

def wahaha():
    for i in range(10):
        time.sleep(1)
        print(i*#)

if __name__ == __main__:
    p2 = Process(target=wahaha)
    p2.start()
    p = Process(target=deamon_func)
    p.daemon = True
    p.start()
    for i in range(3):
        print(i**)
        time.sleep(1)
    p2.join()

# 我还活着
# 我还活着
# *
# 
# 我还活着
# 我还活着
# **
# #
# 我还活着
# 我还活着
# ##
# 我还活着
# 我还活着
# ###
# 我还活着
# 我还活着
# ####
# 我还活着
# 我还活着
# #####
# 我还活着
# 我还活着
# ######
# 我还活着
# 我还活着
# #######
# 我还活着
# 我还活着
# ########
# 我还活着
# 我还活着
# #########
# 我还活着

总结

# 开启一个子进程 start
# 子进程和主进程是异步的
# 如果在主进程中要等待子进程结束之后再执行某行代码:join
# 如果有多个子进程 不能在start一个进程之后就立刻join,把所有的进程放到列表中
# 等待所有的进程都start之后再逐一join
# 守护进程 -- 当主进程的代码执行完毕之后自动结束的子进程叫做守护进程

添加锁是为了保证程序的运行顺序,牺牲了效率但是保护数据的安全

例:未添加阻塞、锁

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

def work(n):
    print(%s: %s is running %(n,os.getpid()))
    time.sleep(random.random())
    print(%s:%s is done %(n,os.getpid()))

if __name__ == __main__:
    for i in range(3):
        p=Process(target=work,args=(i,  ))
        p.start()
# 0: 23192 is running
# 2: 6064 is running
# 1: 22560 is running
# 1:22560 is done
# 0:23192 is done
# 2:6064 is done
# 说明:在没有使用阻塞,锁的时候,因为时间片是随机事件,所以随机子进程开始

例:使用阻塞保证子进程的独立运行

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

def work(n):
    print(%s: %s is running %(n,os.getpid()))
    time.sleep(random.random())
    print(%s:%s is done %(n,os.getpid()))

if __name__ == __main__:
    for i in range(3):
        p=Process(target=work,args=(i,  ))
        p.start()
        p.join()

# 0: 13184 is running
# 0:13184 is done
# 1: 22868 is running
# 1:22868 is done
# 2: 22200 is running
# 2:22200 is done

例:使用锁

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

def work(n, lock):
    lock.acquire()
    print(%s: %s is running %(n,os.getpid()))
    time.sleep(random.random())
    print(%s:%s is done %(n,os.getpid()))
    lock.release()

if __name__ == __main__:
    lock = Lock()
    for i in range(3):
        p=Process(target=work,args=(i,lock))
        p.start()
        
# 0: 18864 is running
# 0:18864 is done
# 1: 23264 is running
# 1:23264 is done
# 2: 15820 is running
# 2:15820 is done

信号量

信号量本质是锁+计数器

from multiprocessing import Process,Semaphore
import time,random

def go_ktv(sem,user):
    sem.acquire()
    print(%s 占到一间ktv小屋 %user)
    time.sleep(random.randint(3,5)) #模拟每个人在ktv中待的时间不同
    sem.release()
    print(%s 走出ktv小屋 % user)

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

    for i in p_l:
        i.join()
    print(============》)

# 信号量的本质就是 锁+计数器
# user12 占到一间ktv小屋
# user8 占到一间ktv小屋
# user0 占到一间ktv小屋
# user7 占到一间ktv小屋
# user12 走出ktv小屋
# user4 占到一间ktv小屋
# user6 占到一间ktv小屋
# user0 走出ktv小屋
# user8 走出ktv小屋
# user3 占到一间ktv小屋
# user7 走出ktv小屋
# user11 占到一间ktv小屋
# ******
# ============》
# 说明:程序设置Semaphore(4),表示同时只能有四个子进程存在,如果有结束的,再添加新的,知道所有子进程结束

事件

事件内部内置了一个标志
wait 方法 如果这个标志是True,那么wait == pass
wait 方法 如果这个标志是False,那么wait就会陷入阻塞,一直阻塞到标志从False变成True

一个事件在创建之初 内部的标志默认是False
False -> True set()
True -> False clear()

红绿灯模型

# 红绿灯模型
from multiprocessing import Process, Event
import time, random

def car(e, n):
    while True:
        if not e.is_set():  # 进程刚开启,is_set()的值是Flase,模拟信号灯为红色
            print(\033[31m红灯亮\033[0m,car%s等着 % n)
            e.wait()    # 阻塞,等待is_set()的值变成True,模拟信号灯为绿色
            print(\033[32m车%s 看见绿灯亮了\033[0m % n)
            time.sleep(random.randint(3, 6))
            if not e.is_set():   #如果is_set()的值是Flase,也就是红灯,仍然回到while语句开始
                continue
            print(车开远了,car, n)
            break

def traffic_lights(e, inverval):
    while True:
        time.sleep(inverval)   # 先睡3秒
        if e.is_set():         # 标志是True
            print(######, e.is_set())
            e.clear()  # ---->将is_set()的值设置为False
        else:                 # 标志是False
            e.set()    # ---->将is_set()的值设置为True
            print(***********,e.is_set())


if __name__ == __main__:
    e = Event()   #e就是事件
    t = Process(target=traffic_lights, args=(e, 3))  # 创建一个进程控制红绿灯
    for i in range(10):
        p=Process(target=car,args=(e,i,))  # 创建是个进程控制10辆车
        p.start()
    t.start()

    print(============》)



# 10个进程 模拟车 :车的行走要依靠当时的交通灯
# 交通灯是绿灯 车就走
# 交通灯是红灯 车就停 停到灯变绿
# wait 来等灯
# set clear 来控制灯

队列

管道+锁 = 队列

管道也是一个可以实现进程之间通信的模型

但是管道没有锁,数据不安全

from multiprocessing import Queue,Process

def func(q,num):
    try:
        t = q.get_nowait()
        print("%s抢到票了"%num)
    except:pass

if __name__ == __main__:
    q = Queue()
    q.put(1)
    for i in range(10):
        Process(target=func,args=(q,i)).start()

3.线程                                                        

什么是进程 :是计算机资源分配的最小单位

什么是线程 :是CPU调度的最小单位

线程和进程的关系:

  每个进程中都至少有一个线程

多线程服务的时候才能感受到线程的存在

本质的区别:

  同一个进程的每个线程之间数据共享

  线程之间也是异步的

  线程是轻量级的:创建一个线程的时间开销要远远小于进程

  线程是CPU调度的最小单位

例:线程的开启

import os
import time
from threading import Thread
def func(i):
    print(os.getpid())
    time.sleep(1)
    print(thread,i)

Thread(target=func,args=(1,)).start()
print(123**)
print(main, os.getpid())
# 23280
# ***************************************************************************************************************************
# main 23280
# thread 

总结:

# 每一个进程里至少有一个主线程负责执行代码
# 在主线程中可以再开启一个新的线程
# 在同一个进程中就有两个线程同时在工作了
# 线程才是CPU调度的最小单位
# 多个线程之间的数据是共享的
# GIL锁 全局解释器锁
# 解释器的锅 Cpython解释器的问题
# 在同一个进程中 同一时刻 只能有一个线程被CPU执行
# 导致高计算型 代码 不适合用python的多线程来解决
# 用多进程或者分布式来解决高计算型代码

守护线程

例:

from threading import Thread
import time
def foo():
    while True:
        print(123)
        time.sleep(1)

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

t1 = Thread(target=foo)
t2 = Thread(target=bar)

t1.daemon = True
t1.start()
t2.start()
print("main-------")


# 主线程结束了之后守护线程也同时结束
# 守护线程会等待主线程完全结束之后才结束

例:

from threading import Thread,Lock
import time
def work():
    global n
    lock.acquire()
    temp=n
    time.sleep(0.1)
    n = temp-1
    lock.release()
if __name__ == __main__:
    lock=Lock()
    n=100
    l=[]
    for i in range(100):
        p=Thread(target=work)
        l.append(p)
        p.start()
    for p in l:
        p.join()
    print(n)

# 当你的程序中出现了取值计算再赋值的操作 数据不安全 —— 加锁

线程池

程序可以开启的进程数和线程数

CPU的个数+1 进程数

CPU的个数*5 线程数

池的介绍
#1 介绍
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor:线程池,提供异步调用
ProcessPoolExecutor: 进程池,提供异步调用
Both implement the same interface, which is defined by the abstract Executor class.

#2 基本方法
#submit(fn, *args, **kwargs)
异步提交任务

#map(func, *iterables, timeout=None, chunksize=1) 
取代for循环submit的操作

#shutdown(wait=True) 
相当于进程池的pool.close()+pool.join()操作
wait=True,等待池内所有任务执行完毕回收完资源后才继续
wait=False,立即返回,并不会等待池内的任务执行完毕
但不管wait参数为何值,整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前

#result(timeout=None)
取得结果

#add_done_callback(fn)
回调函数

例:

import time
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
def func(num):
    print(num)
    time.sleep(1)
    print(num)
if __name__ == __main__:
    t = ThreadPoolExecutor(20)
    for i in range(100):
        t.submit(func,i)
    t.shutdown()  # join整个池子,必须添加
    print(done)
# 说明:线程会以20个为一组开启

例:使用map取代了for+submit

import os,time,random
from concurrent.futures import ThreadPoolExecutor
def task(n):
    print(%s is runing %os.getpid(),n)
    time.sleep(random.randint(1,3))
    return n**2
if __name__ == __main__:
    executor=ThreadPoolExecutor(max_workers=3)
    executor.map(task,range(1,12)) #map取代了for+submit

callback回掉函数 - 当拿到返回值的时候,立刻进行某操作

我有10个http的网页请求,把这10个网页上的信息分析了

例:

import time
import random
from concurrent.futures import ThreadPoolExecutor
from threading import current_thread
urls=[
        https://www.baidu.com,
        https://www.python.org,
        https://www.openstack.org,
        https://help.github.com/,
        http://www.sina.com.cn/
        http://www.cnblogs.com/
        http://www.sogou.com/
        http://www.sohu.com/
    ]

def analies(content):
    print(分析网页,current_thread())
    print(content.result())

def get_url(url):
    print(爬取网页,current_thread())
    time.sleep(random.uniform(1,3))
    # analies(url*10)
    return url*10

t = ThreadPoolExecutor(3)
print(主线程,current_thread())
for url in urls:
    t.submit(get_url,url).add_done_callback(analies)

concurrent.futures callback是由子线程做的

线程的其他知识与进程类似

4.协程                                                        

协程:是单线程下的并发,又称微线程,纤程。英文名Coroutine。一句话说明什么是线程:协程是一种用户态的轻量级线程,即协程是由用户程序自己控制调度的。

进程 计算机中资源分配的最小单位 cpu+1

线程 CPU 调度最小单位 cpu*5

协程 把一个线程拆分成几个 500

进程 线程 都是操作系统在调度

协程 是程序级别调度

减轻了操作系统的负担、增强了用户对程序的可控性

例1:

from gevent import monkey;monkey.patch_all()
import gevent
import time
def eat(name):
    print(%s eat 1 %name)
    time.sleep(2)
    print(%s eat 2 %name)

def play(name):
    print(%s play 1 %name)
    time.sleep(1)
    print(%s play 2 %name)


g1=gevent.spawn(eat,egon)
g2=gevent.spawn(play,name=egon)
g1.join()
g2.join()
# gevent.joinall([g1,g2])
print()
# egon eat 1
# egon play 1
# egon play 2
# egon eat 2
#

例2:

from gevent import monkey;monkey.patch_all()
import gevent
import requests
import time

def get_page(url):
    print(GET: %s %url)
    response=requests.get(url)
    if response.status_code == 200:
        print(%d bytes received from %s %(len(response.text),url))


start_time=time.time()
gevent.joinall([
    gevent.spawn(get_page,https://www.python.org/),
    gevent.spawn(get_page,https://www.yahoo.com/),
    gevent.spawn(get_page,https://github.com/),
])
stop_time=time.time()
print(run time is %s %(stop_time-start_time))
# GET: https://www.python.org/
# GET: https://www.yahoo.com/
# GET: https://github.com/
# 48801 bytes received from https://www.python.org/
# 527368 bytes received from https://www.yahoo.com/
# 59420 bytes received from https://github.com/
# run time is 6.552013397216797

第十五章 并发编程

标签:color   inter   eve   ali   进程 线程   完全   导致   英文   ffffff   

原文地址:https://www.cnblogs.com/gnaix/p/9214480.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!