快捷導(dǎo)航

Django如何使用asyncio協(xié)程和ThreadPoolExecutor多線程

更新時間：2020年10月12日 10:53:58 作者：傻白甜++

這篇文章主要介紹了Django如何使用asyncio協(xié)程和ThreadPoolExecutor多線程,文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下

Django視圖函數(shù)執(zhí)行，不在主線程中，直接loop = asyncio.new_event_loop()
# 不能loop = asyncio.get_event_loop() 會觸發(fā)RuntimeError: There is no current event loop in thread

因為asyncio程序中的每個線程都有自己的事件循環(huán)，但它只會在主線程中為你自動創(chuàng)建一個事件循環(huán)。所以如果你asyncio.get_event_loop在主線程中調(diào)用一次，它將自動創(chuàng)建一個循環(huán)對象并將其設(shè)置為默認值，但是如果你在一個子線程中再次調(diào)用它，你會得到這個錯誤。相反，您需要在線程啟動時顯式創(chuàng)建/設(shè)置事件循環(huán)：

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

在Django單個視圖中使用asyncio實例代碼如下（有多個IO任務(wù)時）

from django.views import View
import asyncio
import time
from django.http import JsonResponse
 
 
class TestAsyncioView(View):
  def get(self, request, *args, **kwargs):
    """
    利用asyncio和async await關(guān)鍵字（python3.5之前使用yield）實現(xiàn)協(xié)程
    """
    self.id = 5
    start_time = time.time()
 
    '''
    # 同步執(zhí)行
    # results = [self.io_task1(self.id),
    # self.io_task2(self.id),
    # self.io_task2(self.id)]
    '''
 
 
    loop = asyncio.new_event_loop() # 或 loop = asyncio.SelectorEventLoop()
    asyncio.set_event_loop(loop)
    self.loop = loop
 
    works = [
      asyncio.ensure_future(self.io_task3(5)),
      asyncio.ensure_future(self.io_task3(5)),
      asyncio.ensure_future(self.io_task3(5)),
      asyncio.ensure_future(self.io_task3(5)),
      asyncio.ensure_future(self.io_task3(5)),
 
    ]
 
    try:
 
      results = loop.run_until_complete(asyncio.gather(*works)) # 兩種寫法
      # results = loop.run_until_complete(self.gather_tasks())
    finally:
      loop.close()
    end_time = time.time()
    return JsonResponse({'results': results, 'cost_time': (end_time - start_time)})
 
  async def gather_tasks(self):
 
    tasks = (
      self.make_future(self.io_task1, self.id),
      self.make_future(self.io_task2, self.id),
      self.make_future(self.io_task2, self.id),
      self.make_future(self.io_task1, self.id),
      self.make_future(self.io_task2, self.id),
      self.make_future(self.io_task2, self.id),
    )
    results = await asyncio.gather(*tasks)
    return results
 
  async def make_future(self, func, *args):
    future = self.loop.run_in_executor(None, func, *args)
    response = await future
    return response
 
  def io_task1(self, sleep_time):
    time.sleep(sleep_time)
    return 66
 
  def io_task2(self, sleep_time):
    time.sleep(sleep_time)
    return 77
 
  async def io_task3(self, sleep_time):
    # await asyncio.sleep(sleep_time)
    s = await self.do(sleep_time)
    return s
 
  async def do(self, sleep_time):
    await asyncio.sleep(sleep_time)
    return 66

在Django單個視圖中使用ThreadPoolExecutor實例代碼如下（有多個IO任務(wù)時）

from django.views import View
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
 
 
class TestThreadView(View):
  def get(self, request, *args, **kargs):
    start_time = time.time()
    future_set = set()
    tasks = (self.io_task1, self.io_task2, self.io_task2, self.io_task1, self.io_task2, self.io_task2)
    with ThreadPoolExecutor(len(tasks)) as executor:
      for task in tasks:
        future = executor.submit(task, 5)
        future_set.add(future)
    for future in as_completed(future_set):
      error = future.exception()
      if error is not None:
        raise error
    results = self.get_results(future_set)
    end_time = time.time()
    return JsonResponse({'results': results, 'cost_time': (end_time - start_time)})
 
  def get_results(self, future_set):
 
    results = []
    for future in future_set:
      results.append(future.result())
    return results
 
  def io_task1(self, sleep_time):
    time.sleep(sleep_time)
    return 66
 
  def io_task2(self, sleep_time):
    time.sleep(sleep_time)
    return 77

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: