Python 之網絡式編程

一 客戶端/服務器架構

即C/S架構，包括

1、硬件C/S架構（打印機）

2、軟件B/S架構（web服務）

C/S架構與Socket的關系：

我們學習Socket就是為了完成C/S的開發

二 OSI七層

引子：　　

　　計算機組成原理：硬件、操作系統、應用軟件三者組成。

　　具備以上條件后，計算機就可以工作，如果你要和別人一起玩，那你就需要上網了。互聯網的核心就是由一堆協議組成，協議就是標准。

 

為什么學習Socket之前要先了解互聯網協議？

　　1、C/S架構的軟件（應用軟件屬於應用層）是基於網絡進行通信的

　　2、網絡的核心即一堆協議，協議即標准，想開發一款基於網絡通信的軟件，就必須遵循這些標准

OSI七層：

三 Socket層

四 Socket是什么 

　　Socket是應用層與TCP/IP協議族通信的中間軟件抽象層，它是一組接口，在設計模式中，Socket其實就是一個門面模式，它把負責的TCP/IP協議族隱藏在Socket接口后面，對用戶來說，一組簡單的接口就是全部，讓Socket去組織數據，以符合指定的協議。

　　所以，我們無需深入學習理解TCP/UDP協議，Socket已經為我們封裝好了，我們只需要遵循Socket的規定去編程，寫出的程序自然就是遵循TCP/UDP標准的。

五 套接字發展史及分類

　　套接字起源於20世紀70年代加利福尼亞大學伯克利分校版本的Unix，即人們所說的BSD Unix。因此，有時人們也把套接字成為“伯克利套接字”或“BSD套接字”。一開始，套接字被設計用在一台主機上多個應用程序之間的通信，這也被稱作進程間通許或IPC。套接字有兩種（或者稱為兩個種族），分別是基於文件型和就網絡型。

基於文件類型的套接字家族

套接字家族的名字：AF_UNIX

　　UNIX一切皆文件，基於文件的套接字調用的就是底層的文件系統來取數據，兩個套接字進程運行在同一機器上，可以通過訪問同一文件系統間接完成通信。

基於網絡類型的套接字家族

套接字家族的名字：AF_INET

　　還有AF_INET6被用於ipv6，還有一些其他的地址家族，不過，他們要么是只用於某個平台，要么就是已經被廢棄，或者是很少被使用，或者是根本沒有實現，所有地址家族中，AF_INET是使用最廣泛的一個，Python支持很多地址家族，但是由於我們只關心網絡編程，所以大部分時候我們只使用AF_INET(AF:Address Family;INET:Internet)

六 套接字工作流程

　　生活中，你要打電話給一個朋友，先撥號，朋友聽到電話鈴聲響后接打電話，這時你和你的朋友就建立起了連接，就可以講話了，等交流結束，掛斷電話結束此次通話。

 利用Socket模擬生活中打電話：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import socket

phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # 買手機;socket.AF_INET:基於網絡協議;socket.SOCK_STREAM:基於流的TCP協議
phone.bind(('127.0.0.1', 8080))  # 綁定手機卡;元祖形式，ip地址+端口
# 注：服務器的ip地址寫本機的ip地址
phone.listen(5)  # 開機
conn, addr = phone.accept()  # 等電話
msg = conn.recv(1024)  # 收消息
print('客戶端發來的消息是:', msg)
conn.send(msg.upper())  # 發消息
conn.close()
phone.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import socket
phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # 買手機
phone.connect(('127.0.0.1', 8080))  # 撥電話
# 注：客戶端的ip地址，寫服務器端的ip地址
phone.send('hello'.encode('utf-8'))  # 發消息
data = phone.recv(1024)  # 收消息
print('收到服務端發來的消息', data)

服務器和客戶端無限循環發送消息：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# import socket
from socket import *
import time
ip_port = ('127.0.0.1', 8080)
back_log = 5
buffer_size = 1024
tcp_server = socket(AF_INET, SOCK_STREAM)
tcp_server.bind(ip_port)
tcp_server.listen(back_log)
print('服務端開始運行')
conn, addr = tcp_server.accept()  # 服務器阻塞
print('雙向鏈接', conn)
print('客戶端地址', addr)
while True:
    time.sleep(1)
    print('[%s]' % time.time())
    data = conn.recv(buffer_size)
    print('客戶端發來的消息是', data.decode('utf-8'))
    conn.send(data.upper())
conn.close()
tcp_server.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from socket import *
import time
ip_port = ('127.0.0.1', 8080)
back_log = 5
buffer_size = 1024
tcp_client = socket(AF_INET, SOCK_STREAM)
tcp_client.connect(ip_port)
while True:
    time.sleep(2)
    print('[%s]' % time.time())
    msg = input('>>:').strip()
    tcp_client.send(msg.encode('utf-8'))
    print('客戶端已經發送消息')
    data = tcp_client.recv(buffer_size)
    print('收到服務端發來消息', data.decode('utf-8'))
tcp_client.close()

Socket收發消息原理圖：

若重啟服務端時，可能會遇到：Address already in use;這個是由於服務端扔然存在四次揮手的time_wait狀態占用地址

解決方案：

# 加入一條socket配置，重用ip和端口
tcp_server = socket(AF_INET, SOCK_STREAM)
tcp_server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)  # <---就是這條，在bind前加
tcp_server.bind(ip_port)

發現系統存在大量TIME_WAIT狀態的連接，通過調整linux內核參數解決，
vi /etc/sysctl.conf

編輯文件，加入以下內容：
net.ipv4.tcp_syncookies = 1
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_tw_recycle = 1
net.ipv4.tcp_fin_timeout = 30
 
然后執行 /sbin/sysctl -p 讓參數生效。
 
net.ipv4.tcp_syncookies = 1 表示開啟SYN Cookies。當出現SYN等待隊列溢出時，啟用cookies來處理，可防范少量SYN攻擊，默認為0，表示關閉；

net.ipv4.tcp_tw_reuse = 1 表示開啟重用。允許將TIME-WAIT sockets重新用於新的TCP連接，默認為0，表示關閉；

net.ipv4.tcp_tw_recycle = 1 表示開啟TCP連接中TIME-WAIT sockets的快速回收，默認為0，表示關閉。

net.ipv4.tcp_fin_timeout 修改系統默認的 TIMEOUT 時間

七 基於UDP的套接字

udp服務端

ss = socket()  # 創建一個服務器的套接字
ss.bind()  # 綁定服務器套接字
while True:  # 服務器無限循環
    cs = ss.recvfrom()/ss.sendto()  # 對話（接收與發送）
ss.close()  #  關閉服務器套接字

udp客戶端

cs = socket()  # 創建客戶套接字
while True:
    cs.sendto()/cs.recvfrom()  # 對話（發送/接收）
cs.close()  # 關閉客戶套接字

基於UDP的套接字：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from socket import *
ip_port = ('127.0.0.1', 8080)
buffer_size = 1024
udp_server = socket(AF_INET, SOCK_DGRAM)  # SOCK_DGRAM:數據報式套接字
udp_server.bind(ip_port)
while True:
    # data = udp_server.recvfrom(buffer_size)
    # print(data)  # (b'hello', ('127.0.0.1', 65047))
    data, addr = udp_server.recvfrom(buffer_size)
    print(data)
    print(addr)
    udp_server.sendto(data.upper(), addr)

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from socket import *
ip_port = ('127.0.0.1', 8080)
buffer_size = 1024
udp_client = socket(AF_INET, SOCK_DGRAM)  # SOCK_DGRAM:數據報式套接字
while True:
    msg = input('>>').strip()
    udp_client.sendto(msg.encode('utf-8'), ip_port)
    data, addr = udp_client.recvfrom(buffer_size)
    print(data.decode('utf-8'))
    print(addr)

 八 什么是粘包？

注：只有TCP有粘包現象，UDP永遠不會粘包

一個socket收發消息的原理圖：

　　發送端可以是一K一K地發送數據，而接收端的應用程序可以兩K兩K地提走數據，當然也有可能一次提走3K或6K數據，或者一次只提走幾個字節的數據，也就是說，應用程序所看到的數據是一個整體，或說是一個流（stream），一條消息有多少字節對應用程序是不可見的，因此TCP協議是面向流的協議，這也是容易出現粘包問題的原因。而UDP是面向消息的協議，每個UDP段都是一條消息，應用程序必須以消息為單位提取數據，不能一次提取任意字節的數據，這一點和TCP是很不同的。怎樣定義消息呢？可以認為對方一次性write/send的數據為一個消息，需要明白的是當對方send一條信息的時候，無論底層怎樣分段分片，TCP協議層會把構成整條消息的數據段排序完成后才呈現在內核緩沖區。

　　例如基於tcp的套接字客戶端往服務端上傳文件，發送時文件內容是按照一段一段的字節流發送的，在接收方看了，根本不知道該文件的字節流從何處開始，在何處結束

　　所謂粘包問題主要還是因為接收方不知道消息之間的界限，不知道一次性提取多少字節的數據所造成的。

　　此外，發送方引起的粘包是由TCP協議本身造成的，TCP為提高傳輸效率，發送方往往要收集到足夠多的數據后才發送一個TCP段。若連續幾次需要send的數據都很少，通常TCP會根據優化算法把這些數據合成一個TCP段后一次發送出去，這樣接收方就收到了粘包數據。

1. TCP（transport control protocol，傳輸控制協議）是面向連接的，面向流的，提供高可靠性服務。收發兩端（客戶端和服務器端）都要有一一成對的socket，因此，發送端為了將多個發往接收端的包，更有效的發到對方，使用了優化方法（Nagle算法），將多次間隔較小且數據量小的數據，合並成一個大的數據塊，然后進行封包。這樣，接收端，就難於分辨出來了，必須提供科學的拆包機制。 即面向流的通信是無消息保護邊界的。
2. UDP（user datagram protocol，用戶數據報協議）是無連接的，面向消息的，提供高效率服務。不會使用塊的合並優化算法，, 由於UDP支持的是一對多的模式，所以接收端的skbuff(套接字緩沖區）采用了鏈式結構來記錄每一個到達的UDP包，在每個UDP包中就有了消息頭（消息來源地址，端口等信息），這樣，對於接收端來說，就容易進行區分處理了。 即面向消息的通信是有消息保護邊界的。
3. tcp是基於數據流的，於是收發的消息不能為空，這就需要在客戶端和服務端都添加空消息的處理機制，防止程序卡住，而udp是基於數據報的，即便是你輸入的是空內容（直接回車），那也不是空消息，udp協議會幫你封裝上消息頭

　　udp的recvfrom是阻塞的，一個recvfrom(x)必須對唯一一個sendinto(y),收完了x個字節的數據就算完成,若是y>x數據就丟失，這意味着udp根本不會粘包，但是會丟數據，不可靠

　　tcp的協議數據不會丟，沒有收完包，下次接收，會繼續上次繼續接收，己端總是在收到ack時才會清除緩沖區內容。數據是可靠的，但是會粘包。

粘包解決方案：

方法一、

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from socket import *
import subprocess

ip_port = ('127.0.0.1', 8080)
back_log = 5
buffer_size = 1024
tcp_server = socket(AF_INET, SOCK_STREAM)
tcp_server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
tcp_server.bind(ip_port)
tcp_server.listen(back_log)
while True:
    conn, addr = tcp_server.accept()
    while True:
        try:
            cmd = conn.recv(buffer_size)
            # if not cmd:break
            # 執行命令，得到命令的運行結果cmd_res
            res = subprocess.Popen(cmd.decode('utf-8'), shell=True,
                                   stderr=subprocess.PIPE,
                                   stdout=subprocess.PIPE,
                                   stdin=subprocess.PIPE
                                   )
            err = res.stderr.read()
            if err:
                cmd_res = err
            else:
                cmd_res = res.stdout.read()
            if not cmd_res:
                cmd_res = 'excute success'.encode('utf-8')
            length = len(cmd_res)
            conn.send(str(length).encode('utf-8'))
            client_ready = conn.recv(buffer_size)
            if client_ready == b'ready':
                conn.send(cmd_res)

        except Exception as EX:
            print(EX)
            break
    conn.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from socket import *

ip_port = ('127.0.0.1', 8080)
back_log = 5
buffer_size = 1024
tcp_client = socket(AF_INET, SOCK_STREAM)
tcp_client.connect(ip_port)
while True:
    cmd = input('>>>').strip()
    if not cmd:continue
    if cmd == 'quit':break
    tcp_client.send(cmd.encode('utf-8'))
    length = tcp_client.recv(buffer_size)
    tcp_client.send(b'ready')
    length = int(length.decode('utf-8'))
    recv_size = 0
    recv_msg = b''
    while recv_size < length:
        # 第一種寫法
        # r_m = tcp_client.recv(buffer_size)
        # recv_msg += r_m
        # recv_size += len(r_m)

        # 第二種寫法
        recv_msg += tcp_client.recv(buffer_size)
        recv_size = len(recv_msg)
    print(recv_msg.decode('gbk'))
tcp_client.close()

方法二、

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from socket import *
import subprocess
import struct

ip_port = ('127.0.0.1', 8080)
back_log = 5
buffer_size = 1024
tcp_server = socket(AF_INET, SOCK_STREAM)
tcp_server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
tcp_server.bind(ip_port)
tcp_server.listen(back_log)
while True:
    conn, addr = tcp_server.accept()
    while True:
        try:
            cmd = conn.recv(buffer_size)
            # if not cmd:break
            # 執行命令，得到命令的運行結果cmd_res
            res = subprocess.Popen(cmd.decode('utf-8'), shell=True,
                                   stderr=subprocess.PIPE,
                                   stdout=subprocess.PIPE,
                                   stdin=subprocess.PIPE
                                   )
            err = res.stderr.read()
            if err:
                cmd_res = err
            else:
                cmd_res = res.stdout.read()
            if not cmd_res:
                cmd_res = 'excute success'.encode('utf-8')
            length = len(cmd_res)
            data_length = struct.pack('i', length)
            conn.send(data_length)
            conn.send(cmd_res)

        except Exception as EX:
            print(EX)
            break
    conn.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from socket import *
import struct

ip_port = ('127.0.0.1', 8080)
back_log = 5
buffer_size = 1024
tcp_client = socket(AF_INET, SOCK_STREAM)
tcp_client.connect(ip_port)
while True:
    cmd = input('>>>').strip()
    if not cmd:continue
    if cmd == 'quit':break
    tcp_client.send(cmd.encode('utf-8'))
    length_data = tcp_client.recv(4)
    length = struct.unpack('i', length_data)[0]
    recv_size = 0
    recv_msg = b''
    while recv_size < length:
        # 第一種寫法
        # r_m = tcp_client.recv(buffer_size)
        # recv_msg += r_m
        # recv_size += len(r_m)

        # 第二種寫法
        recv_msg += tcp_client.recv(buffer_size)
        recv_size = len(recv_msg)
    print(recv_msg.decode('gbk'))
tcp_client.close()

 九 利用socketserver實現並發

基於TCP服務端:

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import socketserver


class MyServer(socketserver.BaseRequestHandler):
    def handle(self):
        print('conn is:', self.request)  # <==>conn
        print('addr is:', self.client_address)  # <==> addr
        while True:
            try:
                # 收消息
                data = self.request.recv(1024)
                if not data:break
                print('收到客戶端的消息是', data)
                # 發消息
                self.request.sendall(data.upper())
            except Exception as EX:
                print('錯誤提示:',EX)
                break


if __name__ == '__main__':
    s = socketserver.ThreadingTCPServer(('127.0.0.1', 8080), MyServer)  # 多線程;第一個參數，地址+端口；第二個參數，類
25　　　# s = socketserver.ForkingTCPServer(('127.0.0.1', 8080), MyServer)  # 多進程；多進程的開銷大於多線程
    s.serve_forever()

基於TCP客戶端:

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from socket import *
ip_port = ('127.0.0.1', 8080)
back_log = 5
buffer_size = 1024
tcp_client = socket(AF_INET, SOCK_STREAM)
tcp_client.connect(ip_port)
while True:
    msg = input('>>').strip()
    if not msg:continue
    if msg == 'quit':break
    tcp_client.send(msg.encode('utf-8'))
    data = tcp_client.recv(buffer_size)
    print('收到服務端發來的消息:', data.decode('utf-8'))
tcp_client.close()

基於UDP服務端 

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import socketserver


class MyServer(socketserver.BaseRequestHandler):
    def handle(self):
        print(self.request)
        print('收到客戶端的消息是:', self.request[0].upper())
        self.request[1].sendto(self.request[0].upper(), self.client_address)


if __name__ == '__main__':
    s = socketserver.ThreadingUDPServer(('127.0.0.1', 8080), MyServer)  # 第一個參數，地址+端口；第二個參數，類
    s.serve_forever()

基於UDP客戶端

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from socket import *
ip_port = ('127.0.0.1', 8080)
buffer_size = 1024
udp_client = socket(AF_INET, SOCK_DGRAM)  # SOCK_DGRAM:數據報式套接字
while True:
    msg = input('>>').strip()
    udp_client.sendto(msg.encode('utf-8'), ip_port)
    data, addr = udp_client.recvfrom(buffer_size)
    print(data.decode('utf-8'))
    print(addr)

 十 認證客戶端的合法性

　　如果想在分布式系統中實現一個簡單的客戶端鏈接認證功能，又不像SSL那么復雜，那么可以利用hmac+加鹽的方法來實現。

# _*_coding:utf-8_*_
__author__ = 'Linhaifeng'
from socket import *
import hmac, os

secret_key = b'alex bang bang bang'


def conn_auth(conn):
    '''
    認證客戶端鏈接
    :param conn:
    :return:
    '''
    print('開始驗證新鏈接的合法性')
    msg = os.urandom(32)  # 隨機生成的;b'\xa3\x9d\xaa\x94\x9e\x89\xe9\xc9\xc3r\xf9E\xe0w\x82=\xaac-\x04\xd8:\xea\x07\xad\x1dx\x1er\xe0\x7f\x02'
    conn.sendall(msg)
    h = hmac.new(secret_key, msg)  # <hmac.HMAC object at 0x000000D5DA4824E0>
    digest = h.digest()  # 隨機生成的;b'\x17!*\xae6\x81\xfe|)\x138\xfa2o%\x1a'
    respone = conn.recv(len(digest))
    return hmac.compare_digest(respone, digest)  # 比較第一個參數和第二個參數；相同，返回True，反之也成立


def data_handler(conn, bufsize=1024):
    if not conn_auth(conn):
        print('該鏈接不合法,關閉')
        conn.close()
        return
    print('鏈接合法,開始通信')
    while True:
        data = conn.recv(bufsize)
        if not data: break
        conn.sendall(data.upper())


def server_handler(ip_port, bufsize, backlog=5):
    '''
    只處理鏈接
    :param ip_port:
    :return:
    '''
    tcp_socket_server = socket(AF_INET, SOCK_STREAM)
    tcp_socket_server.bind(ip_port)
    tcp_socket_server.listen(backlog)
    while True:
        conn, addr = tcp_socket_server.accept()
        print('新連接[%s:%s]' % (addr[0], addr[1]))
        data_handler(conn, bufsize)


if __name__ == '__main__':
    ip_port = ('127.0.0.1', 9999)
    bufsize = 1024
    server_handler(ip_port, bufsize)

# _*_coding:utf-8_*_
__author__ = 'Linhaifeng'
from socket import *
import hmac, os

secret_key = b'alex bang bang bang'


def conn_auth(conn):
    '''
    認證客戶端鏈接
    :param conn:
    :return:
    '''
    print('開始驗證新鏈接的合法性')
    msg = os.urandom(32)
    conn.sendall(msg)
    h = hmac.new(secret_key, msg)
    digest = h.digest()
    respone = conn.recv(len(digest))
    return hmac.compare_digest(respone, digest)


def data_handler(conn, bufsize=1024):
    if not conn_auth(conn):
        print('該鏈接不合法,關閉')
        conn.close()
        return
    print('鏈接合法,開始通信')
    while True:
        data = conn.recv(bufsize)
        if not data: break
        conn.sendall(data.upper())


def server_handler(ip_port, bufsize, backlog=5):
    '''
    只處理鏈接
    :param ip_port:
    :return:
    '''
    tcp_socket_server = socket(AF_INET, SOCK_STREAM)
    tcp_socket_server.bind(ip_port)
    tcp_socket_server.listen(backlog)
    while True:
        conn, addr = tcp_socket_server.accept()
        print('新連接[%s:%s]' % (addr[0], addr[1]))
        data_handler(conn, bufsize)


if __name__ == '__main__':
    ip_port = ('127.0.0.1', 9999)
    bufsize = 1024
    server_handler(ip_port, bufsize)

 十一 FTP服務器

實現如下功能：

　　1、用戶加密認證

　　2、每個用戶都有自己的家目錄，且只能訪問自己的家目錄

　　3、允許用戶在ftp server上隨意切換目錄(cd)

　　4、允許用戶查看當前目錄下的所有文件(ls)

　　5、允許上傳和下載文件

　　6、文件傳輸過程中顯示進度條

　　7、支持文件的斷點續傳

ftp server

目錄結構：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import os,sys
PATH = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(PATH)


from core import main


if __name__ == '__main__':
    main.ArgvHandler()

[DEFAULT]

[alex]
Password =123
Quotation = 100

[root]
Password = root
Quotation = 100

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import os
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

ip = '127.0.0.1'
port = 8080

ACCOUNT_PATH = os.path.join(BASE_DIR, 'conf', 'accounts.cfg')

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import socketserver
import optparse
import socketserver
from conf import settings
from core import server


class ArgvHandler():
    def __init__(self):
        self.op = optparse.OptionParser()
        # self.op.add_option('-s', '--server', dest='server')
        # self.op.add_option('-P', '--port', dest='port')
        options, args = self.op.parse_args()
        # print(options)  # {'server': '127.0.0.1', 'port': '8080'}
        # print(type(options))  # <class 'optparse.Values'>
        # print(options.server)  # 127.0.0.1
        # print(args)

        options, args = self.op.parse_args()
        self.verify_args(options, args)

    def verify_args(self, options, args):
        cmd = args[0]
        # 第一種方法
        # if cmd == 'start':
        #     pass
        # else:
        #     pass
        # 第二種方法
        if hasattr(self, cmd):
            func = getattr(self, cmd)
            func()

    def start(self):
        print('ths server is working...')
        s = socketserver.ThreadingTCPServer((settings.ip, settings.port), server.ServerHandler)
        s.serve_forever()

    def help(self):
        pass

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import socketserver
import json
import configparser
from conf import settings
import os

STATUS_CODE = {
    250: "Invalid cmd format,e.g:{'action':'get','filename':'test.py','size':344}",
    251: "Invalid cmd",
    252: "Invalid auth data",
    253: "Wrong username or password",
    254: "Passed authentication",
    255: "Filename doesn't provided",
    256: "File doesn't exist on server",
    257: "ready to send file",
    258: "md5 verification",
    800: "the file exist,but not enough,is continue?",
    801: "the file exist!",
    802: "ready to receive datas",
    900: "md5 valdate success"
}


class ServerHandler(socketserver.BaseRequestHandler):
    def handle(self):
        while True:
            data = self.request.recv(1024).strip()  # self.request=conn
            data = json.loads(data.decode('utf8'))
            if data.get('action'):
                if hasattr(self, data.get('action')):
                    func = getattr(self, data.get('action'))
                    func(**data)
                else:
                    print('Invalid cmd')
            else:
                print('Invalid cmd')

    def send_response(self, status_code):
        response = {'status_code': status_code}
        self.request.sendall(json.dumps(response).encode('utf8'))

    def auth(self, **data):
        username = data['username']
        password = data['password']
        username = self.authenticate(username, password)
        if username:
            self.send_response(254)
        else:
            self.send_response(253)

    def authenticate(self, username, password):
        cfg = configparser.ConfigParser()
        cfg.read(settings.ACCOUNT_PATH)
        if username in cfg.sections():
            if cfg[username]['Password'] == password:
                self.username = username
                self.mainPath = os.path.join(settings.BASE_DIR, 'home', self.username)
                print('passed authenticate')
                return username

    def put(self, **data):
        print('data', data)
        file_name = data.get('file_name')
        file_size = data.get('file_size')
        target_path = data.get('target_path')
        abs_path = os.path.join(self.mainPath, target_path, file_name)
        has_received = 0
        if os.path.exists(abs_path):
            file_has_size = os.stat(abs_path).st_size
            if file_has_size < file_size:
                # 斷點續傳
                self.request.sendall('800'.encode('utf8'))
                choice = self.request.recv(1024).decode('utf8')
                if choice == 'Y':
                    self.request.sendall(str(file_has_size).encode('utf8'))
                    has_received = file_has_size
                    f = open(abs_path, 'ab')
                else:
                    f = open(abs_path, 'wb')
            else:
                # 文件完全存在
                self.request.sendall('801'.encode('utf8'))
                return
        else:
            self.request.sendall('802'.encode('utf8'))
            f = open(abs_path, 'wb')

        while has_received < file_size:
            try:
                data = self.request.recv(1024)
            except Exception as EX:
                break
            f.write(data)
            has_received += len(data)
        f.close()

    def ls(self, **data):
        file_list = os.listdir(self.mainPath)
        file_str = '\n'.join(file_list)
        if not len(file_list):
            file_str = '<empty dir>'
        self.request.sendall(file_str.encode('utf8'))

    def cd(self, **data):
        dirname = data.get('dirname')
        if dirname == '..':
            self.mainPath = os.path.dirname(self.mainPath)
        else:
            self.mainPath = os.path.join(self.mainPath, dirname)
        self.request.sendall(self.mainPath.encode('utf8'))

    def mkdir(self, **data):
        dirname = data.get('dirname')
        path = os.path.join(self.mainPath, dirname)
        if not os.path.exists(path):
            if '/' in path:
                os.makedirs(path)
            else:
                os.mkdir(path)
            self.request.sendall('create success'.encode('utf8'))
        else:
            self.request.sendall('dirname exist'.encode('utf8'))

ftp client

目錄結構

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# import socket
#
# sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# sk.connect(('127.0.0.1', 8080))
import optparse
import socket
import json
import os, sys
STATUS_CODE = {
    250: "Invalid cmd format,e.g:{'action':'get','filename':'test.py','size':344}",
    251: "Invalid cmd",
    252: "Invalid auth data",
    253: "Wrong username or password",
    254: "Passed authentication",
    255: "Filename doesn't provided",
    256: "File doesn't exist on server",
    257: "ready to send file",
    258: "md5 verification",
    800: "the file exist,but not enough,is continue?",
    801: "the file exist!",
    802: "ready to receive datas",
    900: "md5 valdate success"
}


class ClientHandler(object):
    def __init__(self):
        self.op = optparse.OptionParser()
        self.op.add_option('-s', '--server', dest='server')
        self.op.add_option('-P', '--port', dest='port')
        self.op.add_option('-u', '--username', dest='username')
        self.op.add_option('-p', '--password', dest='password')

        self.options, self.args = self.op.parse_args()

        self.verify_args(self.options, self.args)
        self.make_connection()
        self.mainPath = os.path.dirname(os.path.abspath(__file__))
        self.last = 0

    def verify_args(self, options, args):
        server = options.server
        port = options.port
        username = options.username
        password = options.password
        if int(port) > 0 and int(port) < 65535:
            return True
        else:
            exit('the port is in 0-65535')

    def make_connection(self):
        self.sock = socket.socket()
        self.sock.connect((self.options.server, int(self.options.port)))

    def interactive(self):
        print('begin to interactive...')
        # self.authenticate()
        if self.authenticate():
            while 1:
                cmd_info = input('[%s]' % self.current_dir).strip()
                cmd_list = cmd_info.split()
                if hasattr(self, cmd_list[0]):
                    func = getattr(self, cmd_list[0])
                    func(*cmd_list)

    def put(self, *cmd_list):
        # put 12.png images
        actions, local_path, target_path = cmd_list
        local_path = os.path.join(self.mainPath, local_path)
        file_name = os.path.basename(local_path)
        file_size = os.stat(local_path).st_size
        data = {
            'action': 'put',
            'file_name': file_name,
            'file_size': file_size,
            'target_path': target_path
        }
        self.sock.send(json.dumps(data).encode('utf8'))
        is_exist = self.sock.recv(1024).decode('utf8')
        has_send = 0
        if is_exist == '800':
            # 文件不完整
            choice = input('the file exist,but not enough,is continue?[Y/N]').strip()
            if choice.upper() == 'Y':
                self.sock.sendall('Y'.encode('utf8'))
                continue_position = self.sock.recv(1024).decode('utf8')
                has_send += int(continue_position)
            else:
                self.sock.sendall('N'.encode('utf8'))
        elif is_exist == '801':
            # 文件完全存在
            print('the file exist')
            return
        else:
            pass
        f = open(local_path, 'rb')
        f.seek(has_send)
        while has_send < file_size:
            data = f.read(1024)
            self.sock.sendall(data)
            has_send += len(data)
            self.show_progress(has_send, file_size)
        f.close()
        print('successfully upload!')

    def show_progress(self, has, total):
        rate = float(has)/float(total)
        rate_num = int(rate*100)
        if self.last != rate_num:
            sys.stdout.write('%s%% %s\r' % (rate_num, '#'*rate_num))
        self.last = rate_num

    def ls(self, *cmd_list):
        data = {
            'action': 'ls'
        }
        self.sock.sendall(json.dumps(data).encode('utf8'))
        data = self.sock.recv(1024).decode('utf8')
        print(data)

    def cd(self, *cmd_list):
        data = {
            'action': 'cd',
            'dirname': cmd_list[1]
        }
        self.sock.sendall(json.dumps(data).encode('utf8'))
        data = self.sock.recv(1024).decode('utf8')
        self.current_dir = os.path.basename(data)

    def mkdir(self, *cmd_list):
        data = {
            'action': 'mkdir',
            'dirname': cmd_list[1]
        }
        self.sock.sendall(json.dumps(data).encode('utf8'))
        data = self.sock.recv(1024).decode('utf8')

    def authenticate(self):
        if self.options.username is None or self.options.password is None:
            username = input('username:')
            password = input('password:')
            return self.get_auth_result(username, password)
        return self.get_auth_result(self.options.username, self.options.password)

    def response(self):
        data = self.sock.recv(1024).decode('utf8')
        data = json.loads(data)
        return data

    def get_auth_result(self, username, password):
        data = {
            'action': "auth",
            'username': username,
            'password': password
        }
        self.sock.send(json.dumps(data).encode('utf8'))
        response = self.response()
        print('response:', response['status_code'])
        if response['status_code'] == 254:
            self.username = username
            self.current_dir = username
            print(STATUS_CODE[response['status_code']])
            return True
        else:
            print(STATUS_CODE[response['status_code']])


ch = ClientHandler()
ch.interactive()

 十二 進程與線程

1、為什么要有操作系統？

　　現代計算機系統是由一個或者多個處理器、內存、硬盤、打印機、鍵盤、鼠標和顯示器等組成的。網絡接口以及各種其他輸入/輸出設備組成的復雜系統，每位程序員不可能掌握所有系統實現的細節，並且管理優化這些部件是一件具有挑戰性極強的工作。所以，我們需要為計算機安裝一層軟件，成為操作系統，任務就是用戶程序性提供一個簡單清晰的計算機模型，並管理以上設備。

　　定義：操作系統是一個用來協調、管理和控制計算機硬件和軟件資源的系統程序，它位於硬件和應用程序之間。程序是運行在系統上的具有某種功能的軟件，比如：瀏覽器，音樂播放器等。

　　操作系統內部的定義：操作系統的內核是一個管理和控制程序，負責管理計算機的所有物理資源，其中包括：文件系統、內存管理、設備管理、進程管理。

2、什么是進程？

　　假如有兩個程序A和B，程序A在執行到一半的過程中，需要讀取大量的數據輸入(I/O操作)，而此時CPU只能靜靜地等待任務A讀取完數據才能繼續執行，這樣就白白浪費了CPU資源。是不是在程序A讀取數據的過程中，讓程序B去執行，當程序A讀取完數據之后，讓程序B暫停，然后讓程序A繼續執行？當然沒問題，但這里有一個關鍵詞：切換；既然是切換，那么這就涉及到了狀態的保存，狀態的恢復，加上程序A與程序B所需要的系統資源(內存，硬盤，鍵盤等等)是不一樣的。自然而然的就需要有一個東西去記錄程序A和程序B分別需要什么資源，怎樣去識別程序A和程序B等等，所以就有了一個叫進程的抽象。

定義：

　　進程就是一個程序在一個數據集上的一次動態執行過程。進程一般由程序、數據庫、進程控制塊三部分組成。我們編寫的程序用來描述進程要完成哪些功能以及如何完成。數據集則是程序在執行過程中所需要使用的資源。進程控制塊用來記錄的外部特征，描述進程的執行變化過程，系統可以利用它來控制和管理進程，它是系統感知進程存在的唯一標志。

　　本質上就是一段程序的運行過程（抽象的概念）

3、什么是線程？

　　線程的出現是為了降低上下文切換的消耗，提高系統的並發性，並突破一個進程只能干一樣事的缺陷，讓進程內並發成為可能。

4、進程與線程區別

1、一個程序至少有一個進程，一個進程至少有一個線程（進程可以理解成線程的容器）

2、進程在執行過程中擁有獨立的內存單元，而多個線程共享內存，從而極大地提高了程序的運行效率

3、線程在執行過程中與進程還是有區別的，每個獨立的線程有一個程序運行的入口，順序執行序列和程序的出口。但是線程不能夠獨立執行，必須依存在應用程序中，由應用程序提供多個線程執行控制。

4、進程是具有一定獨立功能的程序關於某個數據集合上的一次運行活動，進程是系統進行資源分配和調度的一個獨立單位；線程是進程的一個實體，是CPU調度和分源的基本單位，它是比進程更小的能獨立運行的基本單位，線程自己基本上不擁有系統資源，只擁有一點運行中必不可少的資源（如程序計數器，一組寄存器和錢），但是它可與同屬一個進程的其他的線程共享進程所擁有的全部資源。一個進程可以創建和撤銷另一個線程；同一個進程中的多個線程之間可以並發執行。

5、線程：最小的執行單元（實例）；進程：最小的資源單位

5、Python的GIL(全局解釋鎖;Global Interpreter Lock)

　　In CPython, the global interpreter lock, or GIL, is a mutex that prevents multiple native threads from executing Python bytecodes at once. This lock is necessary mainly because CPython’s memory management is not thread-safe. (However, since the GIL exists, other features have grown to depend on the guarantees that it enforces.)

上面的核心意思：無論你啟多少個線程，你有多少個CPU，Python在執行的時候會淡定的在同一時刻只允許一個線程運行。

6、 線程的兩種調用方式

　　threading 模塊建立在thread 模塊之上。thread模塊以低級、原始的方式來處理和控制線程，而threading 模塊通過對thread進行二次封裝，提供了更方便的api來處理線程。

調用方式：

方式一、

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import threading
import time


def music():
    print('begin to listen % s' % time.ctime())
    time.sleep(3)
    print('stop to listen %s' % time.ctime())


def game():
    print('begin to play game % s \r\n' % time.ctime())
    time.sleep(5)
    print('stop to play game %s \r\n' % time.ctime())


threads = []
t1 = threading.Thread(target=music)
t2 = threading.Thread(target=game)
threads.append(t1)
threads.append(t2)
if __name__ == '__main__':
    # join()功能：在子線程完成運行之前，這個子線程的父線程講一直被阻塞
    # t1 = threading.Thread(target=music)
    # t2 = threading.Thread(target=game)
    # t1.start()
    # t2.start()
    #
    # t1.join()
    # t2.join()
    # print('end')
    
    # setDaemon():將線程生命為守護線程
    t2.setDaemon(True)  # 注：一定要在start之前設置
    for t in threads:
        t.start()
    print('end')

方式二、

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import threading
import time


class MyThread(threading.Thread):
    def __init__(self, num):
        threading.Thread.__init__(self)
        self.num = num

    def run(self):  # 定義每個線程要運行的函數

        print("running on number:%s" % self.num)

        time.sleep(3)


if __name__ == '__main__':
    t1 = MyThread(1)
    t2 = MyThread(2)
    t1.start()
    t2.start()

    print("ending......")

join():在子線程完成運行之前，這個子線程的父線程將一直被阻塞。

setDaemon(True)：

　　將線程生命為守護線程，必須在start()方法調用之前設置，如果不設置為守護線程，程序會被無限掛起。這個方法基本和join是相反的。當我們在程序運行中，執行一個主線程，如果主線程又創建一個子線程，主線程和子線程就分兵兩路，分別運行，那么當主線程完成想退出時，會驗證子線程是否完成。如果子線程未完成，則主線程會等待子線程完成后再退出。但是有時候我們需要的是，只要主線程完成了，不管子線程是否完成，都要和主線程一起退出，這時就可以用setDaemon方法了。

 其他方法：

# run():用於表示線程活動的方法
# start():啟動線程活動
# isAlive():返回線程是否活動的,返回布爾值,True/False
# getName():返回線程名字
# setName():設置線程名字

threading模塊提供的一些方法:
# threading.currentThread():返回當前的線程變量
# threading.enumerate():返回一個包含正在運行的線程的list。正在運行指線程啟動后-結束前，不包括啟動前和終止后的線程
# threading.activeCount():返回正在運行的線程數量，與len(threading.enumerate())有相同的結果

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import threading
import time


def music():
    print('begin to listen % s' % time.ctime())
    time.sleep(3)
    print('stop to listen %s' % time.ctime())


def game():
    print('begin to play game % s \r\n' % time.ctime())
    time.sleep(5)
    print('stop to play game %s \r\n' % time.ctime())


threads = []
t1 = threading.Thread(target=music)
t2 = threading.Thread(target=game)
threads.append(t1)
threads.append(t2)
if __name__ == '__main__':
    # join()功能：在子線程完成運行之前，這個子線程的父線程講一直被阻塞
    # t1 = threading.Thread(target=music)
    # t2 = threading.Thread(target=game)
    # t1.start()
    # t2.start()
    #
    # t1.join()
    # t2.join()
    # print('end')

    # setDaemon():將線程生命為守護線程
    t2.setDaemon(True)  # 注：一定要在start之前設置
    for t in threads:
        t.start()
        print(t.getName())
        print('count:', threading.activeCount())
    while threading.activeCount() == 3:
        print('end')

7、同步鎖(lock)

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import threading
import time
num = 100


def sub():
    global num
    print('ok')
    lock.acquire()  # 加鎖
    temp = num
    time.sleep(0.001)
    num = temp-1
    lock.release()  # 釋放鎖


li = []
lock = threading.Lock() 20 for i in range(100):
    t1 = threading.Thread(target=sub)
    t1.start()
    li.append(t1)
for l in li:
    l.join()
print(num)

 注：多個線程都在同時操作同一個共享資源，所以造成了資源破壞(join會造成串行，失去線程的意義)，可以通過同步鎖來解決這種問題。

8、遞歸鎖

　　在線程間共享多個資源的時候，如果兩個線程分別占有一部分資源並且同時等待對方的資源，就會造成死鎖，因為系統判斷這部分資源都在使用，所有這兩個線程在無外力作用下將一直等待下去。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import threading
import time


class MyThread(threading.Thread):

    def actionA(self):
        r_lock.acquire()
        print(self.name, 'gotA', time.ctime())  # 重寫線程后的self.name --->線程的名字
        time.sleep(2)
        r_lock.acquire()
        print(self.name, 'gotB', time.ctime())
        time.sleep(1)
        r_lock.release()
        r_lock.release()

    def actionB(self):
        r_lock.acquire()
        print(self.name, 'gotB', time.ctime())  # 重寫線程后的self.name --->線程的名字
        time.sleep(2)
        r_lock.acquire()
        print(self.name, 'gotA', time.ctime())
        time.sleep(1)
        r_lock.release()
        r_lock.release()

    def run(self):
        self.actionA()
        self.actionB()


if __name__ == '__main__':
    r_lock = threading.RLock() 36     li = []
    for t in range(3):
        t = MyThread()
        t.start()
        li.append(t)

    for i in li:
        i.join()

    print('end')

　　為了支持在同一線程中多次請求同一資源，Python提供了“可重入鎖”：threading.Rlock。Rlock內部維護着一個Lock和counter變量，counter記錄了acquire的次數，從而使得資源可以被多次acquire。直到一個線程所有的acquire都被release，其他的線程才能獲得資源。

9、同步對象(Event)

An event is a simple synchronization object;the event represents an internal flag,

and threads can wait for the flag to be set, or set or clear the flag themselves.

event = threading.Event()

# a client thread can wait for the flag to be set
event.wait()

# a server thread can set or reset it
event.set()
event.clear()

If the flag is set, the wait method doesn’t do anything.
If the flag is cleared, wait will block until it becomes set again.
Any number of threads may wait for the same event.

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import threading
import time


class Boss(threading.Thread):
    def run(self):
        print('Boss:今天加班到22：00！\r\n')
        print(event.isSet())  # False
        event.set()
        time.sleep(6)
        print('Boss:可以下班了，明天放假！\r\n')
        print(event.isSet())
        event.set()


class Worker(threading.Thread):
    def run(self):
        event.wait()  # 一旦event被設定，等同於pass
        print('Worker:唉···命真苦！\r\n')
        time.sleep(1)
        event.clear()
        event.wait()
        print('Worker:OhYeah!\r\n')


if __name__ == '__main__':
    event = threading.Event() 30     threads = []
    for i in range(5):
        threads.append(Worker())
    threads.append(Boss())
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print('end')

10、信號量

　　信號量用來控制線程並發數的，BoundedSemaphore或Semaphore管理一個內置的計數器，每當調用acquire()時-1，調用release()時+1。計數器不能小於0，當計數器為0時，acquire()將阻塞線程至同步鎖狀態，直到其他線程調用release()。(類似於停車位的概念)BoundedSemaphore與Semaphore的唯一區別在於前者將在調用release()時檢查計數器的值是否超過了計數器的初始值，如果超過了將拋出一個異常。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import threading
import time


class MyThread(threading.Thread):
    def run(self):
        if semaphore.acquire():
            print(self.name, '\r')
            time.sleep(5)
            semaphore.release()


if __name__ == '__main__':
    semaphore = threading.Semaphore(5)
    threads = []
    for i in range(100):
        threads.append(MyThread())
    for t in threads:
        t.start()

 

11、隊列(queue)

　　列表是不安全的數據結構：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import threading
import time

li = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]


def Foo():
    while li:
        try:
            last_li = li[-1]
            print(last_li)
            li.remove(last_li)
            time.sleep(1)
        except Exception as EX:
            print('錯誤提示:', last_li, EX)


t1 = threading.Thread(target=Foo)
t1.start()
t2 = threading.Thread(target=Foo)
t2.start()

　　queue隊列類的方法：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import queue  # 線程隊列
q = queue.Queue()  # 創建q對象,同步實現的。隊列長度可為無限或者有限。可通過Queque的構造函數的可選參數maxsize來設定隊列長度。如果maxsize小於1就表示隊列長度無限
q.put(12)  # 將一個值放到隊列，調用隊列對象的put()方法在隊尾插入一個項目。put()有兩個參數，第一個item為必需的，為插入項目的值；第二個block為可選參數，默認為1.如果隊列當前為空且block為1，put()方法就使調用線程暫停，直到空處一個數據單元。如果block為0，put方法將引發Full異常
q.put('alex')
q.put({"age": 15})
print(q.get())  # 將一個值從隊列中取出，調用隊列對象的get()方法，從對頭刪除並返回一個實例。可選參數為block，默認為True。如果隊列為空且block為True，get()就使調用線程暫停，直至有項目可用。如果隊列為空且block為False，隊列將引發Empty異常
print(q.qsize())  # 返回隊列的大小
print(q.empty())  # 判斷隊列是否為空，返回布爾值，True/False
print(q.full())  # 判斷隊列是否已經滿了，返回布爾值，True/False
q.join()  # 實際上意味着等到隊列為空，再執行別的操作

'''
Queue模塊的三種隊列及構造函數
1、Python Queue模塊的FIFO隊列，先進先出 class queue.Queue(maxsize)
2、LIFO 類似於堆，即先進后處。  class queue.LifoQueue(maxsize)
3、優先級隊列，級別越低月先出來。 class queue.PriorityQueue(maxsize)
'''

生產者消費者模型

為什么要使用生產者和消費者模式？

　　在線程世界里，生產者就是生產數據的線程，消費者就是消費數據的線程。在多線程開發中，如果生產者處理速度很快，而消費者處理速度很慢，那么生產者就必須等待消費者處理完，才能繼續生產數據。同樣的道理，如果消費者的處理能力大於生產者，那么消費者就必須等待生產者。為了解決這個問題於是引入了生產者和消費者模式。

什么是生產者消費者模式？

　　生產者和消費者模式是通過一個容器來解決生產者和消費者的強耦合問題。生產者和消費者彼此之間不直接通訊，而通過阻塞隊列來進行通訊，所以生產者生產完數據之后不用等待消費者處理，直接扔給阻塞隊列，消費者不找生產者要數據，而是直接從阻塞隊列里取，阻塞隊列就相當於一個緩沖區，平衡了生產者和消費者的處理能力。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import random
import queue
import threading
q = queue.Queue()


def Producer(name):
    count = 0
    while count < 10:
        print('making..')
        time.sleep(random.randrange(3))
        q.put(count)
        print('Producer [%s] has produced %s meat bun。\r' % (name, count))
        count += 1
        print('ok\r')


def Consumer(name):
    count = 0
    while count < 10:
        time.sleep(random.randrange(4))
        if not q.empty():
            data = q.get()
            print('lm Consumer [%s] has eat %s meat bun。\r' % (name, data))
        else:
            print('---no meat bun anymore----\r')
        count += 1


p1 = threading.Thread(target=Producer, args=('alex',))
c1 = threading.Thread(target=Consumer, args=('B',))
p1.start()
c1.start()

12、並發&並行

　　並發：指系統具有處理多個任務(動作)的能力

　　並行：指系統具有同時處理多個任務(動作)的能力

13、同步&異步

　　同步：當進程執行到一個IO(等待外部數據)的時候你等

　　異步：當進程執行到一個IO(等待外部數據)的時候你不等；一直等到數據接收完成，在回來處理

14、任務類型

　　IO密集型：Python的多線程是有意義的

　　計算密集型：Python的多線程就不推薦，可以采用多進程+協程

 16、多進程模塊( multiprocessing)

Multiprocessing is a package that supports spawning processes using an API similar to the threading module. The multiprocessing package offers both local and remote concurrency,effectively side-stepping the Global Interpreter Lock by using subprocesses instead of threads. Due to this, the multiprocessing module allows the programmer to fully leverage multiple processors on a given machine. It runs on both Unix and Windows.

　　由於GIL的存在，Python中的多線程其實並不是真正的多線程，如果想充分地使用多核CPU的資源，在Python中大部分情況下需要使用多進程。multiprocessing包是Python中的多進程管理包。與threading.Thread類似，它可以利用multiprocessing.Process對象來創建一個進程。該進程可以運行在Python程序內部編寫的函數。該Process對象與Thread對象的用法，也有start()，run()，join()的方法。此外multiprocessing包中也有Lock/Event/Semaphore/Condition類(這些對象可以像多線程那樣，通過參數傳遞給各個進程)，用以同步進程，其用法與threading包中的同名類一致。所以，multiprocessing的很大一部分與threading使用同一套API，只不過換到了多進程的情景。

 調用方式一：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from multiprocessing import Process
import time


def Foo(name):
    time.sleep(1)
    print('hello', name, time.ctime())


if __name__ == '__main__':
    p_list = []
    for i in range(200):
        p = Process(target=Foo, args=('alex',))
        p_list.append(p)
        p.start()
    for i in p_list:
        p.join()
    print('end')

調用方式二：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from multiprocessing import Process
import time


class MyProcess(Process):
    def __init__(self):
        super(MyProcess, self).__init__()

    def run(self):
        time.sleep(1)
        print('hello', self.name, time.ctime())


if __name__ == '__main__':
    p_list = []
    for i in range(3):
        p = MyProcess()
        p.start()
        p_list.append(p)
    for p in p_list:
        p.join()
    print('end')

Process類 

構造方法：

Process([group [, target [, name [, args [, kwargs]]]]])

　　group:線程組，目前還沒有實現，庫引用中提示必須是None

　　target:要執行的方法

　　name:進程名

　　args/kwargs:要傳入方法的參數

實例方法：

　　is_alive():返回進程是否在運行

　　join([timeout]):阻塞當前上下文環境的進程，直到調用此方法的進程終止或到達指定的timeout(可選參數)

　　start():進程准備就緒，等待CPU調度

　　run():start()調用run方法，如果實例進程時未指定傳入target，這start執行t默認run()方法

　　terminate():不管任務是否完成，立即停止工作進程

屬性：

　　daemon:和線程的setDeamon功能一樣

　　name:進程名字

　　pid:進程號

17、進程的通信

進程隊列Queue

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import queue
import multiprocessing


def Foo(q):
    q.put(123)
    q.put(456)


if __name__ == '__main__':
    q = multiprocessing.Queue()  # 注意：此處需用進程隊列，不能用線程隊列，即q=queue.Queue()
    p = multiprocessing.Process(target=Foo, args=(q,))
    p.start()
    print(q.get())
    print(q.get())

 管道

The Pipe() function returns a pair of connection objects connected by a pipe which by default is duplex (two-way). For example:

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from multiprocessing import Process, Pipe


def Foo(child_conn):
    child_conn.send([12, {'name': 'alice'}, 'hello'])
    response = child_conn.recv()
    print('response:', response)
    child_conn.close()
    print('q_id2:', id(child_conn))


if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    print('q_ID1', id(child_conn))
    p = Process(target=Foo, args=(child_conn,))
    p.start()
    print(parent_conn.recv())
    parent_conn.send('早上好!')
    p.join()

Managers

Queue和pipe只是實現了數據交互，並沒實現數據共享，即一個進程去更改另一個進程的數據。

A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.

A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array. For example：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from multiprocessing import Process, Manager


def Foo(dic, li, i):
    dic[i] = '1'
    li.append(i)


if __name__ == '__main__':
    with Manager() as manager:
        dic = manager.dict()
        li = manager.list(range(5))
        p_list = []
        for i in range(10):
            p = Process(target=Foo, args=(dic, li, i))
            p.start()
            p_list.append(p)
        for p in p_list:
            p.join()
        print(dic)
        print(li)

 進程同步

Without using the lock output from the different processes is liable to get all mixed up.

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from multiprocessing import Process, Lock


def Foo(lk, i):
    with lk:  # 默認情況下，已經lk.acquire()
        print('hello world %s' % i)


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

進程池

　　進程池內部維護一個進程序列，當使用時，則去進程池中獲取一個進程，如果進程池序列中沒有可供使用的進程，那么程序就會等待，直到進程池中有可進程為止。

進程池中兩個方法：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from multiprocessing import Process, Pool
import time, os


def Foo(i):
    time.sleep(1)
    print('i = \r', i)


def Bar(arg):  # 此處arg=Foo()函數的返回值
    print('pgid-->%s\r' % os.getpid())
    print('ppid-->%s\r' % os.getppid())
    print('logger:%s\r' % arg)


if __name__ == '__main__':
    pool = Pool(5)
    Bar(1)
    print('------------\r')
    for i in range(10):
        # pool.apply(func=Foo, args=(i,))  # 同步接口
        # pool.apply_async(func=Foo, args=(i,))
        pool.apply_async(func=Foo, args=(i,), callback=Bar)  # callback-->回調函數：就是某個動作或者函數執行成功后再去執行的函數

 pool.close() 27 pool.join() # join和close位置不能反 28 print('end\r')

 十三 協程

　　協程：又稱微線程，英文名：Coroutine，本質上是一個線程

　　優點1：協程具有極高的執行效率。因為子程序切換不是線程切換，而是由程序自身控制。因此，沒有線程切換的開銷，和多線程比，線程數量越多，協程的性能優勢就越明顯。

　　優點2：不需要多線程的鎖機制，因為只有一個線程，也不存在同時寫變量沖突，在協程中控制共享資源不加鎖，只需要判斷狀態就好了，所以執行效率比多線程高很多。

　　因為協程是一個線程執行，那怎么利用多核CPU呢？最簡單的方法就是多進程+協程，即充分利用多核，又充分發揮協程的高效率，可獲得極高的性能。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time


def consumer(name):
    print('---->ready to eat meat bun')
    while True:
        new_meat_bun = yield
        print('[%s] is eating meat bun %s' % (name, new_meat_bun))


def producer():
    con1.__next__()
    con2.__next__()
    n = 0
    while 1:
        time.sleep(1)
        print('\033[32;1m[producer]\033[0m is making meat bun %s and %s' % (n, n + 1))
        con1.send(n)
        con2.send(n + 1)
        n += 2


if __name__ == '__main__':
    con1 = consumer('alex')
    con2 = consumer('alice')
    producer()

 Greenlet

　　greenlet是一個用C實現的協程模塊，相比與Python自帶的yield，它可以使你在任意函數之間隨意切換，而不需把這個函數先聲明為generator。(注：需要用pip安裝包;pip install gevent)

from greenlet import greenlet
import time


def A():
    while 1:
        print('-------A-------')
        time.sleep(0.5)
        g2.switch()


def B():
    while 1:
        print('-------B-------')
        time.sleep(0.5)
        g1.switch()


g1 = greenlet(A)  # 創建協程g1
g2 = greenlet(B)

g1.switch()  # 跳轉至協程g1

 gevent

import gevent
import requests, time  # 此處requests需安裝模塊;pip --timeout=100 install requests

start = time.time()


def f(url):
    print('GET: %s' % url)
    resp = requests.get(url)  # 爬網頁的標簽
    data = resp.text
    print('%d bytes received from %s.' % (len(data), url))


gevent.joinall([
    gevent.spawn(f, 'https://www.qq.com/'),
    gevent.spawn(f, 'https://www.baidu.com/'),
    gevent.spawn(f, 'https://www.taobao.com/'),
])

print("cost time:", time.time() - start)

 十四 緩存I/O

　　緩存I/O又被稱作標准I/O，大多數文件系統的默認I/O操作都是緩存I/O。在Linux的緩存I/O機制中，操作系統會將I/O的數據緩存在文件系統的頁緩存(page cache)中，也就是說，數據會先被拷貝到操作系統內核的緩沖區中，然后才會從操作系統內的緩沖區拷貝到應用程序的地址空間。用戶空間沒法直接訪問內核空間的，內核態到用戶態的數據拷貝。

　　緩存I/O的缺點：數據在傳輸過程中需要在應用程序地址空間和內核進行多次數據拷貝操作，這些數據拷貝操作所帶來的CPU以及內存開銷是非常大的。

I/O發生時設計的對象和步驟：

　　對於一個network IO（以read舉例），他會涉及到兩個系統對象，一個是調用這個IO的process(or thread)，另一個就是系統內核(kernel)。當一個read操作發生時，它會經歷兩個階段：

　　　　1、等待數據准備(Waiting for the data to be read)

　　　　2、將數據從內核拷貝到進程中(Copying the data from the kernel to the process)

注：這兩點很重要，因為這些IO Mode的區別就是在這兩個階段上各有不同的情況。

blocking IO(阻塞IO，Linux下)

　　在Linux中，默認情況下所有的socket都是blocking，一個典型的讀操作大概流程圖：

　　當用戶進程調用了recvfrom這個系統調用，kernel就開始了IO的第一階段:准備數據。對於network IO來說，很多時候數據在一開始還沒到達(如：還沒收到一個完整的UDP包)，這個時候kernel就要等待足夠的數據到來。而在用戶進程這邊，整個進程會被阻塞。當kernel一直等到數據准備好了，它就將數據從kernel中拷貝到用戶內存，然后kernel。所以，blocking IO的特點就是在IO執行的兩個階段都被block了。

non-blocking IO(非阻塞IO,Linux下)

　　在Linux下，可以通過設置socket使其變為non-blocking。當對一個non-blocking socket執行時大概的流程:

從上圖可以看出，當用戶進程發出read時，如果kernel中的數據還沒准備好，那么它並不會block用戶進程，而是立即返回一個error。從用戶進程角度講來講，它發起一個read操作后，並不需要等待，而是馬上就得到了一個結果。用戶進程判斷結果是一個error時，它就知道數據還沒准備好，於是它可以再次發送read操作。一旦kernel中的數據准備好了。所以，用戶進程其實是需要不斷的主動詢問kernel數據好了沒有。

import time
import socket
sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sk.bind(('127.0.0.1',8080))
sk.listen(5)
sk.setblocking(False)  # 設置是否阻塞,默認為True，非阻塞
while True:
    try:
        print('waiting client connection...')
        conn,addr = sk.accept()  # 進程主動輪詢
        print('+++',addr)
        data = sk.recv(1024)
        print(str(data,'utf8'))
        conn.close()
    except Exception as EX:
        print('錯誤提示:',EX)
        time.sleep(2)

import time
import socket
sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
while True:
    sk.connect(('127.0.0.1',8080))
    print('hello')
    sk.sendall(bytes('hello','utf8'))
    time.sleep(1)
    break

 IO multiplexing(IO多路復用)

　　有些地方也稱為這種IO方式為event driven IO。它的基本原理就是select/epoll這個function會不斷的輪詢所負責的所有socket，當某個socket有數據到達了，就通知用戶進程，大概流程圖：

　　當用戶進程調用了select，那么真個進程會被block。而同時，kernel會“監視”所有select負責的socket，當任何一個socket中的數據准備好了，select就會返回。這時用戶進程再調用read操作，將數據從kernel拷貝到用戶進程。(如果處理的連接數不是很多的話，使用select/epoll的web server不一定比使用multi-threading+blocking IO的web server性能更好，可能延遲更大；select/epoll的優勢並不是對於單個連接能處理得更快，而是在於能處理更多的連接)

注：

　　1、select函數返回結果中如果有文件可讀了，那么進程就可以同故宮調用accept()或recv()來讓kernel將位於內核中准備到的數據copy到用戶區。

　　2、select的優勢在於可以處理多個連接，不適用於單個連接。

import socket
import select

sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sk.bind(('127.0.0.1', 8080))
sk.listen(5)
inputs = [sk, ]
while True:
    r, w, e = select.select(inputs, [], [], 3)
    for obj in r:
        print('obj:', obj)
        if obj == sk:
            conn, addr = obj.accept()
            print('已連接：', conn)
            inputs.append(conn)
        else:
            data_byte = obj.recv(1024)
            print(str(data_byte, 'utf8'))
            inp = input('回答[%s]號客戶端>>:' % inputs.index(obj))
            obj.sendall(bytes(inp, 'utf8'))
    print('>>>', r)

import socket

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

while True:
    inp = input(">>>>")
    sk.sendall(bytes(inp, "utf8"))
    data = sk.recv(1024)
    print(str(data, 'utf8'))

 Asynchronous I/O(異步IO)

　　流程圖：

　　從圖中可以看出，用戶進程發起read操作之后，立刻就開始去做其它的事。另一方面，從kernel的角度，當他受到一個asynchronous read之后，首先它會立刻返回，所以不會對用戶進程產生任何block。然后，kernel會等待數據准備完成，然后將數據拷貝到用戶內存，當這一切都完成之后，kernel會給用戶進程發送一個signal，告訴它read操作完成了。

selectors模塊：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import selectors
import socket

sel = selectors.DefaultSelector()  # 根據系統，默認選擇最優IO多用戶模式


def accept(sock, mask):
    conn, addr = sock.accept()
    print('accepted', conn, 'from', addr)
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read)


def read(conn, mask):
    try:
        data = conn.recv(1000)
        if not data:
            raise Exception
        print('收到:', data.decode('utf8'))
        conn.send(data.upper())  # Hope it won't block
    except Exception as EX:
        print('closing:', conn)
        sel.unregister(conn)  # 解除綁定
        conn.close()


sock = socket.socket()
sock.bind(('localhost', 8080))
sock.listen(100)
sock.setblocking(False)
sel.register(sock, selectors.EVENT_READ, accept)  # sock與accept綁定
while True:
    events = sel.select()  # 監聽[sock,conn1,conn2....]
    for key, mask in events:
        callback = key.data
        print('>>callback:', callback)
        callback(key.fileobj, mask)

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import socket
sk = socket.socket()
sk.connect(('127.0.0.1', 8080))
while True:
    inp = input('>>>')
    sk.send(inp.encode('utf8'))
    data = sk.recv(1024)
    print(data.decode('utf8'))