我正在一个Django web应用程序中工作,它需要查询PostgreSQL数据库。当使用Pythonthreading接口实现并发时,我得到了查询项的DoesNotExist
错误。当然,这些错误在按顺序执行查询时不会发生。在
让我展示一个单元测试,它是我用来演示意外行为的:
class ThreadingTest(TestCase):
fixtures = ['demo_city',]
def test_sequential_requests(self):
"""
A very simple request to database, made sequentially.
A fixture for the cities has been loaded above. It is supposed to be
six cities in the testing database now. We will made a request for
each one of the cities sequentially.
"""
for number in range(1, 7):
c = City.objects.get(pk=number)
self.assertEqual(c.pk, number)
def test_threaded_requests(self):
"""
Now, to test the threaded behavior, we will spawn a thread for
retrieving each city from the database.
"""
threads = []
cities = []
def do_requests(number):
cities.append(City.objects.get(pk=number))
[threads.append(threading.Thread(target=do_requests, args=(n,))) for n in range(1, 7)]
[t.start() for t in threads]
[t.join() for t in threads]
self.assertNotEqual(cities, [])
如您所见,第一个测试按顺序执行一些数据库请求,这些请求确实没有问题。然而,第二个测试执行完全相同的请求,但是每个请求都在一个线程中生成。这实际上是失败的,返回一个DoesNotExist
异常。在
执行此单元测试的输出如下所示:
^{pr2}$。。。其他线程返回类似的输出。。。
Exception in thread Thread-6:
Traceback (most recent call last):
File "/usr/lib/python2.6/threading.py", line 532, in __bootstrap_inner
self.run()
File "/usr/lib/python2.6/threading.py", line 484, in run
self.__target(*self.__args, **self.__kwargs)
File "/home/jose/Work/cesta/trunk/src/cesta/core/tests/threadbase.py", line 45, in do_requests
cities.append(City.objects.get(pk=number))
File "/home/jose/Work/cesta/trunk/parts/django/django/db/models/manager.py", line 132, in get
return self.get_query_set().get(*args, **kwargs)
File "/home/jose/Work/cesta/trunk/parts/django/django/db/models/query.py", line 349, in get
% self.model._meta.object_name)
DoesNotExist: City matching query does not exist.
FAIL
======================================================================
FAIL: test_threaded_requests (cesta.core.tests.threadbase.ThreadingTest)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/home/jose/Work/cesta/trunk/src/cesta/core/tests/threadbase.py", line 52, in test_threaded_requests
self.assertNotEqual(cities, [])
AssertionError: [] == []
----------------------------------------------------------------------
Ran 2 tests in 0.278s
FAILED (failures=1)
Destroying test database for alias 'default' ('test_cesta')...
请记住,所有这些都发生在PostgreSQL数据库中,它应该是线程安全的,而不是使用SQLite或类似的数据库。测试也是使用PostgreSQL运行的。在
在这一点上,我完全不知道什么会失败。有什么想法或建议吗?在
谢谢!在
编辑:我写了一个小视图来检查它是否能通过测试。以下是视图代码:
def get_cities(request):
queue = Queue.Queue()
def get_async_cities(q, n):
city = City.objects.get(pk=n)
q.put(city)
threads = [threading.Thread(target=get_async_cities, args=(queue, number)) for number in range(1, 5)]
[t.start() for t in threads]
[t.join() for t in threads]
cities = list()
while not queue.empty():
cities.append(queue.get())
return render_to_response('async/cities.html', {'cities': cities},
context_instance=RequestContext(request))
(请不要考虑应用程序的愚蠢之处。请记住,这只是一个概念的证明,不会永远在真正的应用程序。)
结果是代码运行良好,请求在线程中成功发出,并且视图在调用其URL后最终显示城市。在
所以,我认为只有在需要测试代码时,使用线程进行查询才是一个问题。在生产中,它将毫无问题地工作。在
对于成功测试这类代码有什么有用的建议吗?在
这听起来像是交易的问题。如果您在当前请求(或测试)中创建元素,那么它们几乎肯定是在一个未提交的事务中,该事务无法从另一个线程中的单独连接访问。您可能需要manage your transctions manually才能使其工作。在
从文档的这一部分变得更加清晰
现在,事务还没有在TestCase中提交,因此更改对另一个线程不可见。在
尝试使用TransactionTestCase:
TestCase将数据保存在内存中,不向数据库发出提交。可能线程正在尝试直接连接到数据库,而数据还没有提交到那里。请参阅此处的说明: https://docs.djangoproject.com/en/dev/topics/testing/?from=olddocs#django.test.TransactionTestCase
相关问题 更多 >
编程相关推荐