Я успешно решил проблему. Я написал тестовый скрипт, который вызывает этот способ для изоляции сеансов
Вот мой тестовый код:
# coding=utf-8
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.ext.automap import automap_base
from sqlalchemy import or_
from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import NullPool
from config.settings import MYSQLDB_SETTING
import threading
import multiprocessing
import time
class SqlalchemyDriver(object):
def __init__(self, host, user, password, port, database):
db = "mysql+mysqldb://{user}:{password}@{host}:{port}/{database}?"\
"charset=utf8".format(
user=user,
password=password,
host=host,
port=port,
database=database
)
# self.engine = create_engine(db, poolclass=NullPool)
self.engine = create_engine(db, pool_pre_ping=True, pool_size=10, max_overflow=10, pool_timeout=30)
self.Session = scoped_session(sessionmaker(bind=self.engine, autoflush=False))
drv = SqlalchemyDriver(
host=MYSQLDB_SETTING["host"],
user=MYSQLDB_SETTING["user"],
password=MYSQLDB_SETTING["password"],
port=MYSQLDB_SETTING["port"],
database=MYSQLDB_SETTING["database"]
)
class MysqldbDriver(object):
def __init__(self):
self.classes = self.get_table_classes()
self.session = drv.Session()
print(id(self.session))
def close_session(self):
drv.Session.remove()
def get_table_classes(self):
base = automap_base()
base.prepare(drv.engine, reflect=True)
return base.classes
def search_device_by_mac(self, mac_address):
time.sleep(1)
res = self.session.query(self.classes.client_management_device). \
filter_by(mac_address=mac_address).first()
print(res)
return res
def query_func():
mysql = MysqldbDriver()
mysql.search_device_by_mac('00:30:18:09:97:af')
def my_test():
while True:
t1 = threading.Thread(target=query_func)
t2 = threading.Thread(target=query_func)
t3 = threading.Thread(target=query_func)
t4 = threading.Thread(target=query_func)
t5 = threading.Thread(target=query_func)
t6 = threading.Thread(target=query_func)
t1.start()
t2.start()
t3.start()
t4.start()
t5.start()
t6.start()
time.sleep(3)
t1.join()
t2.join()
t3.join()
t4.join()
t5.join()
t6.join()
time.sleep(1)
p1 = multiprocessing.Process(target=my_test)
p2 = multiprocessing.Process(target=my_test)
p3 = multiprocessing.Process(target=my_test)
p4 = multiprocessing.Process(target=my_test)
print('running')
p1.start()
p2.start()
p3.start()
p4.start()
time.sleep(1)
p1.join()
p2.join()
p3.join()
p4.join()
time.sleep(1)