python sqlite3 多线程 批量写入 【代码】

1. 随机生成一个数组数据
2. 在多线程里面批量插入数据
 
几个关注点：
sqlite3.connect(_type, check_same_thread=False) 要设置为False
 
批量写的时候，记得要加锁
 
 

import datetime

import random

import sqlite3

import threading

import logging as log

import time



lock = threading.Lock()

class SQLiteDBCls:



    def __init__(self, cache=True):

        _type = ":memory:"



        self.db = sqlite3.connect(_type, check_same_thread=False)



        self.table_name = 'tick_data'



    def create_index(self):



        cmd = 'CREATE INDEX code_ix ON {} (current)'.format(self.table_name)

        with lock:

            try:



                cursor = self.db.cursor()

                cursor.execute(cmd)

            except Exception as e:

                log.info(e)

                self.db.rollback()

            else:

                self.db.commit()



    def create_table(self):

        # cursor = self.db.cursor()

        cmd = 'create table if not exists {} (id INTEGER PRIMARY KEY AUTOINCREMENT,code text,open double,current time)'.format(

            self.table_name)

        with lock:

            try:

                cursor = self.db.cursor()

                cursor.execute(cmd)

            except Exception as e:

                log.info(e)

                self.db.rollback()

            else:

                self.db.commit()



    def add(self, code, price, t):

        cmd = 'insert into {} (code,open,current) values (?,?,?);'.format(self.table_name)

        with lock:

            try:

                cursor = self.db.cursor()

                cursor.execute(cmd, (code, price, t))

            except Exception as e:

                log.info(e)

                self.db.rollback()

            else:

                self.db.commit()



    def batch_add(self, data):



        # 批量加入

        print('===========',threading.current_thread().getName())

        # log.info(threading.current_thread().getName())

        cmd = 'insert into {} (code,open,current) values (?,?,?)'.format(self.table_name)

        with lock:

            try:

                cursor = self.db.cursor()

                cursor.executemany(cmd, data)

            except Exception as e:

                log.info(e)

                self.db.rollback()

            else:

                self.db.commit()



    def result(self):

        cmd = 'select count(*) from `{}`'.format(self.table_name)



        with lock:



            try:

                cursor = self.db.cursor()

                cursor.execute(cmd)

            except Exception as e:

                log.info(e)

                self.db.rollback()

            else:

                return cursor.fetchone()



 

def data_gen():

    minute = 6000

    code = ['123011.SS','110010.SS','112111.SS']

    for i in range(minute):

        current = (datetime.datetime.now()+datetime.timedelta(minutes=i)).strftime('%H:%M:%D')

        data_list =

        for c in code:

            price = 5+random.random()+120

            data = (c,price,current)

            data_list.append(data)

        yield data_list

        # time.sleep(0.5)



app = SQLiteDBCls(cache=True)

app.create_table()

app.create_index()



def data_validation():

    print(app.result())

    app.sync_up()



def multithread_mode():

    total_count = 0

    thread_list = 

    for d in data_gen():

        print(d)

        total_count+=len(d)

        # app.batch_add(d)

        t=threading.Thread(target=app.batch_add,args=(d,))

        thread_list.append(t)

    for t in thread_list:

        t.start()



    for t in thread_list:

        t.join()



    print(total_count)





if __name__=='__main__':

    multithread_mode()

    data_validation()

 假如不加锁会出错：

File "/home/xda/miniconda3/envs/cpy/lib/python3.9/threading.py", line 910, in run

    self._target(*self._args, **self._kwargs)

  File "/home/xda/github/stock_strategy/sqlite_issue_debug.py", line 77, in batch_add

    self.db.commit()

Exception in thread Thread-3824:

Exception in thread Thread-3826:

Traceback (most recent call last):

  File "/home/xda/miniconda3/envs/cpy/lib/python3.9/threading.py", line 973, in _bootstrap_inner

sqlite3.OperationalError: cannot commit - no transaction is activeTraceback (most recent call last):

  File "/home/xda/github/stock_strategy/sqlite_issue_debug.py", line 72, in batch_add



    cursor.executemany(cmd, data)

sqlite3.InterfaceError: Error binding parameter 0 - probably unsupported type.