python访问数据库如何解决高并发_使用 Python 和 Oracle 数据库实现高并发性
隨著趨勢發(fā)展的核心轉向更多而不是更快發(fā)展，最大限度地提高并發(fā)性的重要性日益凸顯。并發(fā)性使得編程模式發(fā)生了新的轉變，可以編寫異步代碼，從而將多個任務分散到一組線程或進程中并行工作。如果您不是編程新手并且很熟悉 C 或 C++，您可能已經(jīng)對線程和進程有所了解，并且知道它們之間的區(qū)別。在進行并發(fā)編程時，線程提供了進程的輕量級替代物，在大多數(shù)情況下多線程較多進程更受青睞。因此，本文將討論如何通過多線程來實現(xiàn)并發(fā)性。
與很多其他編程語言一樣，在使用多 CPU 計算機時將占用大量 CPU 的任務分散到 Python 中的多個線程中(可以使用 Python 標準庫中的多進程模塊實現(xiàn))可以提高性能。對于單處理器計算機，這樣確實可以并行運行多個操作，而不是只能在任務間切換且在任何指定時間只能執(zhí)行一個任務。相反，在將多線程的 Python 程序移到一個多 CPU 計算機時，由于全局解釋器鎖 (GIL) 的原因您不會注意到任何性能提升，Python 使用 GIL 保護內部數(shù)據(jù)結構，確保在一次只有一個線程運行 CPython 虛擬機。
但是，您可能仍然有興趣向支持數(shù)據(jù)庫的 Python 程序中添加線程以加快其速度。關鍵是 Python 與之交互的底層數(shù)據(jù)庫很可能安裝在并行處理提交的查詢的高性能服務器上。這意味著您可以從提交多個查詢到數(shù)據(jù)庫服務器并在單獨的線程中并行進行的操作中受益，而不是在一個線程中一個接一個地按順序發(fā)出查詢。
要注意的是：盡管利用任務自身的并行性可以顯著提升應用程序性能，但是我們必須認識到，不是所有任務都可并行執(zhí)行。例如，在客戶請求的操作(例如轉賬)完成之前，您無法向客戶發(fā)出確認電子郵件。很顯然，此類任務必須按特定順序執(zhí)行。
另外，構建多線程代碼時還要記住，某些并行運行的線程可能同時嘗試更改共享的對象，這可能導致數(shù)據(jù)丟失、數(shù)據(jù)殘缺，甚至損壞正在更改的對象。要避免此問題，應該控制對共享對象的訪問，使得一個線程一次只能使用一個此類對象。幸運的是，利用 Python 可以實施一個鎖定機制來同步對共享對象的訪問(利用線程模塊中的鎖定工具)。
使用鎖定的缺點是損失了可擴展性。設計可擴展性時，不要忘記，對一個線程內的某個資源進行鎖定將使該資源在所有其他正在運行的線程和進程中不可用，直至該鎖定被釋放為止。因此，要確保高效的資源管理，不應過多地使用鎖定，盡可能避免鎖定，如果需要使用鎖定也要盡可能早地釋放該鎖定。
幸運的是，當您處理存儲在 Oracle 數(shù)據(jù)庫中的資源時不必擔心鎖定。這是因為，在并發(fā)環(huán)境中對共享數(shù)據(jù)提供訪問時，Oracle 數(shù)據(jù)庫將使用其自身的后臺鎖定機制。因此，通常較好的做法是將共享數(shù)據(jù)存儲在 Oracle 數(shù)據(jù)庫中，從而由 Oracle 數(shù)據(jù)庫處理并發(fā)性問題。
異步執(zhí)行操作也是實現(xiàn)可擴展性和受益于并發(fā)性的較好方式。在異步編程中，阻塞代碼排隊等待稍后單獨的線程完成，從而確保您的應用程序可以繼續(xù)執(zhí)行其他任務。使用異步框架(如 Twisted)可以極大地簡化構建異步應用程序的任務。
本文將簡單介紹如何使用 Python 和 Oracle 數(shù)據(jù)庫構建并發(fā)應用程序，描述如何使用 Python 代碼利用線程與 Oracle 數(shù)據(jù)庫交互，并解釋如何將 SQL 查詢并行提交到數(shù)據(jù)庫服務器而不是依次處理。您還將了解如何讓 Oracle 數(shù)據(jù)庫處理并發(fā)性問題以及如何利用 Python 事件驅動的框架 Twisted。
Python 中的多線程編程
線程是并行處理中的一個非常有用的特性。如果您的一個程序正在執(zhí)行耗時的操作并且可以將其分成若干個獨立的任務并行執(zhí)行，那么使用線程可以幫助您構建更加高效、快速的代碼。多線程的另一個有趣的用處是可以提高應用程序的響應能力 — 在后臺執(zhí)行耗時操作的同時，主程序仍然可以做出響應。
當長時間運行的 SQL 語句彼此并無關聯(lián)并且可以并行執(zhí)行時，將這些語句封裝到 Python 的不同線程中是不錯的做法。例如，如果 Web 頁面將初始的 SQL 查詢并行提交到數(shù)據(jù)庫服務器而不是按順序處理它們(使它們一個接一個地排隊等待)，則可顯著減少 Web 頁面的加載時間。
當您需要將某些大對象 (LOB) 上載到數(shù)據(jù)庫時，也會發(fā)現(xiàn)線程很有用。以并行方式執(zhí)行此操作不僅可以減少將 LOB 上載到數(shù)據(jù)庫所需的整體時間，還可以在后臺進行并行上載的同時保持程序主線程的響應能力。
假設您需要將幾個二進制大對象 (BLOB) 上載到數(shù)據(jù)庫并將其保存到 blob_tab 表(您可能已經(jīng)在自定義數(shù)據(jù)庫模式中創(chuàng)建了該表)，如下所示：
CREATE TABLE blob_tab(
id NUMBER PRIMARY KEY,
blobdoc BLOB
);
CREATE SEQUENCE blob_seq;
首先，我們來了解一下如何不利用線程將 BLOB 一個接一個地存儲到 blob_tab 表中。以下 Python 腳本可以完成該任務，永久保存分別使用文件名和 URL 獲得的兩個輸入圖像。該示例假設您已經(jīng)在 usr/pswd 自定義數(shù)據(jù)庫模式中創(chuàng)建了 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 的任務在此處一個接一個地進行，但是，您可能已經(jīng)猜到，這些任務實際上并不存在順序上的先后關聯(lián)性。因此，您可以重構上述代碼，使其在單個線程中讀取和存儲每個圖像，從而通過并行處理提升性能。在這種特殊的情況下值得一提的是，您不必協(xié)調并行運行的線程，從而可以極大地簡化編碼。
以下示例顯示了如何利用面向對象的方法重新編寫上述腳本以使用線程。具體來說，該示例說明了如何從 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 屬性的使用，該屬性作為參數(shù)傳遞到 cx_Oracle.connect 方法。通過將其設置為 true，您指示 Oracle 數(shù)據(jù)庫使用 OCI_THREADED 模式(又稱為 threaded 模式)，從而指明應用程序正在多線程環(huán)境中運行。請注意，在此處針對單線程應用程序使用 threaded 模式并不是一種好的做法。根據(jù) cx_Oracle 文檔，在單線程應用程序中將 threaded 參數(shù)設置為 true 將使性能下降 10% 到 15%。
在本示例中，您將在兩個線程間共享一個連接，但是將為每個線程創(chuàng)建一個單獨的游標對象。此處，讀取 BLOB 然后將其插入數(shù)據(jù)庫的操作是在 threading.Thread 標準 Python 類中 AsyncBlobInsert 自定義子類的改寫的 run 方法中實現(xiàn)的。因此，要在單獨的線程中開始上載 BLOB，您只需創(chuàng)建一個 AsyncBlobInsert 實例，然后調用其 start 方法。
這里要討論一個與腳本有關的問題。執(zhí)行時，它不會等到正在啟動的線程完成 — 啟動子線程后主線程將結束，不會等到子線程完成。如果您并不希望這樣而是希望程序僅在所有線程都完成后再結束，那么您可以在腳本末尾調用每個 AsyncBlobInsert 實例的 join 方法。這將阻塞主線程，使其等待子線程的完成。對前面的腳本進行修改，使其等待 for 循環(huán)中啟動的所有線程完成，如下所示：
...
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()
下一節(jié)中提供了需要強制主線程等待子線程完成的示例。
同步對共享資源的訪問
前面的示例顯示了一個多線程的 Python 應用程序，該程序處理幾個彼此并無關聯(lián)的任務，因此很容易分離并放到不同的線程中進行并行處理。但是在實際中，您經(jīng)常需要處理彼此相互關聯(lián)的操作，并且需要在某個時刻進行同步。
作為單個進程的一部分，線程共享相同的全局內存，因此可以通過共享資源(如變量、類實例、流和文件)在彼此之間傳遞信息。但是，這種在線程間交換信息的簡單方法是有條件的 — 當修改的對象可以同時在另一線程中訪問和/或修改時，您確實要非常謹慎。因此，如果能夠避免沖突，使用一個機制來同步對共享數(shù)據(jù)的訪問，這將是很有用的。
為幫助解決這一問題，Python 允許您指定鎖定，然后可以由某個線程取得該鎖定以確保對該線程中您所使用的數(shù)據(jù)結構進行獨占訪問。Threading 模塊附帶有 Lock 方法，您可以使用該方法指定鎖定。但是請注意，使用 threading.Lock 方法指定的鎖定最初處于未鎖定狀態(tài)。要鎖定一個分配的鎖，您需要顯式調用該鎖定對象的 acquire 方法。之后，可以對需要鎖定的對象執(zhí)行操作。例如，當向線程中的 stdout 標準輸出流進行寫入時，您可能需要使用鎖，以免其他使用 stdout 的線程發(fā)生重疊。進行此操作后，您需要使用鎖定對象的 release 方法釋放該鎖，以使釋放的數(shù)據(jù)結構可用于其他線程中的進一步處理。
關于鎖定要注意的是，它們并不綁定到單個線程。在一個線程中指定的鎖，可以由另一個線程獲得，并由第三個線程釋放。以下腳本例舉了實際操作中的一個簡單的鎖。此處，為在子線程中進行使用，您在主線程中指定了一個鎖，在向 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('')
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)
在上面的腳本中，您首先在主線程中創(chuàng)建了一個文檔對象模型 (DOM) 文檔對象，然后在并行運行的子線程中修改該文檔，添加包含從數(shù)據(jù)庫獲取的信息的標簽。此處，您將針對 HR 演示模式中的 employees 表使用了兩個簡單的查詢。為避免在向 DOM 對象并行寫入期間發(fā)生沖突，您需要在每個子線程中獲取在主線程中指定的鎖。一個子線程獲得該鎖后，另一個子線程將無法修改此處處理的 DOM 對象，直至第一個線程釋放該鎖。
然后，您可以使用主線程同步在各子線程中對 DOM 對象所做的更新，在主線程中調用每個子線程對象的 join 方法。之后，您可以在主流中對 DOM 文檔對象進行進一步處理。在該特定示例中，您只是將其寫入 stdout 標準輸出流。
因此，您可能已經(jīng)注意到，此處展示的示例實際上并沒有討論如何鎖定數(shù)據(jù)庫訪問操作，例如，發(fā)出查詢或針對并行線程中的同一數(shù)據(jù)庫表進行更新。實際上，Oracle 數(shù)據(jù)庫有自己的強大鎖定機制，可確保并發(fā)環(huán)境中的數(shù)據(jù)完整性。而您的任務是正確使用這些機制。下一節(jié)中，我們將討論如何利用 Oracle 數(shù)據(jù)庫特性控制對共享數(shù)據(jù)的并發(fā)訪問，從而讓數(shù)據(jù)庫處理并發(fā)性問題。
使 Oracle 數(shù)據(jù)庫管理并發(fā)性
如上所述，當對存儲在 Oracle 數(shù)據(jù)庫中的共享數(shù)據(jù)進行訪問或操作時，您不必在 Python 代碼中手動實施資源鎖定。為解決并發(fā)性問題，Oracle 數(shù)據(jù)庫根據(jù)事務概念在后臺使用不同類型的鎖和多版本并發(fā)性控制系統(tǒng)。在實際操作中，這意味著，您只需考慮如何正確利用事務以確保正確訪問、更新或更改數(shù)據(jù)庫數(shù)據(jù)。具體來說，您必須謹慎地在自動提交事務模式和手動提交事務模式之間做出選擇，將多個 SQL 語句組合到一個事務中時也需小心仔細。最后，必須避免發(fā)生并發(fā)事務間的破壞性交互。
在這里，需要記住的是，您在 Python 代碼中使用的事務與連接而非游標相關聯(lián)，這意味著您可以輕松地按照邏輯將使用不同游標但通過相同連接執(zhí)行的語句組合到一個事務中。但是，如果您希望實施兩個并發(fā)事務，則需要創(chuàng)建兩個不同的連接對象。
在前面的“Python 中的多線程編程”一節(jié)中討論的多線程示例中，您將連接對象的 autocommit 模式設置為 true，從而指示 cx_Oracle 模塊在每個 INSERT 語句后隱式執(zhí)行 COMMIT。在這種特定情況下，使用自動提交模式是合理的，因為這樣可以避免子線程和主線程間的同步，從而可以在主線程中手動執(zhí)行 COMMIT，如下所示：
...
#main thread waits until all child threads are done
for t in th:
t.join()
#and then issues a commit
dbconn.commit()
但是，在有些情況下，您需要用到上述方案。考慮以下示例。假設您在兩個并行線程中分別執(zhí)行以下兩個操作。在一個線程中，您將采購訂單文檔保存到數(shù)據(jù)庫中，包括訂單詳細信息。在另一個線程中，您對包含該訂單中涉及產(chǎn)品的相關信息的表進行修改，更新可供購買的產(chǎn)品數(shù)量。
很顯然，上述兩個操作必須封裝到一個事務中。為此，您必須關閉 autocommit 模式，該模式為默認模式。此外，您還將需要使用主線程同步并行線程，然后顯式執(zhí)行 COMMIT，如上述代碼段所示。
雖然上述方案可以輕松實現(xiàn)，但在實際中，您可能最希望在數(shù)據(jù)庫中實施第二個操作，即更新可供購買的產(chǎn)品的數(shù)量，將 BEFORE INSERT 觸發(fā)器放到存儲訂單詳細信息的表上，這樣它可以自動更新包含相關產(chǎn)品信息的表中的相應記錄。這將簡化 Python 端的代碼并消除編寫多線程 Python 腳本的需求，讓 Oracle 數(shù)據(jù)庫來處理數(shù)據(jù)完整性問題。實際上，如果在放入 details 表的 BEFORE INSERT 觸發(fā)器中更新產(chǎn)品表時出現(xiàn)問題，Oracle 數(shù)據(jù)庫將自動回滾將新行插入到 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 并發(fā)性模式基于無阻塞調用概念。您調用一個函數(shù)來請求某些數(shù)據(jù)并指定一個在請求數(shù)據(jù)就緒時調用的回調函數(shù)。而于此同時，程序可以繼續(xù)執(zhí)行其他任務。
twisted.enterprise.adbapi 模塊是一個異步封裝程序，可用于任何 DB-API 兼容的 Python 模塊，使您可以以無阻塞模式執(zhí)行數(shù)據(jù)庫相關任務。例如，使用它，您的應用程序不必等待數(shù)據(jù)的連接建立或查詢完成，而是并行執(zhí)行其他任務。本節(jié)將介紹幾個與 Oracle 數(shù)據(jù)庫交互的 Twisted 應用程序的簡單示例。
Twisted 不隨 Python 提供，需要下載并在裝有 Python 的系統(tǒng)中安裝。您可以從 Twisted Matrix Labs Web 站點?http://twistedmatrix.com?下載適合您 Python 版本的 Twisted 安裝程序包。下載程序包之后，只需在 Twisted 設置向導中進行幾次點擊即可完成安裝，安裝大約需要一分鐘的時間。
Twisted 是一個事件驅動的框架，因此，其事件循環(huán)一旦啟動即持續(xù)運行，直到事件完成。在 Twisted 中，事件循環(huán)使用名為 reactor 的對象進行實施。使用 reactor.run 方法啟動 Twisted 事件循環(huán)，使用 reactor.stop 停止該循環(huán)。而另一個名為 Deferred 的 Twisted 對象用于管理回調。以下是簡化了的現(xiàn)實中的 Twisted 事件循環(huán)和回調示例。__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 數(shù)據(jù)庫模塊。甚至您在指定 adbapi.ConnectionPool 輸入?yún)?shù)時可以使用的一組關鍵字也取決于您使用的數(shù)據(jù)庫模塊類型。
盡管與不同的 Python 數(shù)據(jù)庫模塊結合使用時語法上有一些不同，但是通過 twisted.enterprise.adbapi，您可以編寫異步代碼，從而可以在后臺安全處理數(shù)據(jù)庫相關任務的同時，繼續(xù)執(zhí)行您的程序流。以下示例展示了一個以異步方式查詢數(shù)據(jù)庫的簡單 Twisted Web 應用程序。該示例假設您已經(jīng)創(chuàng)建了 blob_tab 表并為其填充了數(shù)據(jù)(如本文開始部分“Python 中的多線程編程”一節(jié)中所述)。
#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("""
BLOBs manipulatingWriting BLOBs from the database to your disk
""")
for id, blobdoc in results:
request.write("/tmp/picture%s.bmp
" % id)
blob = blobdoc.read()
output = open("/tmp/picture%s.bmp" % id, 'wb')
output.write(blob)
output.close()
request.write("""
Operation completed
""")
request.finish( )
def _exception(self, error, request):
request.write("Error obtaining BLOBs: %s" % error.getErrorMessage())
request.write("""
Could not complete operation
總結
以上是生活随笔為你收集整理的python访问数据库如何解决高并发_使用 Python 和 Oracle 数据库实现高并发性的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： float去掉小数点之后_float类型
- 下一篇： websocket python爬虫_p