Descrierea scenariului
Să presupunem că există un tabel mysql care este tăiat orizontal, răspândit pe mai multe gazde, fiecare cu n tabele fragmentate.
Ce ar trebui să fac dacă trebuie să accesez aceste tabele simultan și să obțin rapid rezultatele interogărilor?
Iată o soluție pentru implementarea acestei cerințe folosind biblioteca asyncio asyncio și biblioteca aasincronă aiomysql din python3.
Demonstrație de cod
Înregistrarea importurilor
import random
import asyncio
De la import aiomysql create_pool
# Să presupunem că tabelul mysql este răspândit pe 8 gazde, fiecare cu 16 sub-tabele
TBLES = {
"192.168.1.01": "table_000-015", #000-015 indică faptul că indicația sub acest IP este continuă de la table_000 la table_015
"192.168.1.02": "table_016-031",
"192.168.1.03": "table_032-047",
"192.168.1.04": "table_048-063",
"192.168.1.05": "table_064-079",
"192.168.1.06": "table_080-095",
"192.168.1.07": "table_096-0111",
"192.168.1.08": "table_112-0127",
}
USER = "xxx"
PASSWD = "xxxx"
# funcția wrapper pentru a prinde excepțiile
def query_wrapper(func):
Wrapper async def (*args, **kwargs):
Încearcă:
await func(*args, **kwargs)
cu excepția excepției ca e:
Print(e)
Înveliș de returnare
# Funcția efectivă de gestionare a accesului SQL implementează cereri asincrone, neblocante, prin AIOMYSQL
@query_wrapper
Async Def query_do_something(IP, DB, tabel):
async cu create_pool(host=ip, db=db, user=USER, password=PASSWD) ca pool:
async cu pool.get() ca conn:
async cu conn.cursor() ca cur:
sql = ("selectează xxx din {} unde xxxx")
await cur.execute(sql.format(table))
res = așteaptă cur.fetchall()
# Atunci fă ceva...
# Generează o coadă de acces SQL, fiecare element al cozii conține funcții și parametri pentru accesarea unui tabel
def gen_tasks():
Sarcini = []
pentru IP, TBLS în TBLES.items():
COLS = re.split('_|-', TBL-uri)
tblpre = "_".join(cols[:-2])
min_num = int(cols[-2])
max_num = int(cols[-1])
pentru num în interval(min_num, max_num+1):
tasks.append(
(query_do_something, IP, 'your_dbname', '{}_{}'.format(tblpre, num))
)
random.shuffle(sarcini)
Sarcini de întoarcere
# Rulează coada de cereri de acces SQL în loturi
Def run_tasks(sarcini, batch_len):
Încearcă:
Pentru IDX în Range (0, Len(Task), batch_len):
batch_tasks = sarcini[idx:idx+batch_len]
logging.info("actual batch, start_idx:%s len:%s" % (idx, len(batch_tasks)))
pentru i în interval(0, len(batch_tasks)):
l = batch_tasks
batch_tasks= asyncio.ensure_future(
l[0](*l[1:])
)
loop.run_until_complete(asyncio.gather(*batch_tasks))
cu excepția excepției ca e:
Logging.warn(e)
# metoda principală, care implementează apelul asincron al funcțiilor prin asincio
def main():
buclă = asyncio.get_event_loop()
sarcini = gen_tasks()
batch_len = len(TBLES.keys()) * 5 # totul depinde de tine
run_tasks(sarcini, batch_len)
loop.close()
|
Postat pe 29.11.2017 21:15:55
|
|
|
Postat pe 30.11.2017 09:03:52
|
Ca și cum ar