python3 多進程、隊列、進程池、協程

多進程

import multiprocessing
import time


def demo1():
    while True:
        print("demo1")
        time.sleep(1)


def demo2():
    while True:
        print("demo2")
        time.sleep(1)


def main():
    p1 = multiprocessing.Process(target=demo1)
    p2 = multiprocessing.Process(target=demo2)
    p1.start()
    p2.start()


if __name__ == '__main__':
    main()

互斥鎖

import threading
import time

g_num = 0


def demo1(num):
    global g_num
    # 上鎖，如果之前沒有被上鎖，則上鎖成功
    # 如果上鎖之前，已經被鎖上，那么會堵塞在這里，直到這個鎖被解開為止
    mutex.acquire()
    for i in range(num):
        g_num += 1
    # 解鎖
    mutex.release()
    print("%s" % g_num)


def demo2(num):
    global g_num
    mutex.acquire()
    for i in range(num):
        g_num += 1
    # 解鎖
    mutex.release()
    print("%s" % g_num)


# 創建一個互斥鎖，默認沒有上鎖
mutex = threading.Lock()


def main():
    t1 = threading.Thread(target=demo1, args=(1000000,))
    t2 = threading.Thread(target=demo2, args=(1000000,))

    t1.start()
    t2.start()

    time.sleep(3)
    print("%s" % g_num)


if __name__ == '__main__':
    main()

隊列

import multiprocessing


def download_from_web(q):
    # 模擬網上下載的數據
    data = [1, 2, 3, 4, 5, 6]
    for temp in data:
        q.put(temp)


def analysis_data(q):
    # 數據處理
    waiting = list()
    while True:
        data = q.get()
        waiting.append(data)
        if q.empty():
            break

    print(waiting)


def main():
    # 創建隊列
    q = multiprocessing.Queue()
    p1 = multiprocessing.Process(target=download_from_web, args=(q,))
    p2 = multiprocessing.Process(target=analysis_data, args=(q,))

    p1.start()
    p2.start()


if __name__ == '__main__':
    main()

進程池

from multiprocessing import Pool
import time
import random
import os


def worker(msg):
    t_start = time.time()
    print("%s 開始執行,進程號是 %d" % (msg, os.getpid()))
    time.sleep(random.random() * 2)
    t_stop = time.time()
    print(msg, "執行完畢，耗時%.2f秒" % (t_stop - t_start))


if __name__ == '__main__':
    # 定義1個進程池，最大進程數為3
    po = Pool(3)
    for i in range(8):
        # 調用目標，傳遞給目標的參數元組
        po.apply_async(worker, (i,))
    print("-- start --")
    # 關閉進程池，不再接收新的請求
    po.close()
    # 等待po中所有子進程執行完成，必須放在close語句后面
    po.join()
    print("-- end --")

協程

from gevent import monkey
import gevent
import time

monkey.patch_all()


def f1(n):
    for i in range(n):
        print(gevent.getcurrent(), i)
        time.sleep(0.5)


def f2(n):
    for i in range(n):
        print(gevent.getcurrent(), i)
        time.sleep(0.5)


def f3(n):
    for i in range(n):
        print(gevent.getcurrent(), i)
        time.sleep(0.5)


if __name__ == '__main__':
    gevent.joinall([
        gevent.spawn(f1, 3),
        gevent.spawn(f2, 3),
        gevent.spawn(f3, 3)
    ])

 使用協程下載圖片

import gevent
import time
from urllib import request
from gevent import monkey
from numba import jit

monkey.patch_all()


@jit
def cbk(a, b, c):
    """ 回調函數
    @a:已經下載的數據塊
    @b:數據塊的大小
    @c:遠程文件的大小
    """
    per = 100.0 * a * b / c
    if per > 100:
        per = 100
    print('\r下載已經完成 %.2f%%' % per, end="")


imgurl = ["https://ww3.sinaimg.cn/mw600/006XNEY7gy1fwj4w6ttb6j30hs0k4tc9.jpg",
          "https://ww3.sinaimg.cn/mw600/0073tLPGgy1fx6pxmpwmuj30qo0zkn2x.jpg"]

headers = [
    ("User-agent",
     "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/68.0.3440.106 Safari/537.36")]


@jit
def downloader(jpg_name, url):
    opener = request.build_opener()
    opener.addheaders = headers
    request.install_opener(opener)
    request.urlretrieve(url, jpg_name, cbk)
    time.sleep(0.1)


@jit
def main():
    time1 = time.time()
    gevent.joinall([
        gevent.spawn(downloader, "1.jpg", imgurl[0]),
        gevent.spawn(downloader, "2.jpg", imgurl[1]),
    ])
    time2 = time.time()
    print(time2 - time1)


if __name__ == '__main__':
    main()