進程和線程

進程是程序運行的實例。一個進程里面可以包含多個線程，因此同一進程下的多個線程之間可以共享線程內(nèi)的所有資源，它是操作系統(tǒng)動態(tài)運行的基本單元；每一個線程是進程下的一個實例，可以動態(tài)調(diào)度和獨立運行，由于線程和進程有很多類似的特點，因此，線程又被稱為輕量級的進程。線程的運行在進程之下，進程的存在依賴于線程；

開胃菜

基于 Python3 創(chuàng)建一個簡單的進程示例

          
            from threading import Thread
from time import sleep


class CookBook(Thread):
    def __init__(self):
        Thread.__init__(self)
        self.message = "Hello Parallel Python CookBook!!\n"

    def print_message(self):
        print(self.message)

    def run(self):
        print("Thread Starting\n")
        x = 0
        while x < 10:
            self.print_message()
            sleep(2)
            x += 1
        print("Thread Ended!\n")


print("Process Started")
hello_python = CookBook()

hello_python.start()
print("Process Ended")
          
        

需要注意的是，永遠不要讓線程在后臺默默執(zhí)行，當(dāng)其執(zhí)行完畢后要及時釋放資源。

基于線程的并行

多線程編程一般使用共享內(nèi)存空間進行線程間的通信，這就使管理內(nèi)存空間成為多線程編程的關(guān)鍵。Python 通過標(biāo)準(zhǔn)庫 threading 模塊來管理線程，具有以下的組件：

• 線程對象
• Lock 對象
• RLock 對象
• 信號對象
• 條件對象
• 事件對象

定義一個線程

基本語法

示例代碼如下所示

          
            import threading


def function(i):
    print("function called by thread: {0}".format(i))
    return


threads = []
for i in range(5):
    t = threading.Thread(target=function, args=(i,))
    threads.append(t)
    t.start()

lambda t, threads: t.join()
          
        

需要注意的是，線程創(chuàng)建后并不會自動運行，需要主動調(diào)用 start() 方法來啟動線程， join() 會讓調(diào)用它的線程被阻塞直到執(zhí)行結(jié)束。（PS：可通過調(diào)用 t.setDaemon(True) 使其為后臺線程避免主線程被阻塞）

線程定位

示例代碼如下所示

          
            import threading
import time


def first_function():
    print("{0} is starting".format(threading.currentThread().getName()))
    time.sleep(2)
    print("{0} is Exiting".format(threading.currentThread().getName()))


def second_function():
    print("{0} is starting".format(threading.currentThread().getName()))
    time.sleep(2)
    print("{0} is Exiting".format(threading.currentThread().getName()))


def third_function():
    print("{0} is starting".format(threading.currentThread().getName()))
    time.sleep(2)
    print("{0} is Exiting".format(threading.currentThread().getName()))

if __name__ == "__main__":
    t1 = threading.Thread(target=first_function,name="first")
    t2 = threading.Thread(target=second_function,name="second")
    t3 = threading.Thread(target=third_function,name="third")

    t1.start()
    t2.start()
    t3.start()
    t1.join()
    t2.join()
    t3.join()
          
        

通過設(shè)置 threading.Thread() 函數(shù)的 name 參數(shù)來設(shè)置線程名稱，通過 threading.currentThread().getName() 來獲取當(dāng)前線程名稱；線程的默認(rèn)名稱會以 Thread-{i} 格式來定義

自定義一個線程對象

示例代碼如下所示

          
            import threading
import time

exitFlag = 0


class myThread(threading.Thread):
    def __init__(self, threadID, name, counter):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.counter = counter

    def run(self):
        print("Starting:{0}".format(self.name))
        print_time(self.name, self.counter, 5)
        print("Exiting:{0}".format(self.name))


def print_time(threadName, delay, counter):
    while counter:
        if exitFlag:
            thread.exit()
        time.sleep(delay)
        print("{0} {1}".format(threadName, time.ctime(time.time())))
        counter -= 1


t1 = myThread(1, "Thread-1", 1)
t2 = myThread(2, "Thread-2", 1)

t1.start()
t2.start()

t1.join()
t2.join()

print("Exiting Main Thread.")
          
        

如果想自定義一個線程對象，首先就是要定義一個繼承 threading.Thread 類的子類，實現(xiàn) 構(gòu)造函數(shù) ， 并重寫 run() 方法即可。

線程同步

Lock

示例代碼如下所示

          
            import threading

shared_resource_with_lock = 0
shared_resource_with_no_lock = 0
COUNT = 100000
shared_resource_lock = threading.Lock()


def increment_with_lock():
    global shared_resource_with_lock
    for i in range(COUNT):
        shared_resource_lock.acquire()
        shared_resource_with_lock += 1
        shared_resource_lock.release()


def decrement_with_lock():
    global shared_resource_with_lock
    for i in range(COUNT):
        shared_resource_lock.acquire()
        shared_resource_with_lock -= 1
        shared_resource_lock.release()


def increment_without_lock():
    global shared_resource_with_no_lock
    for i in range(COUNT):
        shared_resource_with_no_lock += 1


def decrement_wthout_lock():
    global shared_resource_with_no_lock
    for i in range(COUNT):
        shared_resource_with_no_lock -= 1


if __name__ == "__main__":
    t1 = threading.Thread(target=increment_with_lock)
    t2 = threading.Thread(target=decrement_with_lock)
    t3 = threading.Thread(target=increment_without_lock)
    t4 = threading.Thread(target=decrement_wthout_lock)
    t1.start()
    t2.start()
    t3.start()
    t4.start()
    t1.join()
    t2.join()
    t3.join()
    t4.join()
    print("the value of shared variable with lock management is :{0}".format(
        shared_resource_with_lock))
    print("the value of shared variable with race condition is :{0}".format(
        shared_resource_with_no_lock))
          
        

通過 threading.Lock() 方法我們可以拿到線程鎖，一般有兩種操作方式： acquire() 和 release() 在兩者之間是加鎖狀態(tài)，如果釋放失敗的話會顯示 RuntimError() 的異常。

RLock

RLock 也叫遞歸鎖，和 Lock 的區(qū)別在于：誰拿到誰釋放,是通過 threading.RLock() 來拿到的；

示例代碼如下所示

          
            import threading
import time


class Box(object):
    lock = threading.RLock()

    def __init__(self):
        self.total_items = 0

    def execute(self, n):
        Box.lock.acquire()
        self.total_items += n
        Box.lock.release()

    def add(self):
        Box.lock.acquire()
        self.execute(1)
        Box.lock.release()

    def remove(self):
        Box.lock.acquire()
        self.execute(-1)
        Box.lock.release()


def adder(box, items):
    while items > 0:
        print("adding 1 item in the box")
        box.add()
        time.sleep(1)
        items -= 1


def remover(box, items):
    while items > 0:
        print("removing 1 item in the box")
        box.remove()
        time.sleep(1)
        items -= 1


if __name__ == "__main__":
    items = 5
    print("putting {0} items in the box".format(items))
    box = Box()
    t1 = threading.Thread(target=adder, args=(box, items))
    t2 = threading.Thread(target=remover, args=(box, items))

    t1.start()
    t2.start()

    t1.join()
    t2.join()
    print("{0} items still remain in the box".format(box.total_items))
          
        

信號量

示例代碼如下所示

          
            import threading
import time
import random

semaphore = threading.Semaphore(0)


def consumer():
    print("Consumer is waiting.")
    semaphore.acquire()
    print("Consumer notify:consumed item numbers {0}".format(item))


def producer():
    global item
    time.sleep(10)
    item = random.randint(0, 10000)
    print("producer notify:produced item number {0}".format(item))
    semaphore.release()


if __name__ == "__main__":
    for i in range(0, 5):
        t1 = threading.Thread(target=producer)
        t2 = threading.Thread(target=consumer)
        t1.start()
        t2.start()
        t1.join()
        t2.join()

    print("program terminated.")
          
        

信號量初始化為 0 ，然后在兩個并行線程中，通過調(diào)用 semaphore.acquire() 函數(shù)會阻塞消費者線程，直到 semaphore.release() 在生產(chǎn)者中被調(diào)用，這里模擬了 生產(chǎn)者-消費者 模式來進行了測試；如果信號量的計數(shù)器到了0，就會阻塞 acquire() 方法，直到得到另一個線程的通知。如果信號量的計數(shù)器大于0，就會對這個值-1然后分配資源。

使用條件進行線程同步

解釋條件機制最好的例子還是生產(chǎn)者-消費者問題。在本例中，只要緩存不滿，生產(chǎn)者一直向緩存生產(chǎn)；只要緩存不空，消費者一直從緩存取出（之后銷毀）。當(dāng)緩沖隊列不為空的時候，生產(chǎn)者將通知消費者；當(dāng)緩沖隊列不滿的時候，消費者將通知生產(chǎn)者。

示例代碼如下所示

          
            from threading import Thread, Condition
import time

items = []
condition = Condition()


class consumer(Thread):
    def __init__(self):
        Thread.__init__(self)

    def consume(self):
        global condition
        global items
        condition.acquire()
        if len(items) == 0:
            condition.wait()
            print("Consumer notify:no item to consum")
        items.pop()
        print("Consumer notify: consumed 1 item")
        print("Consumer notify: item to consume are:{0}".format(len(items)))

        condition.notify()
        condition.release()

    def run(self):
        for i in range(0, 20):
            time.sleep(2)
            self.consume()


class producer(Thread):
    def __init__(self):
        Thread.__init__(self)

    def produce(self):
        global condition
        global items
        condition.acquire()
        if len(items) == 10:
            condition.wait()
            print("Producer notify:items producted are:{0}".format(len(items)))
            print("Producer notify:stop the production!!")
        items.append(1)
        print("Producer notify:total items producted:{0}".format(len(items)))
        condition.notify()
        condition.release()

    def run(self):
        for i in range(0, 20):
            time.sleep(1)
            self.produce()


if __name__ == "__main__":
    producer = producer()
    consumer = consumer()
    producer.start()
    consumer.start()
    producer.join()
    consumer.join()
          
        

通過 condition.acquire() 來獲取鎖對象， condition.wait() 會使當(dāng)前線程進入阻塞狀態(tài)，直到收到 condition.notify() 信號，同時，調(diào)用信號的通知的對象也要及時調(diào)用 condition.release() 來釋放資源；

使用事件進行線程同步

事件是線程之間用于通信的對。有的線程等待信號，有的線程發(fā)出信號。

示例代碼如下所示

          
            import time
from threading import Thread, Event
import random

items = []
event = Event()


class consumer(Thread):
    def __init__(self, items, event):
        Thread.__init__(self)
        self.items = items
        self.event = event

    def run(self):
        while True:
            time.sleep(2)
            self.event.wait()
            item = self.items.pop()
            print('Consumer notify:{0} popped from list by {1}'.format(
                item, self.name))


class producer(Thread):
    def __init__(self, integers, event):
        Thread.__init__(self)
        self.items = items
        self.event = event

    def run(self):
        global item
        for i in range(100):
            time.sleep(2)
            item = random.randint(0, 256)
            self.items.append(item)
            print('Producer notify: item  N° %d appended to list by %s' %
                  (item, self.name))
            print('Producer notify: event set by %s' % self.name)
            self.event.set()
            print('Produce notify: event cleared by %s ' % self.name)
            self.event.clear()


if __name__ == "__main__":
    t1 = producer(items, event)
    t2 = consumer(items, event)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
          
        

使用 with 語法簡化代碼

          
            import threading
import logging

logging.basicConfig(level=logging.DEBUG,
                    format='(%(threadName)-10s) %(message)s')


def threading_with(statement):
    with statement:
        logging.debug("%s acquired via with" % statement)


def Threading_not_with(statement):
    statement.acquire()
    try:
        logging.debug("%s acquired directly " % statement)
    finally:
        statement.release()


if __name__ == "__main__":
    lock = threading.Lock()
    rlock = threading.RLock()
    condition = threading.Condition()
    mutex = threading.Semaphore(1)
    threading_synchronization_list = [lock, rlock, condition, mutex]

    for statement in threading_synchronization_list:
        t1 = threading.Thread(target=threading_with, args=(statement,))
        t2 = threading.Thread(target=Threading_not_with, args=(statement,))
        t1.start()
        t2.start()
        t1.join()
        t2.join()
          
        

使用 queue 進行線程通信

Queue 常用的方法有以下四個：

• put()：往 queue 中添加一個元素
• get()：從 queue 中刪除一個元素，并返回該元素
• task_done()：每次元素被處理的時候都需要調(diào)用這個方法
• join()：所有元素都被處理之前一直阻塞

          
            from threading import Thread, Event
from queue import Queue
import time
import random


class producer(Thread):
    def __init__(self, queue):
        Thread.__init__(self)
        self.queue = queue

    def run(self):
        for i in range(10):
            item = random.randint(0, 256)
            self.queue.put(item)
            print("Producer notify: item item N° %d appended to queue by %s" %
                  (item, self.name))
            time.sleep(1)


class consumer(Thread):
    def __init__(self, queue):
        Thread.__init__(self)
        self.queue = queue

    def run(self):
        while True:
            item = self.queue.get()
            print('Consumer notify : %d popped from queue by %s' %
                  (item, self.name))
            self.queue.task_done()


if __name__ == "__main__":
    queue = Queue()
    t1 = producer(queue)
    t2 = consumer(queue)
    t3 = consumer(queue)
    t4 = consumer(queue)
    t1.start()
    t2.start()
    t3.start()
    t4.start()
    t1.join()
    t2.join()
    t3.join()
    t4.join()
          
        

基于進程的并行

multiprocessing 是 Python 標(biāo)準(zhǔn)庫中的模塊，實現(xiàn)了共享內(nèi)存機制。

異步編程

使用 concurrent.futures 模塊

該模塊具有線程池和進程池，管理并行編程任務(wù)、處理非確定性的執(zhí)行流程、進程/線程同步等功能；此模塊由以下部分組成

• concurrent.futures.Executor: 這是一個虛擬基類，提供了異步執(zhí)行的方法。
• submit(function, argument): 調(diào)度函數(shù)（可調(diào)用的對象）的執(zhí)行，將 argument 作為參數(shù)傳入。
• map(function, argument): 將 argument 作為參數(shù)執(zhí)行函數(shù)，以 異步 的方式。
• shutdown(Wait=True): 發(fā)出讓執(zhí)行者釋放所有資源的信號。
• concurrent.futures.Future: 其中包括函數(shù)的異步執(zhí)行。Future對象是submit任務(wù)（即帶有參數(shù)的functions）到executor的實例。

示例代碼如下所示

          
            import concurrent.futures
import time

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


def evaluate_item(x):
    result_item = count(x)
    return result_item


def count(number):
    for i in range(0, 1000000):
        i = i + 1
    return i * number


if __name__ == "__main__":
    # 順序執(zhí)行
    start_time = time.time()
    for item in number_list:
        print(evaluate_item(item))
    print("Sequential execution in " + str(time.time() - start_time), "seconds")
    # 線程池執(zhí)行
    start_time_1 = time.time()
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = [executor.submit(evaluate_item, item)
                   for item in number_list]
        for future in concurrent.futures.as_completed(futures):
            print(future.result())
    print("Thread pool execution in " +
          str(time.time() - start_time_1), "seconds")
    # 線程池執(zhí)行
    start_time_2 = time.time()
    with concurrent.futures.ProcessPoolExecutor(max_workers=5) as executor:
        futures = [executor.submit(evaluate_item, item)
                   for item in number_list]
        for future in concurrent.futures.as_completed(futures):
            print(future.result())
    print("Process pool execution in " +
          str(time.time() - start_time_2), "seconds")
          
        

使用 Asyncio 管理事件循環(huán)

Python 的 Asyncio 模塊提供了管理事件、協(xié)程、任務(wù)和線程的方法，以及編寫并發(fā)代碼的原語。此模塊的主要組件和概念包括：

• 事件循環(huán): 在Asyncio模塊中，每一個進程都有一個事件循環(huán)。
• 協(xié)程: 這是子程序的泛化概念。協(xié)程可以在執(zhí)行期間暫停，這樣就可以等待外部的處理（例如IO）完成之后，從之前暫停的地方恢復(fù)執(zhí)行。
• Futures: 定義了 Future 對象，和 concurrent.futures 模塊一樣，表示尚未完成的計算。
• Tasks: 這是Asyncio的子類，用于封裝和管理并行模式下的協(xié)程。

Asyncio 提供了以下方法來管理事件循環(huán)：

• loop = get_event_loop(): 得到當(dāng)前上下文的事件循環(huán)。
• loop.call_later(time_delay, callback, argument): 延后 time_delay 秒再執(zhí)行 callback 方法。
• loop.call_soon(callback, argument): 盡可能快調(diào)用 callback, call_soon() 函數(shù)結(jié)束，主線程回到事件循環(huán)之后就會馬上調(diào)用 callback 。
• loop.time(): 以float類型返回當(dāng)前時間循環(huán)的內(nèi)部時間。
• asyncio.set_event_loop(): 為當(dāng)前上下文設(shè)置事件循環(huán)。
• asyncio.new_event_loop(): 根據(jù)此策略創(chuàng)建一個新的時間循環(huán)并返回。
• loop.run_forever(): 在調(diào)用 stop() 之前將一直運行。

示例代碼如下所示

          
            import asyncio
import datetime
import time


def fuction_1(end_time, loop):
    print("function_1 called")
    if(loop.time() + 1.0) < end_time:
        loop.call_later(1, fuction_2, end_time, loop)
    else:
        loop.stop()


def fuction_2(end_time, loop):
    print("function_2 called")
    if(loop.time() + 1.0) < end_time:
        loop.call_later(1, function_3, end_time, loop)
    else:
        loop.stop()


def function_3(end_time, loop):
    print("function_3 called")
    if(loop.time() + 1.0) < end_time:
        loop.call_later(1, fuction_1, end_time, loop)
    else:
        loop.stop()


def function_4(end_time, loop):
    print("function_4 called")
    if(loop.time() + 1.0) < end_time:
        loop.call_later(1, function_4, end_time, loop)
    else:
        loop.stop()


loop = asyncio.get_event_loop()

end_loop = loop.time() + 9.0
loop.call_soon(fuction_1, end_loop, loop)
loop.run_forever()
loop.close()
          
        

使用 Asyncio 管理協(xié)程

示例代碼如下所示

          
            import asyncio
import time
from random import randint


@asyncio.coroutine
def StartState():
    print("Start State called \n")
    input_val = randint(0, 1)
    time.sleep(1)
    if input_val == 0:
        result = yield from State2(input_val)
    else:
        result = yield from State1(input_val)
    print("Resume of the Transition:\nStart State calling" + result)


@asyncio.coroutine
def State1(transition_value):
    outputVal = str("State 1 with transition value=%s \n" % (transition_value))
    input_val = randint(0, 1)
    time.sleep(1)
    print("...Evaluating...")
    if input_val == 0:
        result = yield from State3(input_val)
    else:
        result = yield from State2(input_val)


@asyncio.coroutine
def State2(transition_value):
    outputVal = str("State 2 with transition value= %s \n" %
                    (transition_value))
    input_Val = randint(0, 1)
    time.sleep(1)
    print("...Evaluating...")
    if (input_Val == 0):
        result = yield from State1(input_Val)
    else:
        result = yield from State3(input_Val)
    result = "State 2 calling " + result
    return outputVal + str(result)


@asyncio.coroutine
def State3(transition_value):
    outputVal = str("State 3 with transition value = %s \n" %
                    (transition_value))
    input_val = randint(0, 1)
    time.sleep(1)
    print("...Evaluating...")
    if(input_val == 0):
        result = yield from State1(input_val)
    else:
        result = yield from State2(input_val)
    result = "State 3 calling " + result
    return outputVal + str(result)


@asyncio.coroutine
def EndState(transition_value):
    outputVal = str("End State With transition value = %s \n" %
                    (transition_value))
    print("...Stop Computation...")
    return outputVal


if __name__ == "__main__":
    print("Finites State Machine simulation with Asyncio Coroutine")
    loop = asyncio.get_event_loop()
    loop.run_until_complete(StartState())
          
        

使用 Asyncio 控制任務(wù)

示例代碼如下所示

          
            import asyncio


@asyncio.coroutine
def factorial(number):
    f = 1
    for i in range(2, number + 1):
        print("Asyncio.Task:Compute factorial(%s)" % (i))
        yield from asyncio.sleep(1)
        f *= i
    print("Asyncio.Task - factorial(%s) = %s" % (number, f))


@asyncio.coroutine
def fibonacci(number):
    a, b = 0, 1
    for i in range(number):
        print("Asyncio.Task:Complete fibonacci (%s)" % (i))
        yield from asyncio.sleep(1)
        a, b = b, a+b
    print("Asyncio.Task - fibonaci (%s)= %s" % (number, a))


@asyncio.coroutine
def binomialCoeff(n, k):
    result = 1
    for i in range(1, k+1):
        result = result * (n-i+1) / i
        print("Asyncio.Task:Compute binomialCoeff (%s)" % (i))
        yield from asyncio.sleep(1)
    print("Asyncio.Task - binomialCoeff (%s,%s) = %s" % (n, k, result))


if __name__ == "__main__":
    tasks = [asyncio.Task(factorial(10)), asyncio.Task(
        fibonacci(10)), asyncio.Task(binomialCoeff(20, 10))]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()
          
        

使用Asyncio和Futures

示例代碼如下所示

          
            import asyncio
import sys


@asyncio.coroutine
def first_coroutine(future, N):
    count = 0
    for i in range(1, N + 1):
        count = count + i
    yield from asyncio.sleep(4)
    future.set_result(
        "first coroutine (sum of N integers) result = " + str(count))


@asyncio.coroutine
def second_coroutine(future, N):
    count = 1
    for i in range(2, N + 1):
        count *= i
    yield from asyncio.sleep(3)
    future.set_result("second coroutine (factorial) result = " + str(count))


def got_result(future):
    print(future.result())


if __name__ == "__main__":
    N1 = 1
    N2 = 1
    loop = asyncio.get_event_loop()
    future1 = asyncio.Future()
    future2 = asyncio.Future()
    tasks = [
        first_coroutine(future1, N1),
        second_coroutine(future2, N2)
    ]
    future1.add_done_callback(got_result)
    future2.add_done_callback(got_result)
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()
          
        

分布式編程

略

GPU 編程

略

相關(guān)參考

• Python并行編程 中文版