from peewee import *
from collections import deque
from threading import Thread
import time
import datetime
import traceback as tr
# thread_safe=True
database = SqliteDatabase('test1.db', check_same_thread=False)
class Test(Model):
ts = IntegerField()
class Meta:
database = database
Test.create_table()
class ThreadTest(object):
def __init__(self):
pass
def f1(self):
while True:
try:
time.sleep(0.1)
p = Test(ts=int(time.time() * 1000))
q.append(p)
except Exception:
print(f'{time.ctime()}, {tr.format_exc()}')
def f2(self):
while True:
try:
time.sleep(0.1)
p = Test(ts=int(time.time() * 1000))
q.append(p)
except Exception:
print(f'{time.ctime()}, {tr.format_exc()}')
def engine(self):
global q
q = deque(maxlen=10000000)
t1 = Thread(target=self.f1)
t2 = Thread(target=self.f2)
t1.start()
t2.start()
while True:
try:
if q:
print(q)
process = q.pop()
process.save()
else:
time.sleep(0.08)
except Exception:
print(f'{time.ctime()}, {tr.format_exc()}')
if __name__ == '__main__':
__test_case = ThreadTest()
__test_case.engine()
python并发写入sqlite而不报错的方法
©著作权归作者所有,转载或内容合作请联系作者
- 文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
- 文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
- 文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
推荐阅读更多精彩内容
- 上篇文章中将内容写到excel中使用import xlwt包，后来发现文件写入超过65536行的时候就会报错，无法...
- 一 问题：使用AD 时原理图中有元器件引脚未连接但编译时不报错。 二 解决方法：在原理图界面选择菜单栏的工程---...
- 这8种学生永远拿不到高分！早看早受益！ 下面是一位资深班主任总结了8种成绩提不上去的原因，分别对应8类孩子，如果你...
- 我遇到这个问题是因为Androidstudio虚拟内存不足造成的，解决办法就是找到Android studio安装...
- 这8种学生永远拿不到高分！早看早受益！ 下面是一位资深班主任总结了8种成绩提不上去的原因，分别对应8类孩子，如果你...
