Detailed example of how Python handles concurrency issues through futures

This article mainly introduces Python to deal with concurrency issues through future. It is very good and has reference value. Friends in need can refer to it

future first introduction

Use the following script to have a preliminary understanding of future:

Example 1: Ordinary looping method

import os
import time
import sys
import requests
POP20_CC = (
 "CN IN US ID BR PK NG BD RU JP MX PH VN ET EG DE IR TR CD FR"
).split()
BASE_URL = 'http://flupy.org/data/flags'
DEST_DIR = 'downloads/'
def save_flag(img,filename):
 path = os.path.join(DEST_DIR,filename)
 with open(path,'wb') as fp:
 fp.write(img)
def get_flag(cc):
 url = "{}/{cc}/{cc}.gif".format(BASE_URL,cc=cc.lower())
 resp = requests.get(url)
 return resp.content
def show(text):
 print(text,end=" ")
 sys.stdout.flush()
def download_many(cc_list):
 for cc in sorted(cc_list):
 image = get_flag(cc)
 show(cc)
 save_flag(image,cc.lower()+".gif")
 return len(cc_list)
def main(download_many):
 t0 = time.time()
 count = download_many(POP20_CC)
 elapsed = time.time()-t0
 msg = "\n{} flags downloaded in {:.2f}s"
 print(msg.format(count,elapsed))
if __name__ == '__main__':
 main(download_many)

Example 2: Implemented through future method, here we reuse part of the above code

 from concurrent import futures
from flags import save_flag, get_flag, show, main
MAX_WORKERS = 20
def download_one(cc):
 image = get_flag(cc)
 show(cc)
 save_flag(image, cc.lower()+".gif")
 return cc
def download_many(cc_list):
 workers = min(MAX_WORKERS,len(cc_list))
 with futures.ThreadPoolExecutor(workers) as executor:
 res = executor.map(download_one, sorted(cc_list))
 return len(list(res))
if __name__ == '__main__':
 main(download_many)

Run three times respectively, the average speed of the two is: 13.67 and 1.59s, you can see The difference is still very big.

future

future is an important component of concurrent.futures module and asyncio module

From Starting from python3.4, there are two classes named Future in the standard library: concurrent.futures.Future and asyncio.Future
These two classes have the same function: instances of both Future classes represent things that may be completed or not yet completed. Delayed calculation. Similar to the Deferred class in Twisted and the Future class in the Tornado framework

Note: Normally you should not create a future yourself, but instantiate it by the concurrent framework (concurrent.futures or asyncio)

Reason: future represents something that will eventually happen, and the only way to determine that something will happen is that the execution time has been arranged, so only when something is handed over to the concurrent.futures.Executor subclass for processing, A concurrent.futures.Future instance will be created.
For example: the parameter of the Executor.submit() method is a callable object. After calling this method, the time will be scheduled for the incoming callable object and a

future

# will be returned. ##Client code should not change the state of the future. The concurrency framework will change the state of the future object after the delayed calculation represented by the future ends. We cannot control when the calculation ends.

Both futures have a .done() method. This method does not block. The return value is a Boolean value, indicating whether the callable object linked to the future has been executed. Client code usually does not ask whether the future has finished running, but will wait for notification. Therefore, both Future classes have the .add_done_callback() method. This method has only one parameter, and the type is a callable object. The specified callable object will be called after the future runs. The

.result() method has the same function in the two Future classes: returning the result of the callable object, or re-throwing the exception thrown when executing the callable object. But if the future does not end running, the behavior of the result method in the two Future classes is very different.

For concurrent.futures.Future instances, calling the .result() method will block the thread of the caller until there is a result to return. At this time, the result method can receive an optional timeout. Parameter, if the future does not complete running within the specified time, a TimeoutError exception will be thrown.

The asyncio.Future.result method does not support setting the timeout. It is best to use the yield from structure to obtain future results, but concurrent.futures.Future cannot do this.

No matter Whether it is asyncio or concurrent.futures.Future, there will be several functions that return future, and other functions use future. In the first example, the Executor.map we used uses future, and the return value is an iterator. The iterator The __next__ method calls the result method of each future, so what we get is the result of each future, not the future itself.

Regarding the use of the future.as_completed function, here we use two loops, one with For creating and scheduling future, the other one is used to get the result of future

from concurrent import futures
from flags import save_flag, get_flag, show, main
MAX_WORKERS = 20
def download_one(cc):
 image = get_flag(cc)
 show(cc)
 save_flag(image, cc.lower()+".gif")
 return cc
def download_many(cc_list):
 cc_list = cc_list[:5]
 with futures.ThreadPoolExecutor(max_workers=3) as executor:
 to_do = []
 for cc in sorted(cc_list):
  future = executor.submit(download_one,cc)
  to_do.append(future)
  msg = "Secheduled for {}:{}"
  print(msg.format(cc,future))
 results = []
 for future in futures.as_completed(to_do):
  res = future.result()
  msg = "{}result:{!r}"
  print(msg.format(future,res))
  results.append(res)
 return len(results)
if __name__ == '__main__':
 main(download_many)

The result is as follows:

Note: Python code cannot control GIL, standard All functions in the library that perform blocking IO operations will release the GIL when waiting for the operating system to return results and run other threads for execution. It is precisely because of this that Python threads can play a role in IO-intensive applications

The above are concurrent.futures to start the thread, the following is to start the process through it

concurrent.futures starts the process

ProcessPoolExecutor class in concurrent.futures Distribute work to multiple Python processes, so if you need to do CPU-intensive processing, using this module can bypass the GIL and utilize all CPU cores.

The principle is that a ProcessPoolExecutor creates N independent Python interpreters, where N is the number of CPU cores available on the system.

The usage method is the same as the ThreadPoolExecutor method

The above is the detailed content of Detailed example of how Python handles concurrency issues through futures. For more information, please follow other related articles on the PHP Chinese website!