進程線程（二）

時間 2019-12-11

標籤進程線程简体版

原文原文鏈接

進程

啓動多個進程，進程之間經過操做系統調用，操做系統又有一個時間片的概念，這樣多CPU的話就能夠調用多個進程。

線程

啓動多個線程，真正被CPU執行的最小單位是線程，在Cpython中因爲GIL鎖的概念，同一時刻只能有一個線程工做。再其它語言中容許用一時間調用多個線程執行。不會影響高IO的操做。

弊端：開啓一個線程，建立一個線程 須要建立寄存器，堆棧。
 

協程

本質就是一個線程，在多個任務之間來回調用，不須要建立寄存器，堆棧。

進程：
    進程是一個正在運行的程序，程序不能單獨運行，只有將程序裝載到內存中，系統爲它分配資源才能運行，而這種執行的程序稱之爲進程。
進程的提出是在多道編程中，咱們容許多個程序同時加載到內存中，在操做系統的調度下，能夠實現併發地執行，大大提升了CPU的利用率。進程就是爲了在CPU上實現多道編程而提出的。

可是進程也有不少缺點例如：
    - 進程只能在同一時間幹一件事，若是同時幹兩件事或多件事，進程就無能爲力了。
    - 進程在執行的過程當中若是出現阻塞，例如等待輸入，整個進程會被掛起。
    - 進程開銷大，建立，撤銷與切換存在較大的時空開銷。

後來出現了線程，線程是CPU調度的最小單位，每一個進程中至少有一個線程。

線程比進程的優點：
    - 地址空間和其它資源(如打開文件)：進程間相互獨立，同一進程的各線程共享。
    - 通訊： 進程間通訊IPC，線程間能夠直接讀寫進程的資源。
    - 調度和切換：線程上下文切換要比進程上下文切換要快得多。

全局解釋器鎖GIL
    - 因爲CPython解釋器鎖的緣由，同一時刻只能有一個線程運行，因此Python在CPython解釋器中的多線程形同虛設。


遞歸鎖和互斥鎖
    遞歸鎖(RLock)是爲了解決死鎖問題，且在線程中，遞歸鎖能夠被acquire屢次。
    互斥鎖(Lock)

信號量        KTV同一時刻只能有幾我的進入到這個房間  同一時間只能有N個線程處理

事件

View Code

建立進程兩種方式python

守護進程web

進程對象.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('\033[32m%s買到票了\033[0m'%i)
    else:
        print('\033[31m%s沒買到票\033[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()

View Code

信號量編程

事例：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()

View Code

事件json

一個事件被建立以後，默認是阻塞狀態。
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('\033[0;32;40mcar%i經過\033[0m' % i)

def light(e):
    while True:
        if e.is_set():
            e.clear()
            print('\033[31m紅燈亮了\033[0m')
        else:
            e.set()
            print('\033[32m綠燈亮了\033[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())

View Code

進程通訊 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()

View Code

管道網絡

進程池

#爲何會有進程池的概念？
	每開啓一個進程，就會建立屬於這個進程的內存空間(寄存器 堆棧 文件)，因此開啓多個進程會耗時。
	進程過多，操做系統須要調度，當切換時，須要當前進程保存狀態，因此不會無休止的任由進程的建立。
	進程池剛開始會在進程池中建立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

開啓100個任務，分別用多進程和進程池執行，發現進程池所花費的時間少

異步開啓子進程併發

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()           # 感知進程池中的任務執行結束


--------結果：
子進程正常輸出，主進程等待子進程結束而結束。

p.apply_async 異步執行子進程

基於進程池實現socketserver app

#服務端

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效果，支持5個進程同時鏈接通訊

線程

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()

基於多線程實現secketserver

遞歸鎖和互斥鎖

遞歸鎖(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()

事件鏈接數據庫，每次等1秒有3次機會

協程

在一個線程中實現併發效果
可以規避一些任務中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()