问题描述:
使用django自带的test做测试,尝试去数据库中取数据,主线程中没有问题,非主线程中取不到数据。
示例代码:
class MyTestCase(TestCase): def setUp(self):
MyModel.objects.create(k='k0', v='v0')
MyModel.objects.create(k='k1', v='v1')
def test_multithread(self):
kv = MyModel.objects.get(k__exact='k0')
print kv def func():
kv = MyModel.objects.get(k__exact='k0')
print kv
t = threading.Thread(target=func, args=())
t.setDaemon(False)
t.start()
主线程中的查询语句正确输出,func函数中的查询语句报错(DoesNotExist: MyModel matching query does not exist.)
解决:
1)不是多线程的原罪,只是数据库的线程安全策略,锁的时间比较长。尝试把func函数改成这样:
def func():
time.sleep(10)
kv = MyModel.objects.get(k__exact='k0')
print kv 能运行成功。所以一开始我猜测是锁的时间比较长。。。 2)真正的问题是这样的,MyTestCase继承自TestCase,而在TestCase中数据库的改动不会commit(或者不会立即commit),所以在另外一个
线程中看不到这些数据。Django提供的另一个测试基类(TransactionTestCase
)可以解决这个问题,试试
class MyTestCase(TransactionTestCase
) 就可以运行正确。 参考:http://*.com/questions/10948537/database-errors-in-django-when-using-threading 补充:
如果以上方法仍然解决不了这个问题,可以选择不在内存中建测试数据库,在setting.py中加入:
import sys
if 'test' in sys.argv:
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': ':memory:',
'TEST':
{
'NAME': 'test_db.sqlite3',
}
},
} 和TransactionTestCase 一起使用。