进程 线程（二）

进程

启动多个进程，进程之间通过操作系统调用，操作系统又有一个时间片的概念，这样多CPU的话就可以调用多个进程。

线程

启动多个线程，真正被CPU执行的最小单位是线程，在Cpython中由于GIL锁的概念，同一时刻只能有一个线程工作。再其它语言中允许用一时间调用多个线程执行。不会影响高IO的操作。

弊端：开启一个线程，创建一个线程 需要创建寄存器，堆栈。
 

协程

本质就是一个线程，在多个任务之间来回调用，不需要创建寄存器，堆栈。

进程：
    进程是一个正在运行的程序，程序不能单独运行，只有将程序装载到内存中，系统为它分配资源才能运行，而这种执行的程序称之为进程。
进程的提出是在多道编程中，我们允许多个程序同时加载到内存中，在操作系统的调度下，可以实现并发地执行，大大提高了CPU的利用率。进程就是为了在CPU上实现多道编程而提出的。

但是进程也有很多缺点例如：
    - 进程只能在同一时间干一件事，如果同时干两件事或多件事，进程就无能为力了。
    - 进程在执行的过程中如果出现阻塞，例如等待输入，整个进程会被挂起。
    - 进程开销大，创建，撤销与切换存在较大的时空开销。

后来出现了线程，线程是CPU调度的最小单位，每个进程中至少有一个线程。

线程比进程的优势：
    - 地址空间和其它资源(如打开文件)：进程间相互独立，同一进程的各线程共享。
    - 通信： 进程间通信IPC，线程间可以直接读写进程的资源。
    - 调度和切换：线程上下文切换要比进程上下文切换要快得多。

全局解释器锁GIL
    - 由于CPython解释器锁的原因，同一时刻只能有一个线程运行，所以Python在CPython解释器中的多线程形同虚设。


递归锁和互斥锁
    递归锁(RLock)是为了解决死锁问题，且在线程中，递归锁可以被acquire多次。
    互斥锁(Lock)

信号量        KTV同一时刻只能有几个人进入到这个房间  同一时间只能有N个线程处理

事件

　　

创建进程两种方式

守护进程

进程对象.deamon  值为True的时候，表示新的子进程是一个守护进程，守护进程随着主进程代码色执行结束而结束。在start之前运行

　　

进程同步控制 

进程锁 

为了数据安全
事例：12306售票一个文件{"ticket":1}，有1张票，有10个人过来买票，每个人都看到为1张票，抢票。

# 锁

# 火车票
import json
import time
from multiprocessing import Process
from multiprocessing import Lock

# def show(i):
#     with open('ticket') as f:
#         dic = json.load(f)
#     print('余票: %s'%dic['ticket'])

def buy_ticket(i,lock):
    lock.acquire() #拿钥匙进门
    with open('ticket') as f:
        dic = json.load(f)
        time.sleep(0.1)
    if dic['ticket'] > 0 :
        dic['ticket'] -= 1
        print('33[32m%s买到票了33[0m'%i)
    else:
        print('33[31m%s没买到票33[0m'%i)
    time.sleep(0.1)
    with open('ticket','w') as f:
        json.dump(dic,f)
    lock.release()      # 还钥匙

if __name__ == '__main__':
    # for i in range(10):
    #     p = Process(target=show,args=(i,))
    #     p.start()
    lock = Lock()
    for i in range(10):
        p = Process(target=buy_ticket, args=(i,lock))
        p.start()

信号量

事例：KTV20个人想进房间，一个房间同时只能有4个人。

# 多进程中的组件
# ktv
# 4个
# 一套资源  同一时间 只能被n个人访问
# 某一段代码 同一时间 只能被n个进程执行
import time
import random
from multiprocessing import Process
from multiprocessing import Semaphore

def ktv(i,sem):
    sem.acquire()    #获取钥匙
    print('%s走进ktv'%i)
    time.sleep(random.randint(1,5))
    print('%s走出ktv'%i)
    sem.release()   


if __name__ == '__main__' :
    sem = Semaphore(4)
    for i in range(20):
        p = Process(target=ktv,args=(i,sem))
        p.start()

事件

一个事件被创建之后，默认是阻塞状态。
set和clear 分别用来修改一个时间的状态 True或False
is_set 	   用来查看一个事件的状态
wait	   依据事件的状态决定自己是否阻塞，False阻塞，True不阻塞。如果为True代码执行。
事例： 20辆车等红绿灯

# 红绿灯事件
import time
import random
from multiprocessing import Event,Process
def cars(e,i):
    if not e.is_set():
        print('car%i在等待'%i)
        e.wait()    # 阻塞 直到得到一个 事件状态变成 True 的信号
    print('33[0;32;40mcar%i通过33[0m' % i)

def light(e):
    while True:
        if e.is_set():
            e.clear()
            print('33[31m红灯亮了33[0m')
        else:
            e.set()
            print('33[32m绿灯亮了33[0m')
        time.sleep(2)

if __name__ == '__main__':
    e = Event()
    traffic = Process(target=light,args=(e,))
    traffic.start()
    for i in range(20):
        car = Process(target=cars, args=(e,i))
        car.start()
        time.sleep(random.random())

进程通信 IPC

队列

两个队列通信

from multiprocessing import Queue,Process
def produce(q):
    q.put('hello')

def consume(q):
    print(q.get())

if __name__ == '__main__':
    q = Queue()
    p = Process(target=produce,args=(q,))
    p.start()
    c = Process(target=consume, args=(q,))
    c.start()

  

管道

进程池

#为什么会有进程池的概念？
	每开启一个进程，就会创建属于这个进程的内存空间(寄存器 堆栈 文件)，所以开启多个进程会耗时。
	进程过多，操作系统需要调度，当切换时，需要当前进程保存状态，所以不会无休止的任由进程的创建。
	进程池刚开始会在进程池中创建4个进程，无论多少个任务来时都由这几个进程处理。　　

进程池基本事例

import time
from multiprocessing import Pool,Process

def func(n):
    for i in range(10):
        print(n+1)

if __name__ == '__main__':

    start = time.time()
    pool = Pool(5)               # 5个进程
    pool.map(func,range(100))    # 100个任务
    t1 = time.time() - start



    start1 = time.time()
    p_lst = []
    for i in range(100):
        p = Process(target=func,args=(i,))
        p_lst.append(p)
        p.start()
    for p in p_lst :p.join()
    t2 = time.time() - start1
    print(t1,t2)




------------结果


0.03202390670776367 0.18317079544067383

异步开启子进程 

import os
import time
from multiprocessing import Pool
def func(n):
    print('start func%s'%n,os.getpid())
    time.sleep(1)
    print('end func%s' % n,os.getpid())

if __name__ == '__main__':
    p = Pool(5)
    for i in range(10):
        p.apply_async(func,args=(i,))

--------结果：
没有任何输出


import os
import time
from multiprocessing import Pool
def func(n):
    print('start func%s'%n,os.getpid())
    time.sleep(1)
    print('end func%s' % n,os.getpid())

if __name__ == '__main__':
    p = Pool(5)
    for i in range(10):
        p.apply_async(func,args=(i,))
    p.close()          # 结束进程池接收任务
    p.join()           # 感知进程池中的任务执行结束


--------结果：
子进程正常输出，主进程等待子进程结束而结束。

基于进程池实现socketserver 

#服务端

import socket
from multiprocessing import Pool

def func(conn):
    conn.send(b'hello')
    print(conn.recv(1024).decode('utf-8'))
    conn.close()

if __name__ == '__main__':
    p = Pool(5)
    sk = socket.socket()
    sk.bind(('127.0.0.1',8080))
    sk.listen()
    while True:
        conn, addr = sk.accept()
        p.apply_async(func,args=(conn,))
    sk.close()






#客户端
import socket

sk = socket.socket()
sk.connect(('127.0.0.1',8080))

ret = sk.recv(1024).decode('utf-8')
print(ret)
msg = input('>>>').encode('utf-8')
sk.send(msg)
sk.close()

线程

socket聊天并发效果

#服务端

import socket
from threading import Thread

def chat(conn):
    conn.send(b'hello')
    msg = conn.recv(1024).decode('utf-8')
    print(msg)
    conn.close()

sk = socket.socket()
sk.bind(('127.0.0.1',8080))
sk.listen()
while True:
    conn,addr = sk.accept()
    Thread(target=chat,args = (conn,)).start()
sk.close()


#客户端

import socket

sk = socket.socket()
sk.connect(('127.0.0.1',8080))

msg = sk.recv(1024)
print(msg)
inp = input('>>> ').encode('utf-8')
sk.send(inp)
sk.close()

递归锁和互斥锁

递归锁(RLock)是为了解决死锁问题，且在线程中，递归锁可以被acquire多次。
互斥锁(Lock)

信号量

KTV同一时刻只能有几个人进入到这个房间 同一时间只能有N个线程处理

Event 事件 

# 事件被创建的时候
# False状态
    # wait() 阻塞
# True状态
    # wait() 非阻塞
# clear 设置状态为False
# set  设置状态为True


#  起两个线程
#  第一个线程 : 连接数据库
        # 等待一个信号 告诉我我们之间的网络是通的
        # 连接数据库
#  第二个线程 : 检测与数据库之间的网络是否连通
        # time.sleep(0,2) 2
        # 将事件的状态设置为True
import time
import random
from threading import Thread,Event

def connect_db(e):
    count = 0
    while count < 3:
        e.wait(0.5)       # 状态为False的时候,我只等待1s就结束
        if e.is_set() == True:
            print('连接数据库')
            break
        else:
            count += 1
            print('第%s次连接失败'%count)
    else:
        raise TimeoutError('数据库连接超时')


def check_web(e):
    time.sleep(random.randint(0,3))
    e.set()

e = Event()
t1 = Thread(target=connect_db,args=(e,))
t2 = Thread(target=check_web,args=(e,))
t1.start()
t2.start()

协程 

在一个线程中实现并发效果
能够规避一些任务中IO操作
在任务的执行过程中，检测到IO就切换到其他任务

　　

基于协程爬虫 

from gevent import monkey;monkey.patch_all()
import gevent
from urllib.request import urlopen    # 内置的模块
def get_url(url):
    response = urlopen(url)
    content = response.read().decode('utf-8')
    return len(content)

g1 = gevent.spawn(get_url,'http://www.baidu.com')
g2 = gevent.spawn(get_url,'http://www.sogou.com')
g3 = gevent.spawn(get_url,'http://www.taobao.com')
g4 = gevent.spawn(get_url,'http://www.hao123.com')
g5 = gevent.spawn(get_url,'http://www.cnblogs.com')
gevent.joinall([g1,g2,g3,g4,g5])
print(g1.value)
print(g2.value)
print(g3.value)
print(g4.value)
print(g5.value)

协程实现socketserver 

#服务端

from gevent import monkey;monkey.patch_all()
import socket
import gevent
def talk(conn):
    conn.send(b'hello')
    print(conn.recv(1024).decode('utf-8'))
    conn.close()

sk = socket.socket()
sk.bind(('127.0.0.1',8080))
sk.listen()
while True:
    conn,addr = sk.accept()
    gevent.spawn(talk,conn)
sk.close()




#客户端

import socket
sk = socket.socket()
sk.connect(('127.0.0.1',8080))
print(sk.recv(1024))
msg = input('>>>').encode('utf-8')
sk.send(msg)
sk.close()