Python高性能分布式執行框架-Ray

這是別人說的，咱也不敢說，咱也不敢問 ! 了解大致的邏輯就好.你只需要知道他超級牛逼,超級方便

創新互聯公司服務項目包括北林網站建設、北林網站制作、北林網頁制作以及北林網絡營銷策劃等。多年來，我們專注于互聯網行業，利用自身積累的技術優勢、行業經驗、深度合作伙伴關系等，向廣大中小型企業、政府機構等提供互聯網行業的解決方案，北林網站推廣取得了明顯的社會效益與經濟效益。目前，我們服務的客戶以成都為中心已經輻射到北林省份的部分城市，未來相信會繼續擴大服務區域并繼續獲得客戶的支持與信任！

安裝就是簡單的 pip install ray , 需要提醒的就是ray現在只有linux編譯版本, windows就別想著用了,為了這,我硬生生把開發環境從windows切到了linux.

首先來看一下最簡單的Ray程序是如何編寫的。

在Ray里，通過Python注解@ray.remote定義remote函數。使用此注解聲明的函數都會自帶一個默認的方法remote，通過此方法發起的函數調用都是以提交分布式任務的方式異步執行的，函數的返回值是一個對象id，使用ray.get內置操作可以同步獲取該id對應的對象

參考:

高性能分布式執行框架——Ray

取代 Python 多進程！伯克利開源分布式框架 Ray

官方文檔

基于python的高性能實時并行機器學習框架之Ray介紹

Python的函數都有哪些

【常見的內置函數】

1、enumerate(iterable,start=0)

是python的內置函數，是枚舉、列舉的意思，對于一個可迭代的(iterable)/可遍歷的對象(如列表、字符串)，enumerate將其組成一個索引序列，利用它可以同時獲得索引和值。

2、zip(*iterables,strict=False)

用于將可迭代的對象作為參數，將對象中對應的元素打包成一個個元組，然后返回由這些元組組成的列表。如果各個迭代器的元素個數不一致，則返回列表長度與最短的對象相同，利用*號操作符，可以將元組解壓為列表。

3、filter(function,iterable)

filter是將一個序列進行過濾，返回迭代器的對象，去除不滿足條件的序列。

4、isinstance(object,classinfo)

是用來判斷某一個變量或者是對象是不是屬于某種類型的一個函數，如果參數object是classinfo的實例，或者object是classinfo類的子類的一個實例，

返回True。如果object不是一個給定類型的的對象， 則返回結果總是False

5、eval(expression[,globals[,locals]])

用來將字符串str當成有效的表達式來求值并返回計算結果，表達式解析參數expression并作為Python表達式進行求值(從技術上說是一個條件列表)，采用globals和locals字典作為全局和局部命名空間。

【常用的句式】

1、format字符串格式化

format把字符串當成一個模板，通過傳入的參數進行格式化，非常實用且強大。

2、連接字符串

常使用+連接兩個字符串。

3、if...else條件語句

Python條件語句是通過一條或多條語句的執行結果(True或者False)來決定執行的代碼塊。其中if...else語句用來執行需要判斷的情形。

4、for...in、while循環語句

循環語句就是遍歷一個序列，循環去執行某個操作，Python中的循環語句有for和while。

5、import導入其他腳本的功能

有時需要使用另一個python文件中的腳本，這其實很簡單，就像使用import關鍵字導入任何模塊一樣。

[Django] celery的替代品 funboost

Django開發web應用的過程中，一個老大難問題是異步調度問題。例如用戶傳來一個非常耗時的請求，這時候最好的處理方式是先把這個操作請求記錄下來，先響應請求，等后面有空的時候再去計算，而不是讓用戶干等著著急。

這種優化方式就是典型的生產者+消息隊列+消費者設計模式，而Django框架本身并沒有直接提供該設計模式的實現，大多教程都是利用第三方組件celery+redis來實現這個調度。

遺憾的是celery和redis官方都不支持windows，而我習慣的開發環境還是win10，所以需要找一個替代品。經過調研，發現了一個很好的【python分布式函數調度框架——funboost】. 它的優點很多，對Django開發來說，最大的亮點是完全無需啟動第三方服務，即可實現生產消費設計模式。一個 pip install funboost 即可干活，開箱即用。它可以使用SQLite文件來做消息隊列，足以應對小型應用開發。當然也可以使用Kafka這種高級的消息中間件，實現高可用。

要說缺點吧，這個組件的日志打印太啰嗦，而且沒有提供關閉選項，控制臺已被它刷屏。

如何用 Python 構建一個簡單的分布式系統

分布式爬蟲概覽

何謂分布式爬蟲？

通俗的講，分布式爬蟲就是多臺機器多個

spider

對多個

url

的同時處理問題，分布式的方式可以極大提高程序的抓取效率。

構建分布式爬蟲通暢需要考慮的問題

（1）如何能保證多臺機器同時抓取同一個URL？

（2）如果某個節點掛掉，會不會影響其它節點，任務如何繼續？

（3）既然是分布式，如何保證架構的可伸縮性和可擴展性？不同優先級的抓取任務如何進行資源分配和調度？

基于上述問題，我選擇使用celery作為分布式任務調度工具，是分布式爬蟲中任務和資源調度的核心模塊。它會把所有任務都通過消息隊列發送給各個分布式節點進行執行，所以可以很好的保證url不會被重復抓取；它在檢測到worker掛掉的情況下，會嘗試向其他的worker重新發送這個任務信息，這樣第二個問題也可以得到解決；celery自帶任務路由，我們可以根據實際情況在不同的節點上運行不同的抓取任務（在實戰篇我會講到）。本文主要就是帶大家了解一下celery的方方面面(有celery相關經驗的同學和大牛可以直接跳過了)

Celery知識儲備

celery基礎講解

按celery官網的介紹來說

Celery

是一個簡單、靈活且可靠的，處理大量消息的分布式系統，并且提供維護這樣一個系統的必需工具。它是一個專注于實時處理的任務隊列，同時也支持任務調度。

下面幾個關于celery的核心知識點

broker：翻譯過來叫做中間人。它是一個消息傳輸的中間件，可以理解為一個郵箱。每當應用程序調用celery的異步任務的時候，會向broker傳遞消息，而后celery的worker將會取到消息，執行相應程序。這其實就是消費者和生產者之間的橋梁。

backend:

通常程序發送的消息，發完就完了，可能都不知道對方時候接受了。為此，celery實現了一個backend，用于存儲這些消息以及celery執行的一些消息和結果。

worker:

Celery類的實例，作用就是執行各種任務。注意在celery3.1.25后windows是不支持celery

worker的！

producer:

發送任務，將其傳遞給broker

beat:

celery實現的定時任務。可以將其理解為一個producer，因為它也是通過網絡調用定時將任務發送給worker執行。注意在windows上celery是不支持定時任務的！

下面是關于celery的架構示意圖，結合上面文字的話應該會更好理解

由于celery只是任務隊列，而不是真正意義上的消息隊列，它自身不具有存儲數據的功能，所以broker和backend需要通過第三方工具來存儲信息，celery官方推薦的是

RabbitMQ和Redis，另外mongodb等也可以作為broker或者backend，可能不會很穩定，我們這里選擇Redis作為broker兼backend。

實際例子

先安裝celery

pip

install

celery

我們以官網給出的例子來做說明，并對其進行擴展。首先在項目根目錄下，這里我新建一個項目叫做celerystudy，然后切換到該項目目錄下，新建文件tasks.py，然后在其中輸入下面代碼

這里我詳細講一下代碼：我們先通過app=Celery()來實例化一個celery對象，在這個過程中，我們指定了它的broker，是redis的db

2,也指定了它的backend,是redis的db3,

broker和backend的連接形式大概是這樣

redis://:password@hostname:port/db_number

然后定義了一個add函數，重點是@app.task，它的作用在我看來就是將add()

注冊為一個類似服務的東西，本來只能通過本地調用的函數被它裝飾后，就可以通過網絡來調用。這個tasks.py中的app就是一個worker。它可以有很多任務，比如這里的任務函數add。我們再通過在命令行切換到項目根目錄，執行

celery

-A

tasks

worker

-l

info

啟動成功后就是下圖所示的樣子

這里我說一下各個參數的意思，-A指定的是app(即Celery實例)所在的文件模塊，我們的app是放在tasks.py中，所以這里是

tasks；worker表示當前以worker的方式運行，難道還有別的方式？對的，比如運行定時任務就不用指定worker這個關鍵字;

-l

info表示該worker節點的日志等級是info，更多關于啟動worker的參數(比如-c、-Q等常用的)請使用

celery

worker

--help

進行查看

將worker啟動起來后，我們就可以通過網絡來調用add函數了。我們在后面的分布式爬蟲構建中也是采用這種方式分發和消費url的。在命令行先切換到項目根目錄，然后打開python交互端

from

tasks

import

addrs

=

add.delay(2,

2)

這里的add.delay就是通過網絡調用將任務發送給add所在的worker執行，這個時候我們可以在worker的界面看到接收的任務和計算的結果。

這里是異步調用，如果我們需要返回的結果，那么要等rs的ready狀態true才行。這里add看不出效果，不過試想一下，如果我們是調用的比較占時間的io任務，那么異步任務就比較有價值了

上面講的是從Python交互終端中調用add函數，如果我們要從另外一個py文件調用呢？除了通過import然后add.delay()這種方式，我們還可以通過send_task()這種方式，我們在項目根目錄另外新建一個py文件叫做

excute_tasks.py，在其中寫下如下的代碼

from

tasks

import

addif

__name__

==

'__main__':

add.delay(5,

10)

這時候可以在celery的worker界面看到執行的結果

此外，我們還可以通過send_task()來調用，將excute_tasks.py改成這樣

這種方式也是可以的。send_task()還可能接收到為注冊（即通過@app.task裝飾）的任務，這個時候worker會忽略這個消息

定時任務

上面部分講了怎么啟動worker和調用worker的相關函數，這里再講一下celery的定時任務。

爬蟲由于其特殊性，可能需要定時做增量抓取，也可能需要定時做模擬登陸，以防止cookie過期，而celery恰恰就實現了定時任務的功能。在上述基礎上，我們將tasks.py文件改成如下內容

然后先通過ctrl+c停掉前一個worker，因為我們代碼改了，需要重啟worker才會生效。我們再次以celery

-A

tasks

worker

-l

info這個命令開啟worker。

這個時候我們只是開啟了worker，如果要讓worker執行任務，那么還需要通過beat給它定時發送，我們再開一個命令行，切換到項目根目錄，通過

這樣就表示定時任務已經開始運行了。

眼尖的同學可能看到我這里celery的版本是3.1.25，這是因為celery支持的windows最高版本是3.1.25。由于我的分布式微博爬蟲的worker也同時部署在了windows上，所以我選擇了使用

3.1.25。如果全是linux系統，建議使用celery4。

此外，還有一點需要注意，在celery4后，定時任務（通過schedule調度的會這樣，通過crontab調度的會馬上執行）會在當前時間再過定時間隔執行第一次任務，比如我這里設置的是60秒的間隔，那么第一次執行add會在我們通過celery

beat

-A

tasks

-l

info啟動定時任務后60秒才執行；celery3.1.25則會馬上執行該任務

Python中的多進程與多線程/分布式該如何使用

Python提供了非常好用的多進程包multiprocessing，你只需要定義一個函數，Python會替你完成其他所有事情。

借助這個包，可以輕松完成從單進程到并發執行的轉換。

1、新建單一進程

如果我們新建少量進程，可以如下：

import multiprocessing

import time

def func(msg):

for i in xrange(3):

print msg

time.sleep(1)

if __name__ == "__main__":

p = multiprocessing.Process(target=func, args=("hello", ))

p.start()

p.join()

print "Sub-process done."12345678910111213

2、使用進程池

是的，你沒有看錯，不是線程池。它可以讓你跑滿多核CPU，而且使用方法非常簡單。

注意要用apply_async，如果落下async，就變成阻塞版本了。

processes=4是最多并發進程數量。

import multiprocessing

import time

def func(msg):

for i in xrange(3):

print msg

time.sleep(1)

if __name__ == "__main__":

pool = multiprocessing.Pool(processes=4)

for i in xrange(10):

msg = "hello %d" %(i)

pool.apply_async(func, (msg, ))

pool.close()

pool.join()

print "Sub-process(es) done."12345678910111213141516

3、使用Pool，并需要關注結果

更多的時候，我們不僅需要多進程執行，還需要關注每個進程的執行結果，如下：

import multiprocessing

import time

def func(msg):

for i in xrange(3):

print msg

time.sleep(1)

return "done " + msg

if __name__ == "__main__":

pool = multiprocessing.Pool(processes=4)

result = []

for i in xrange(10):

msg = "hello %d" %(i)

result.append(pool.apply_async(func, (msg, )))

pool.close()

pool.join()

for res in result:

print res.get()

print "Sub-process(es) done."1234567891011121314151617181920

2014.12.25更新

根據網友評論中的反饋，在Windows下運行有可能崩潰(開啟了一大堆新窗口、進程)，可以通過如下調用來解決：

multiprocessing.freeze_support()1

附錄（自己的腳本）：

#!/usr/bin/python

import threading

import subprocess

import datetime

import multiprocessing

def dd_test(round, th):

test_file_arg = 'of=/zbkc/test_mds_crash/1m_%s_%s_{}' %(round, th)

command = "seq 100 | xargs -i dd if=/dev/zero %s bs=1M count=1" %test_file_arg

print command

subprocess.call(command,shell=True,stdout=open('/dev/null','w'),stderr=subprocess.STDOUT)

def mds_stat(round):

p = subprocess.Popen("zbkc mds stat", shell = True, stdout = subprocess.PIPE)

out = p.stdout.readlines()

if out[0].find('active') != -1:

command = "echo '0205pm %s round mds status OK, %s' /round_record" %(round, datetime.datetime.now())

command_2 = "time (ls /zbkc/test_mds_crash/) 2/round_record"

command_3 = "ls /zbkc/test_mds_crash | wc -l /round_record"

subprocess.call(command,shell=True)

subprocess.call(command_2,shell=True)

subprocess.call(command_3,shell=True)

return 1

else:

command = "echo '0205 %s round mds status abnormal, %s, %s' /round_record" %(round, out[0], datetime.datetime.now())

subprocess.call(command,shell=True)

return 0

#threads = []

for round in range(1, 1600):

pool = multiprocessing.Pool(processes = 10) #使用進程池

for th in range(10):

# th_name = "thread-" + str(th)

# threads.append(th_name) #添加線程到線程列表

# threading.Thread(target = dd_test, args = (round, th), name = th_name).start() #創建多線程任務

pool.apply_async(dd_test, (round, th))

pool.close()

pool.join()

#等待線程完成

# for t in threads:

# t.join()

if mds_stat(round) == 0:

subprocess.call("zbkc -s",shell=True)

break

Python分布式進程中你會遇到的坑

寫在前面

小驚大怪

你是不是在用Python3或者在windows系統上編程？最重要的是你對進程和線程不是很清楚？那么恭喜你，在python分布式進程中，會有坑等著你去挖。。。(hahahaha，此處允許我嚇唬一下你)開玩笑的啦，不過，如果你知道序列中不支持匿名函數，那這個坑就和你say byebye了。好了話不多數，直接進入正題。

分布式進程

正如大家所知道的Process比Thread更穩定，而且Process可以分布到多臺機器上，而Thread最多只能分布到同一臺機器的多個CPU上。Python的multiprocessing模塊不但支持多進程，其中managers子模塊還支持把多進程分布到多臺機器上。一個服務進程可以作為調度者，將任務分布到其他多個進程中，依靠網絡通信。由于managers模塊封裝很好，不必了解網絡通信的細節，就可以很容易地編寫分布式多進程程序。

代碼記錄

舉個例子

如果我們已經有一個通過Queue通信的多進程程序在同一臺機器上運行，現在，由于處理任務的進程任務繁重，希望把發送任務的進程和處理任務的進程分布到兩臺機器上，這應該怎么用分布式進程來實現呢？你已經知道了原有的Queue可以繼續使用，而且通過managers模塊把Queue通過網絡暴露出去，就可以讓其他機器的進程來訪問Queue了。好，那我們就這么干！

寫個task_master.py

我們先看服務進程。服務進程負責啟動Queue，把Queue注冊到網絡上，然后往Queue里面寫入任務。

請注意，當我們在一臺機器上寫多進程程序時，創建的Queue可以直接拿來用，但是，在分布式多進程環境下，添加任務到Queue不可以直接對原始的task_queue進行操作，那樣就繞過了QueueManager的封裝，必須通過manager.get_task_queue()獲得的Queue接口添加。然后，在另一臺機器上啟動任務進程（本機上啟動也可以）

寫個task_worker.py

任務進程要通過網絡連接到服務進程，所以要指定服務進程的IP。

運行結果

現在，可以試試分布式進程的工作效果了。先啟動task_master.py服務進程：

task_master.py進程發送完任務后，開始等待result隊列的結果。現在啟動task_worker.py進程：

看到沒，結果都出錯了，我們好好分析一下到底哪出錯了。。。

錯誤分析

在task_master.py的報錯提示中，我們知道它說lambda錯誤，這是因為序列化不支持匿名函數，所以我們得修改代碼，重新對queue用QueueManager進行封裝放到網絡中。

其中task_queue和result_queue是兩個隊列，分別存放任務和結果。它們用來進行進程間通信，交換對象。

因為是分布式的環境，放入queue中的數據需要等待Workers機器運算處理后再進行讀取，這樣就需要對queue用QueueManager進行封裝放到網絡中，這是通過上面的2行代碼來實現的。我們給return_task_queue的網絡調用接口取了一個名get_task_queue,而return_result_queue的名字是get_result_queue，方便區分對哪個queue進行操作。task.put(n)即是對task_queue進行寫入數據，相當于分配任務。而result.get()即是等待workers機器處理后返回的結果。

值得注意 在windows系統中你必須要寫IP地址，而其他操作系統比如linux操作系統則就不要了。

修改后的代碼

在task_master.py中修改如下：

在task_worker.py中修改如下：

先運行task_master.py,然后再運行task_worker.py

(1)task_master.py運行結果如下

(2)task_worker.py運行結果如下

知識補充

這個簡單的Master/Worker模型有什么用？其實這就是一個簡單但真正的分布式計算，把代碼稍加改造，啟動多個worker，就可以把任務分布到幾臺甚至幾十臺機器上，比如把計算n*n的代碼換成發送郵件，就實現了郵件隊列的異步發送。

Queue對象存儲在哪？注意到task_worker.py中根本沒有創建Queue的代碼，所以，Queue對象存儲在task_master.py進程中：

而Queue之所以能通過網絡訪問，就是通過QueueManager實現的。由于QueueManager管理的不止一個Queue，所以，要給每個Queue的網絡調用接口起個名字，比如get_task_queue。task_worker這里的QueueManager注冊的名字必須和task_manager中的一樣。對比上面的例子，可以看出Queue對象從另一個進程通過網絡傳遞了過來。只不過這里的傳遞和網絡通信由QueueManager完成。

authkey有什么用？這是為了保證兩臺機器正常通信，不被其他機器惡意干擾。如果task_worker.py的authkey和task_master.py的authkey不一致，肯定連接不上。

當前名稱：Python的分布式函數的簡單介紹
鏈接地址：http://newbst.com/article46/hsephg.html

成都網站建設公司_創新互聯，為您提供網站維護、域名注冊、網站建設、網站收錄、企業建站、網站制作

廣告

聲明：本網站發布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源： 創新互聯