Опис сценарію
Припустимо, що існує таблиця mysql, розрізана горизонтально, розкидана між кількома хостами, кожен із яких має n таблиць із шардом.
Що робити, якщо потрібно одночасно отримувати ці таблиці і швидко отримувати результати запиту?
Ось рішення для реалізації цієї вимоги за допомогою бібліотеки asyncio asyncio та асинхронної бібліотеки aiomysql python3.
Демо коду
Імпортне логування
імпорт випадково
імпорт asyncio
З імпорту aiomysql create_pool
# Припустимо, що таблиця mysql розподілена між 8 хостами, кожен з яких має 16 підтаблиць
TBLES = {
"192.168.1.01": "table_000-015", #000-015 означає, що індикація під цією IP є безперервною від table_000 до 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",
}
КОРИСТУВАЧ = "xxx"
PASSWD = "xxxx"
# функція обгортки для виявлення винятків
def query_wrapper(func):
Асинхронна обгортка захисту (*args, **kwargs):
Спробуй:
Await func(*args, **kwargs)
за винятком винятку як e:
print(e)
Зворотна обгортка
# Фактична функція обробки доступу SQL реалізує асинхронні неблокуючі запити через AIOMYSQL
@query_wrapper
Асинхронний захист query_do_something(IP, ДБ, таблиця):
async з create_pool(host=ip, db=db, user=USER, password=PASSWD) як пул:
Async з pool.get() як конном:
async з conn.cursor() як cur:
sql = ("виберіть xxx з {}, де xxxx")
Await cur.execute(sql.format(table))
res = чекати cur.fetchall()
# Тоді зроби щось...
# Згенеруйте чергу доступу до SQL, кожен елемент черги містить функції та параметри для доступу до таблиці
def gen_tasks():
завдання = []
Для IP, tbls у TBLES.items():
cols = re.split('_|-', TBLS)
tblpre = "_".join(cols[:-2])
min_num = int(cols[-2])
max_num = int(cols[-1])
для числа в діапазоні(min_num, max_num+1):
tasks.append(
(query_do_something, ip, 'your_dbname', '{}_{}'.format(tblpre, num))
)
random.shuffle(завдання)
Завдання повернення
# Запускайте чергу запитів на доступ до SQL пакетами
def run_tasks(завдання, batch_len):
Спробуй:
Для idx у діапазоні(0, len(завдання), batch_len):
batch_tasks = завдання[idx:idx+batch_len]
logging.info("поточна партія, start_idx:%s len:%s" % (idx, len(batch_tasks)))
для i в діапазоні(0, len(batch_tasks)):
l = batch_tasks
batch_tasks= asyncio.ensure_future(
l[0](*l[1:])
)
loop.run_until_complete(asyncio.gather(*batch_tasks))
за винятком винятку як e:
logging.warn(e)
# основний метод, який реалізує асинхронний виклик функцій через asyncio
def main():
loop = asyncio.get_event_loop()
завдання = gen_tasks()
batch_len = len(TBLES.keys()) * 5 # все залежить від тебе
run_tasks(завдання, batch_len)
loop.close()
|
Опубліковано 29.11.2017 21:15:55
|
|
|
Опубліковано 30.11.2017 09:03:52
|
Наприклад,