Django檢視函式執行，不在主執行緒中，直接loop = asyncio.new_event_loop()
# 不能loop = asyncio.get_event_loop() 會觸發RuntimeError: There is no current event loop in thread

因為asyncio程式中的每個執行緒都有自己的事件迴圈，但它只會在主執行緒中為你自動建立一個事件迴圈。所以如果你asyncio.get_event_loop在主執行緒中呼叫一次，它將自動建立一個迴圈物件並將其設定為預設值，但是如果你在一個子執行緒中再次呼叫它，你會得到這個錯誤。相反，您需要線上程啟動時顯式建立/設定事件迴圈：

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

在Django單個檢視中使用asyncio例項程式碼如下（有多個IO任務時）

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關鍵字（python3.5之前使用yield）實現協程
    """
    self.id = 5
    start_time = time.time()
 
    '''
    # 同步執行
    # 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)),]
 
    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.make_future(self.io_task1,)
    results = await asyncio.gather(*tasks)
    return results
 
  async def make_future(self,func,*args):
    future = self.loop.run_in_executor(None,*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任務時）

from django.views import View
import time
from concurrent.futures import ThreadPoolExecutor,as_completed
 
 
class TestThreadView(View):
  def get(self,**kargs):
    start_time = time.time()
    future_set = set()
    tasks = (self.io_task1,self.io_task2,self.io_task1,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 77

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支援我們。