使用 Python 和 Oracle 資料庫實現高併發性
瞭解如何藉助執行緒和併發性提升支援 Oracle 資料庫的 Python 應用程式的吞吐量和響應性。
作者：Yuli Vasiliev
2009 年 4 月釋出
隨著趨勢發展的核心轉向更多而不是更快發展，最大限度地提高併發性的重要性日益凸顯。併發性使得程式設計模式發生了新的轉變，可以編寫非同步程式碼，從而將多個任務分散到一組執行緒或程序中並行工作。如果您不是程式設計新手並且很熟悉 C 或 C++，您可能已經對執行緒和程序有所瞭解，並且知道它們之間的區別。在進行併發程式設計時，執行緒提供了程序的輕量級替代物，在大多數情況下多執行緒較多程序更受青睞。因此，本文將討論如何通過多執行緒來實現併發性。
與很多其他程式語言一樣，在使用多 CPU 計算機時將佔用大量 CPU 的任務分散到 Python 中的多個執行緒中（可以使用 Python 標準庫中的多程序模組實現）可以提高效能。對於單處理器計算機，這樣確實可以並行執行多個操作，而不是隻能在任務間切換且在任何指定時間只能執行一個任務。相反，在將多執行緒的 Python 程式移到一個多 CPU 計算機時，由於全域性直譯器鎖 (GIL) 的原因您不會注意到任何效能提升，Python 使用 GIL 保護內部資料結構，確保在一次只有一個執行緒執行 CPython 虛擬機器。
但是，您可能仍然有興趣向支援資料庫的 Python 程式中新增執行緒以加快其速度。關鍵是 Python 與之互動的底層資料庫很可能安裝在並行處理提交的查詢的高效能伺服器上。這意味著您可以從提交多個查詢到資料庫伺服器並在單獨的執行緒中並行進行的操作中受益，而不是在一個執行緒中一個接一個地按順序發出查詢。
要注意的是：儘管利用任務自身的並行性可以顯著提升應用程式效能，但是我們必須認識到，不是所有任務都可並行執行。例如，在客戶請求的操作（例如轉賬）完成之前，您無法向客戶發出確認電子郵件。很顯然，此類任務必須按特定順序執行。
另外，構建多執行緒程式碼時還要記住，某些並行執行的執行緒可能同時嘗試更改共享的物件，這可能導致資料丟失、資料殘缺，甚至損壞正在更改的物件。要避免此問題，應該控制對共享物件的訪問，使得一個執行緒一次只能使用一個此類物件。幸運的是，利用 Python 可以實施一個鎖定機制來同步對共享物件的訪問（利用執行緒模組中的鎖定工具）。
使用鎖定的缺點是損失了可擴充套件性。設計可擴充套件性時，不要忘記，對一個執行緒內的某個資源進行鎖定將使該資源在所有其他正在執行的執行緒和程序中不可用，直至該鎖定被釋放為止。因此，要確保高效的資源管理，不應過多地使用鎖定，儘可能避免鎖定，如果需要使用鎖定也要儘可能早地釋放該鎖定。
幸運的是，當您處理儲存在 Oracle 資料庫中的資源時不必擔心鎖定。這是因為，在併發環境中對共享資料提供訪問時，Oracle 資料庫將使用其自身的後臺鎖定機制。因此，通常較好的做法是將共享資料儲存在 Oracle 資料庫中，從而由 Oracle 資料庫處理併發性問題。
非同步執行操作也是實現可擴充套件性和受益於併發性的較好方式。在非同步程式設計中，阻塞程式碼排隊等待稍後單獨的執行緒完成，從而確保您的應用程式可以繼續執行其他任務。使用非同步框架（如 Twisted）可以極大地簡化構建非同步應用程式的任務。
本文將簡單介紹如何使用 Python 和 Oracle 資料庫構建併發應用程式，描述如何使用 Python 程式碼利用執行緒與 Oracle 資料庫互動，並解釋如何將 SQL 查詢並行提交到資料庫伺服器而不是依次處理。您還將瞭解如何讓 Oracle 資料庫處理併發性問題以及如何利用 Python 事件驅動的框架 Twisted。
Python 中的多執行緒程式設計
執行緒是並行處理中的一個非常有用的特性。如果您的一個程式正在執行耗時的操作並且可以將其分成若干個獨立的任務並行執行，那麼使用執行緒可以幫助您構建更加高效、快速的程式碼。多執行緒的另一個有趣的用處是可以提高應用程式的響應能力 — 在後臺執行耗時操作的同時，主程式仍然可以做出響應。
當長時間執行的 SQL 語句彼此並無關聯並且可以並行執行時，將這些語句封裝到 Python 的不同執行緒中是不錯的做法。例如，如果 Web 頁面將初始的 SQL 查詢並行提交到資料庫伺服器而不是按順序處理它們（使它們一個接一個地排隊等待），則可顯著減少 Web 頁面的載入時間。
當您需要將某些大物件 (LOB) 上載到資料庫時，也會發現執行緒很有用。以並行方式執行此操作不僅可以減少將 LOB 上載到資料庫所需的整體時間，還可以在後臺進行並行上載的同時保持程式主執行緒的響應能力。
假設您需要將幾個二進位制大物件 (BLOB) 上載到資料庫並將其儲存到 blob_tab 表（您可能已經在自定義資料庫模式中建立了該表），如下所示：
CREATE TABLE blob_tab( id NUMBER PRIMARY KEY, blobdoc BLOB ); CREATE SEQUENCE blob_seq; |
首先，我們來了解一下如何不利用執行緒將 BLOB 一個接一個地儲存到 blob_tab 表中。以下 Python 指令碼可以完成該任務，永久儲存分別使用檔名和 URL 獲得的兩個輸入影象。該示例假設您已經在 usr/pswd 自定義資料庫模式中建立了 blob_tab 表和 blob_seq 序列：
#File: singlethread.py
#Storing BLOBs in a single thread sequentially, one after another
import cx_Oracle
from urllib import urlopen
inputs = []
#if you?ˉre a Windows user, the path could be 'c:/temp/figure1.bmp'
inputs.append(open('/tmp/figure1.bmp', 'rb'))
inputs.append(urlopen('http://localhost/mypictures/figure2.bmp', 'rb'))
#obtaining a connection and predefining a memory area for a BLOB
dbconn = cx_Oracle.connect('usr', 'pswd', '127.0.0.1/XE')
dbconn.autocommit = True
cur = dbconn.cursor()
cur.setinputsizes(blobdoc=cx_Oracle.BLOB)
#executing INSERT statements saving BLOBs to the database
for input in inputs:
blobdoc = input.read()
cur.execute("INSERT INTO blob_tab (ID, BLOBDOC) VALUES(blob_seq.NEXTVAL, :blobdoc)", {'blobdoc':blobdoc})
input.close()
dbconn.close()
|
儘管獲取和儲存 figure1.bmp 和 figure2.bmp 的任務在此處一個接一個地進行，但是，您可能已經猜到，這些任務實際上並不存在順序上的先後關聯性。因此，您可以重構上述程式碼，使其在單個執行緒中讀取和儲存每個影象，從而通過並行處理提升效能。在這種特殊的情況下值得一提的是，您不必協調並行執行的執行緒，從而可以極大地簡化編碼。
以下示例顯示瞭如何利用面向物件的方法重新編寫上述指令碼以使用執行緒。具體來說，該示例說明了如何從 threading 模組擴充套件 Thread 類，針對特定任務對其進行自定義。
#File: multithread.py
#Storing BLOBs in separate threads in parallel
import cx_Oracle
import threading
from urllib import urlopen
#subclass of threading.Thread
class AsyncBlobInsert(threading.Thread):
def __init__(self, cur, input):
threading.Thread.__init__(self)
self.cur = cur
self.input = input
def run(self):
blobdoc = self.input.read()
self.cur.execute("INSERT INTO blob_tab (ID, BLOBDOC) VALUES(blob_seq.NEXTVAL, :blobdoc)", {'blobdoc':blobdoc})
self.input.close()
self.cur.close()
#main thread starts here
inputs = []
inputs.append(open('/tmp/figure1.bmp', 'rb'))
inputs.append(urlopen('http://localhost/_figure2.bmp', 'rb'))
dbconn = cx_Oracle.connect('usr', 'pswd', '127.0.0.1/XE',threaded=True)
dbconn.autocommit = True
for input in inputs:
cur = dbconn.cursor()
cur.setinputsizes(blobdoc=cx_Oracle.BLOB)
th = AsyncBlobInsert(cur, input)
th.start()
|
在上述程式碼中，注意 threaded 屬性的使用，該屬性作為引數傳遞到 cx_Oracle.connect 方法。通過將其設定為 true，您指示 Oracle 資料庫使用 OCI_THREADED 模式（又稱為 threaded 模式），從而指明應用程式正在多執行緒環境中執行。請注意，在此處針對單執行緒應用程式使用 threaded 模式並不是一種好的做法。根據 cx_Oracle 文件，在單執行緒應用程式中將 threaded 引數設定為 true 將使效能下降 10% 到 15%。
在本示例中，您將在兩個執行緒間共享一個連線，但是將為每個執行緒建立一個單獨的遊標物件。此處，讀取 BLOB 然後將其插入資料庫的操作是在 threading.Thread 標準 Python 類中 AsyncBlobInsert 自定義子類的改寫的 run 方法中實現的。因此，要在單獨的執行緒中開始上載 BLOB，您只需建立一個 AsyncBlobInsert 例項，然後呼叫其 start 方法。
這裡要討論一個與指令碼有關的問題。執行時，它不會等到正在啟動的執行緒完成 — 啟動子執行緒後主執行緒將結束，不會等到子執行緒完成。如果您並不希望這樣而是希望程式僅在所有執行緒都完成後再結束，那麼您可以在指令碼末尾呼叫每個 AsyncBlobInsert 例項的 join 方法。這將阻塞主執行緒，使其等待子執行緒的完成。對前面的指令碼進行修改，使其等待 for 迴圈中啟動的所有執行緒完成，如下所示：
... th = [] for i, input in enumerate(inputs): cur = dbconn.cursor() cur.setinputsizes(blobdoc=cx_Oracle.BLOB) th.append(AsyncBlobInsert(cur, input)) th[i].start() #main thread waits until all child threads are done for t in th: t.join() |
下一節中提供了需要強制主執行緒等待子執行緒完成的示例。
同步對共享資源的訪問
前面的示例顯示了一個多執行緒的 Python 應用程式，該程式處理幾個彼此並無關聯的任務，因此很容易分離並放到不同的執行緒中進行並行處理。但是在實際中，您經常需要處理彼此相互關聯的操作，並且需要在某個時刻進行同步。
作為單個程序的一部分，執行緒共享相同的全域性記憶體，因此可以通過共享資源（如變數、類例項、流和檔案）在彼此之間傳遞資訊。但是，這種線上程間交換資訊的簡單方法是有條件的 — 當修改的物件可以同時在另一執行緒中訪問和/或修改時，您確實要非常謹慎。因此，如果能夠避免衝突，使用一個機制來同步對共享資料的訪問，這將是很有用的。
為幫助解決這一問題，Python 允許您指定鎖定，然後可以由某個執行緒取得該鎖定以確保對該執行緒中您所使用的資料結構進行獨佔訪問。Threading 模組附帶有 Lock 方法，您可以使用該方法指定鎖定。但是請注意，使用 threading.Lock 方法指定的鎖定最初處於未鎖定狀態。要鎖定一個分配的鎖，您需要顯式呼叫該鎖定物件的 acquire 方法。之後，可以對需要鎖定的物件執行操作。例如，當向執行緒中的 stdout 標準輸出流進行寫入時，您可能需要使用鎖，以免其他使用 stdout 的執行緒發生重疊。進行此操作後，您需要使用鎖定物件的 release 方法釋放該鎖，以使釋放的資料結構可用於其他執行緒中的進一步處理。
關於鎖定要注意的是，它們並不繫結到單個執行緒。在一個執行緒中指定的鎖，可以由另一個執行緒獲得，並由第三個執行緒釋放。以下指令碼例舉了實際操作中的一個簡單的鎖。此處，為在子執行緒中進行使用，您在主執行緒中指定了一個鎖，在向 DOM 文件寫入之前獲得它，然後立即釋放。
#File: synchmultithread.py
#Using locks for synchronization in a multithreaded script
import sys
import cx_Oracle
import threading
from xml.dom.minidom import parseString
from urllib import urlopen
#subclass of threading.Thread
class SynchThread(threading.Thread):
def __init__(self, cur, query, dom):
threading.Thread.__init__(self)
self.cur = cur
self.query = query[1]
self.tag = query[0]
self.dom = dom
def run(self):
self.cur.execute(self.query)
rslt = self.cur.fetchone()[0]
self.cur.close()
mutex.acquire()
sal = self.dom.getElementsByTagName('salary')[0]
newtag = self.dom.createElement(self.tag)
newtext = self.dom.createTextNode('%s'%rslt)
newtag.appendChild(newtext)
sal.appendChild(newtag)
mutex.release()
#main thread starts here
domdoc = parseString('<employees><salary/></employees>')
dbconn = cx_Oracle.connect('hr', 'hr', '127.0.0.1/XE',threaded=True)
mutex = threading.Lock()
queries = {}
queries['avg'] = "SELECT AVG(salary) FROM employees"
queries['max'] = "SELECT MAX(salary) FROM employees"
th = []
for i, query in enumerate(queries.items()):
cur = dbconn.cursor()
th.append(SynchThread(cur, query, domdoc))
th[i].start()
#forcing the main thread to wait until all child threads are done
for t in th:
t.join()
#printing out the result xml document
domdoc.writexml(sys.stdout)
|
在上面的指令碼中，您首先在主執行緒中建立了一個文件物件模型 (DOM) 文件物件，然後在並行執行的子執行緒中修改該文件，新增包含從資料庫獲取的資訊的標籤。此處，您將針對 HR 演示模式中的 employees 表使用了兩個簡單的查詢。為避免在向 DOM 物件並行寫入期間發生衝突，您需要在每個子執行緒中獲取在主執行緒中指定的鎖。一個子執行緒獲得該鎖後，另一個子執行緒將無法修改此處處理的 DOM 物件，直至第一個執行緒釋放該鎖。
然後，您可以使用主執行緒同步在各子執行緒中對 DOM 物件所做的更新，在主執行緒中呼叫每個子執行緒物件的 join 方法。之後，您可以在主流中對 DOM 文件物件進行進一步處理。在該特定示例中，您只是將其寫入 stdout 標準輸出流。
因此，您可能已經注意到，此處展示的示例實際上並沒有討論如何鎖定資料庫訪問操作，例如，發出查詢或針對並行執行緒中的同一資料庫表進行更新。實際上，Oracle 資料庫有自己的強大鎖定機制，可確保併發環境中的資料完整性。而您的任務是正確使用這些機制。下一節中，我們將討論如何利用 Oracle 資料庫特性控制對共享資料的併發訪問，從而讓資料庫處理併發性問題。
使 Oracle 資料庫管理併發性
如上所述，當對儲存在 Oracle 資料庫中的共享資料進行訪問或操作時，您不必在 Python 程式碼中手動實施資源鎖定。為解決併發性問題，Oracle 資料庫根據事務概念在後臺使用不同型別的鎖和多版本併發性控制系統。在實際操作中，這意味著，您只需考慮如何正確利用事務以確保正確訪問、更新或更改資料庫資料。具體來說，您必須謹慎地在自動提交事務模式和手動提交事務模式之間做出選擇，將多個 SQL 語句組合到一個事務中時也需小心仔細。最後，必須避免發生併發事務間的破壞性互動。
在這裡，需要記住的是，您在 Python 程式碼中使用的事務與連線而非遊標相關聯，這意味著您可以輕鬆地按照邏輯將使用不同遊標但通過相同連線執行的語句組合到一個事務中。但是，如果您希望實施兩個併發事務，則需要建立兩個不同的連線物件。
在前面的“Python 中的多執行緒程式設計”一節中討論的多執行緒示例中，您將連線物件的 autocommit 模式設定為 true，從而指示 cx_Oracle 模組在每個 INSERT 語句後隱式執行 COMMIT。在這種特定情況下，使用自動提交模式是合理的，因為這樣可以避免子執行緒和主執行緒間的同步，從而可以在主執行緒中手動執行 COMMIT，如下所示：
... #main thread waits until all child threads are done for t in th: t.join() #and then issues a commit dbconn.commit() |
但是，在有些情況下，您需要用到上述方案。考慮以下示例。假設您在兩個並行執行緒中分別執行以下兩個操作。在一個執行緒中，您將採購訂單文件儲存到資料庫中，包括訂單詳細資訊。在另一個執行緒中，您對包含該訂單中涉及產品的相關資訊的表進行修改，更新可供購買的產品數量。
很顯然，上述兩個操作必須封裝到一個事務中。為此，您必須關閉 autocommit 模式，該模式為預設模式。此外，您還將需要使用主執行緒同步並行執行緒，然後顯式執行 COMMIT，如上述程式碼段所示。
雖然上述方案可以輕鬆實現，但在實際中，您可能最希望在資料庫中實施第二個操作，即更新可供購買的產品的數量，將 BEFORE INSERT 觸發器放到儲存訂單詳細資訊的表上，這樣它可以自動更新包含相關產品資訊的表中的相應記錄。這將簡化 Python 端的程式碼並消除編寫多執行緒 Python 指令碼的需求，讓 Oracle 資料庫來處理資料完整性問題。實際上，如果在放入 details 表的 BEFORE INSERT 觸發器中更新產品表時出現問題，Oracle 資料庫將自動回滾將新行插入到 details 表的操作。在 Python 端，需要進行的操作僅是將用於儲存訂單詳細資訊的所有 INSERT 封裝到一個事務中，如下所示：
...
dbconn = cx_Oracle.connect('hr', 'hr', '127.0.0.1/XE',threaded=True)
dbconn.autocommit = False
cur = dbconn.cursor()
...
for detail in details:
id = detail['id']
quantity = person['quantity']
cur.execute("INSERT INTO details(id, quantity) VALUES(:id, :quantity)", {'id':id, 'quantity':quantity})
dbconn.commit()
...
|
使用 Python 事件驅動的框架 Twisted
Twisted 提供了一種不增加複雜性的編碼事件驅動應用程式的好方法，使 Python 中的多執行緒程式設計更加簡單、安全。Twisted 併發性模式基於無阻塞呼叫概念。您呼叫一個函式來請求某些資料並指定一個在請求資料就緒時呼叫的回撥函式。而於此同時，程式可以繼續執行其他任務。
twisted.enterprise.adbapi 模組是一個非同步封裝程式，可用於任何 DB-API 相容的 Python 模組，使您可以以無阻塞模式執行資料庫相關任務。例如，使用它，您的應用程式不必等待資料的連線建立或查詢完成，而是並行執行其他任務。本節將介紹幾個與 Oracle 資料庫互動的 Twisted 應用程式的簡單示例。
Twisted 不隨 Python 提供，需要下載並在裝有 Python 的系統中安裝。您可以從 Twisted Matrix Labs Web 站點 http://twistedmatrix.com 下載適合您 Python 版本的 Twisted 安裝程式包。下載程式包之後，只需在 Twisted 設定嚮導中進行幾次點選即可完成安裝，安裝大約需要一分鐘的時間。
Twisted 是一個事件驅動的框架，因此，其事件迴圈一旦啟動即持續執行，直到事件完成。在 Twisted 中，事件迴圈使用名為 reactor 的物件進行實施。使用 reactor.run 方法啟動 Twisted 事件迴圈，使用 reactor.stop 停止該迴圈。而另一個名為 Deferred 的 Twisted 物件用於管理回撥。以下是簡化了的現實中的 Twisted 事件迴圈和回撥示例。__name__ 測試用於確保解決方案將僅在該模組作為主指令碼呼叫但不匯入時（即，必須從命令列、使用 IDLE Python GUI 或通過單擊圖示呼叫該解決方案）執行。
#File: twistedsimple.py
#A simple example of a Twisted app
from twisted.internet import reactor
from twisted.enterprise import adbapi
def printResult(rslt):
print rslt[0][0]
reactor.stop()
if __name__ == "__main__":
dbpool = adbapi.ConnectionPool('cx_Oracle', user='hr', password ='hr', dsn='127.0.0.1/XE')
empno = 100
deferred = dbpool.runQuery("SELECT last_name FROM employees WHERE employee_id = :empno", {'empno':empno})
deferred.addCallback(printResult)
reactor.run()
|
請注意，twisted.enterprise.adbapi 模組基於標準 DB-API 介面構建，並在後臺使用您在呼叫 adbapi.ConnectionPool 方法時指定的 Python 資料庫模組。甚至您在指定 adbapi.ConnectionPool 輸入引數時可以使用的一組關鍵字也取決於您使用的資料庫模組型別。
儘管與不同的 Python 資料庫模組結合使用時語法上有一些不同，但是通過 twisted.enterprise.adbapi，您可以編寫非同步程式碼，從而可以在後檯安全處理資料庫相關任務的同時，繼續執行您的程式流。以下示例展示了一個以非同步方式查詢資料庫的簡單 Twisted Web 應用程式。該示例假設您已經建立了 blob_tab 表併為其填充了資料（如本文開始部分“Python 中的多執行緒程式設計”一節中所述）。
#File: twistedTCPServer.py
#Querying database asynchronously with Twisted
from twisted.web import resource, server
from twisted.internet import reactor
from twisted.enterprise import adbapi
class BlobLoads(resource.Resource):
def __init__(self, dbconn):
self.dbconn = dbconn
resource.Resource.__init__(self)
def _getBlobs(self, txn, query):
txn.execute(query)
return txn.fetchall()
def render_GET(self, request):
query = "select id, blobdoc from blob_tab"
self.dbconn.runInteraction(self._getBlobs, query).addCallback(
self._writeBlobs, request).addErrback(
self._exception, request)
return server.NOT_DONE_YET
def _writeBlobs(self, results, request):
request.write("""
<html>
<head><title>BLOBs manipulating</title></head>
<body>
<h2>Writing BLOBs from the database to your disk</h2>
""")
for id, blobdoc in results:
request.write("<i>/tmp/picture%s.bmp</i><br/>" % id)
blob = blobdoc.read()
output = open("/tmp/picture%s.bmp" % id, 'wb')
output.write(blob)
output.close()
request.write("""
<p>Operation completed</p>
</body>
</html>
""")
request.finish( )
def _exception(self, error, request):
request.write("Error obtaining BLOBs: %s" % error.getErrorMessage())
request.write("""
<p>Could not complete operation</p>
</body>
</html>
""")
request.finish( )
class SiteResource(resource.Resource):
def __init__(self, dbconn):
resource.Resource.__init__(self)
self.putChild('', BlobLoads(dbconn))
if __name__ == "__main__":
dbconn = adbapi.ConnectionPool('cx_Oracle', user='usr', password ='pswd', dsn='127.0.0.1/XE')
site = server.Site(SiteResource(dbconn))
print "Listening on port 8000"
reactor.listenTCP(8000, site)
reactor.run()
|
執行時，該指令碼在埠 8000 啟動 TCP 伺服器監聽。接受客戶端連線後，該指令碼將下載 blob_tab 資料庫中儲存的所有影象，並將其作為單獨的檔案儲存在 /tmp 資料夾中，然後將相應的訊息傳送回客戶端。要測試應用程式，您需要執行指令碼，然後將瀏覽器指向http://localhost:8000。
關於上述程式碼最應注意的是，它在繼續執行程式流的前提下，以無阻塞模式運行鍼對資料庫發出的查詢。要確保它以此方式工作，可以在對 runInteraction 的呼叫（runInteraction 指示 Twisted 依次對 _getBlobs 和 _writeBlobs 進行非同步呼叫）下插入一些程式碼以增強 render_GET 方法。新插入的程式碼應使用 request.write 方法將一些內容傳送回客戶端，這樣您可以看到，該輸出出現在客戶端瀏覽器的 _writeBlobs 中生成該輸出之前。
結論
當下，併發性在資料密集型應用程式中頻繁使用。高效使用併發性是提升應用程式效能的關鍵所在。編寫併發應用程式最高效的一種方法是使用多執行緒。但是，正如您在本文中所瞭解到的，由於全域性直譯器鎖 (GIL) 的原因，Python 中的多執行緒化對多處理器計算機沒有任何好處 (GIL)。但是，當將其用於開發資料庫密集型程式碼以及非同步、事件驅動的程式碼時，您仍然可以受益於多執行緒。
本文是併發性之路的良好起點，為您提供了有價值的背景資訊，有助於決策如何充分利用併發性來設計支援 Oracle 資料庫的 Python 應用程式。