9）網(wǎng)絡(luò)并發(fā) 編程、進(jìn)程、線程和協(xié)程

                          并發(fā)網(wǎng)絡(luò)

Server：

#!/usr/bin/env python
#-*-conding:utf-8-*-
import socket
sk = socket.socket()
sk.bind(('127.0.0.1',9009))
sk.listen()
conn,addr = sk.accept()
conn.send(b'heooo')
msg = conn.recv(1024)
print(msg)
conn.close()
sk.close()

Client：

#!/usr/bin/env python
#-*-conding:utf-8-*-
import socket
sk = socket.socket()
sk.connect(('127.0.0.1',9009))
print(sk.recv(1024))
sk.send(b'busdf')
sk.close()

socketserver并發(fā)的模塊

模塊：

#!/usr/bin/env python
#-*-conding:utf-8-*-
import socketserver
class Myserver(socketserver.BaseRequestHandler):
    def handle(self):
        conn = self.request
        print(conn)
myserver = socketserver.ThreadingTCPServer(('127.0.0.1',9988),Myserver)
myserver.serve_forever()

Server端：

 Client端：

并發(fā)編程：

一個(gè)程序在同一時(shí)刻做多件事情

基于多道批處理系統(tǒng)和分時(shí)系統(tǒng)

解決程序中IO操作影響程序的效率問題

多個(gè)程序、作業(yè)在遇到IO操作的時(shí)候，操作系統(tǒng)會(huì)幫助你進(jìn)行切換

讓cpu的利用率得到最大的提高

操作系統(tǒng) 只負(fù)責(zé)管理調(diào)度進(jìn)程

什么叫進(jìn)程：

運(yùn)行中的程序：

進(jìn)程：是操作系統(tǒng)總資源分配的最小單位

每一個(gè)運(yùn)行中的程序都需要有自己的內(nèi)存、資源

都分配給進(jìn)程  紀(jì)錄執(zhí)行的狀態(tài) 管理自己的內(nèi)存資源

import os,time
print(os.getpid())
time.sleep(1000)

Multiprocess模塊（包）異步執(zhí)行，就是兩個(gè)程序一起執(zhí)行

import os

import time
from  multiprocessing import Process
def lxf(*args):
    print(args,os.getpid())
    time.sleep(2)

if __name__=='__main__':
    print(os.getpid())
    p = Process(target=lxf,args=(12,10,))
    p.start()
    print(os.getpid())

import os
import time
from  multiprocessing import Process
def func(num):
    print(num,os.getpid())
    time.sleep(10)
if __name__=='__main__':
    print(os.getpid())
    p = Process(target=func,args=(10,))
    p.start()
    print(os.getpid())

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__':
    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)

什么是進(jìn)程： 運(yùn)行中的程序，計(jì)算機(jī)中最小的資源分配單位

程序開始執(zhí)行就會(huì)產(chǎn)生一個(gè)主進(jìn)程

Python中可以主進(jìn)程中用代碼啟動(dòng)一個(gè)進(jìn)程----------------子進(jìn)程

同時(shí)主進(jìn)程也唄稱為父進(jìn)程

父子進(jìn)程之間的代碼執(zhí)行是一步的，各自執(zhí)行自己的

父子進(jìn)程之間的數(shù)據(jù)是不可以共享

from  multiprocessing import Process
import time
n = 100
def func():
    global n
    n = 0
    print('&&&&')

if __name__ =='__main__':
    Process(target=func).start()
    time.sleep(1)
    print(n)

開啟多個(gè)子進(jìn)程：

from multiprocessing import Process
import time
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()

開啟10個(gè)進(jìn)程去發(fā)郵件：

from multiprocessing import Process
import time
def func(n):
    time.sleep(1)
    print('-' * n)
if __name__ == '__main__':
    for i in range(10):
        p = Process(target=func, args=(1,))
        p.start()
    print('子進(jìn)程開始了')
    p.join()
    print('十條信息發(fā)送完了')

完善10條短信發(fā)送：

import os
import time
from  multiprocessing import Process
def func(n):
    time.sleep(1)
    print('&'*n)

if __name__=='__main__':
    l = []
    for i in range(10):
        p = Process(target=func,args=(i,))
        p.start()
        l.append(p)
    for p in l:
        p.join()
    print('發(fā)完了10條短信')

守護(hù)進(jìn)程：也是一個(gè)子進(jìn)程

當(dāng)主進(jìn)程的代碼執(zhí)行完畢之后會(huì)自動(dòng)結(jié)束的子進(jìn)程交做守護(hù)進(jìn)程

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

、

守護(hù)進(jìn)程

守護(hù)進(jìn)程也是一個(gè)子進(jìn)程

當(dāng)主進(jìn)程的代碼執(zhí)行完畢之后自動(dòng)結(jié)束的子進(jìn)程叫做守護(hù)進(jìn)程

如果要在主進(jìn)程當(dāng)中要等待子進(jìn)程結(jié)束之后在執(zhí)行某段代碼：join方法

如果有多個(gè)子進(jìn)程，不能在start一個(gè)進(jìn)程之后就立刻join，吧所有的進(jìn)程放到列表中，等待所有進(jìn)程都start之后在逐一join

import os
import time
from  multiprocessing import Process
def daemon_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=daemon_func)
    p.daemon = True
    p.start()
    for i in range(3):
        print(i*'*')
        time.sleep(1)

鎖：

例子：

import os
import time
import random
from  multiprocessing import Process

def work(n):
    print('%s:%s is runing' %(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=(1,))
        p.start()

改進(jìn)一下：使用鎖：

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

def work(n,lock):
    lock.acquire()
    print('%s:%s is runing' %(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=(1,lock))
        p.start()

搶火車票：使用鎖：犧牲效率，但是保證安全

from multiprocessing import Process,Lock
import time,json,random
def search():
    dic = json.load(open('db'))
    print('\033[43m剩余票數(shù)%s\033[0m'%dic['count'])
def get(num):
    dic=json.load(open('db'))
    time.sleep(random.random())
    if dic['count'] >0:
        dic['count']-=1
        time.sleep(0.2)
        json.dump(dic,open('db','w'))
        print('\033[43m%s購(gòu)票成功\033[0m'%num)

def task(num,lock):
    search()
    lock.acquire()
    get(num)
    lock.release()


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

信號(hào)量：

#!/usr/bin/env python
#-*-conding:utf-8 -*-
from multiprocessing import Semaphore
sem = Semaphore(5)
sem.acquire()
print(1)
sem.acquire()
print(2)
sem.acquire()
print(3)
sem.acquire()
print(4)
sem.acquire()
print(5)  #到這里就卡住了
sem.acquire()
print(6)
sem.acquire()
print(7)

#!/usr/bin/env python
#-*-conding:utf-8 -*-
from multiprocessing import Process,Semaphore
import time,random
def go_ktv(sem,user):
    sem.acquire()
    print('%s 占到了一間×××包間'%(user))
    time.sleep(random.randint(3,5))# 模擬每個(gè)人在里面呆的時(shí)間不同
    sem.release()
    print('%s 走出×××包間' % (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('========>')

事件：

內(nèi)部?jī)?nèi)置了一個(gè)標(biāo)志

Wait 方法 如果這個(gè)標(biāo)志是True 那么wait == pass

Wait 方法 如果這個(gè)標(biāo)志是False 那么wait 會(huì)陷入阻塞，一直阻塞到標(biāo)志從False變成True

一個(gè)時(shí)間在創(chuàng)建之初 內(nèi)部的標(biāo)志默認(rèn)是False

紅綠燈

隊(duì)列：Queue：

自己人為的去傳輸?shù)慕Y(jié)果的東西

import os
import time
from  multiprocessing import Queue,Process
def func(n,q):
    q.put(n*n)

if __name__=='__main__':
    q = Queue()
    Process(target=func,args=(100,q,)).start()
    print(q.get())

管道+鎖 ==隊(duì)列

管道也是一個(gè)可以實(shí)現(xiàn)進(jìn)程之間通信的模型

但是管道沒有鎖，數(shù)據(jù)不安全

數(shù)據(jù)共享：

線程：

啟動(dòng)線程：

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



Thread(target=func,args=(1,)).start()
print(10*'*')
print('main:',os.getpid())

線程之間的數(shù)據(jù)是共享的：

實(shí)現(xiàn)線程之間并發(fā)：

GIL鎖  全局解釋器鎖：

是cpython解釋器的問題

在同一個(gè)進(jìn)程中 同一時(shí)刻  只能有一個(gè)線程唄cpu執(zhí)行

導(dǎo)致高計(jì)算型， 代碼 不適合用python的多線程來解決

用多進(jìn)程或者分布式來解決高計(jì)算型代碼

守護(hù)線程：

線程鎖

池的概念：

線程池：

import time
from concurrent.futures import ThreadPoolExecutor
def func(num):
    time.sleep(5)
    print(num)

t = ThreadPoolExecutor(20)
for i in range(100):
    t.submit(func,i)

t.shutdown()  #join 整個(gè)池子

print(3333335555) 

協(xié)程：