需求：
小程序后台用的sqllite数据库，刚开始用的时候，没有考虑多线程，而且当时因为数据量少，没有出现过多线程查询报错，现在数据量大了。多线程查询经常报错
ProgrammingError: Recursive use of cursors not allowed.
就是这个头疼的错。在网上查了大量的资料，要么就是加lock=threading.lock(),要么就是加sleep.终究还是解决不了问题。
刚好最近在网上看了一个小哥哥用Queue来解决这个问题。我改进了一下。目前能够使用该方法进行增删改查。查询出来的结果以字典的形式返回。
话不多说，下面上代码
代码
# -*- coding: UTF-8 -*- import sqlite3 import time from Queue import Queue from threading import Thread def sqllite_escape(key_word): key_word = key_word.encode("utf-8") key_word = key_word.replace("'", "''") return key_word class SelectConnect(object): ''' 只能用来查询 ''' def __init__(self): # isolation_level=None为智能提交模式，不需要commit self.conn = sqlite3.connect('resource/data.ta', check_same_thread=False, isolation_level=None) self.conn.execute('PRAGMA journal_mode = WAL') cursor = self.conn.cursor() cursor.execute('PRAGMA synchronous=OFF') self.conn.text_factory = str # 把结果用元祖的形式取出来 self.curosr = self.conn.cursor() self.conn.row_factory = self.dict_factory # 把结果用字典的形式取出来 self.curosr_diction = self.conn.cursor() def commit(self): self.conn.commit() def dict_factory(self, cursor, row): d = {} for idx, col in enumerate(cursor.description): d[col[0]] = row[idx] return d def close_db(self): # self.curosr.close() self.conn.close() class SqliteMultithread(Thread): """ Wrap sqlite connection in a way that allows concurrent requests from multiple threads. This is done by internally queueing the requests and processing them sequentially in a separate thread (in the same order they arrived). """ def __init__(self, filename, autocommit, journal_mode): super(SqliteMultithread, self).__init__() self.filename = filename self.autocommit = autocommit self.journal_mode = journal_mode self.reqs = Queue() # use request queue of unlimited size self.setDaemon(True) # python2.5-compatible self.running = True self.start() def dict_factory(self, cursor, row): # field = [i[0] for i in cursor.description] # value = [dict(zip(field, i)) for i in records] d = {} for idx, col in enumerate(cursor.description): d[col[0]] = row[idx] return d def run(self): if self.autocommit: conn = sqlite3.connect(self.filename, isolation_level=None, check_same_thread=False) else: conn = sqlite3.connect(self.filename, check_same_thread=False) conn.execute('PRAGMA journal_mode = %s' % self.journal_mode) conn.text_factory = str cursor = conn.cursor() cursor.execute('PRAGMA synchronous=OFF') conn.row_factory = self.dict_factory curosr_diction = conn.cursor() curosr_diction.execute('PRAGMA synchronous=OFF') # 把结果用字典的形式取出来 while self.running: req, arg, res = self.reqs.get() if req == '--close--': break elif req == '--commit--': conn.commit() else: # print(arg) curosr_diction.execute(req, arg) # if res: # for rec in cursor: # res.put(rec) # res.put('--no more--') if res: res.put(curosr_diction.fetchall()) if self.autocommit: conn.commit() conn.close() def execute(self, req, arg=None, res=None): """ `execute` calls are non-blocking: just queue up the request and return immediately. """ self.reqs.put((req, arg or tuple(), res)) def executemany(self, req, items): for item in items: self.execute(req, item) def select_all_dict(self, req, arg=None): ''' 直接返回一个list :param req: :param arg: :return: ''' res = Queue() # results of the select will appear as items in this queue self.execute(req, arg, res) rec = res.get() return rec def select_one_dict(self, req, arg=None): ''' 直接返回list里的第一个元素，并且以字典展示 :param req: :param arg: :return: ''' res = Queue() # results of the select will appear as items in this queue self.execute(req, arg, res) rec = res.get() if len(rec) != 0: rec = rec[0] else: rec = None return rec def commit(self): self.execute('--commit--') def close(self): self.execute('--close--') class Cursor(object): ''' 以元祖的形式查询出数据 ''' def __init__(self): old_con = SelectConnect() self.conn = old_con.conn self.curosr = old_con.curosr self.curosr2 = SqliteMultithread('resource/data.ta', autocommit=True, journal_mode="WAL") def execute(self, string, *args): try: if string.startswith('select'): return self.curosr.execute(string, *args) else: return self.curosr2.execute(string, *args) except Exception: print("失败一次") print(string) time.sleep(0.1) self.execute(string, *args) def executescript(self, string): try: self.curosr.executescript(string) except Exception: print("失败一次") print(string) time.sleep(0.1) self.executescript(string) def fetchall(self): return self.curosr.fetchall() def fetchone(self): return self.curosr.fetchone() def rowcount(self): return self.curosr.rowcount def close(self): self.curosr2.running = False self.curosr.close() self.conn.close() class Curosrdiction(object): ''' 以字典的形式查询出数据，建议全部用这种。 ''' def __init__(self): old_con = SelectConnect() self.conn = old_con.conn self.curosrdiction = old_con.curosr_diction self.curosr2 = SqliteMultithread('resource/data.ta', autocommit=True, journal_mode="WAL") def execute(self, string, *args): try: if string.startswith('select'): return self.curosrdiction.execute(string, *args) else: return self.curosr2.execute(string, *args) except Exception: print("失败一次") print(string) time.sleep(0.1) self.execute(string, *args) def executescript(self, string): result = True try: self.curosrdiction.executescript(string) except Exception: print("失败一次") # print(string) time.sleep(0.1) # self.executescript(string) result = False return result def fetchall(self): return self.curosrdiction.fetchall() def fetchone(self): return self.curosrdiction.fetchone() def rowcount(self): return self.curosrdiction.rowcount def select_all_dict(self, string, *args): return self.curosr2.select_all_dict(string, *args) def select_one_dict(self, string, *args): return self.curosr2.select_one_dict(string, *args) def close(self): self.curosr2.running = False self.curosrdiction.close() self.conn.close() def commit(self): self.conn.commit() self.curosr2.commit() # curosr = Cursor() curosr_diction = Curosrdiction() def commit(): curosr_diction.commit() def close_db(): # curosr.close() curosr_diction.close()
