Hướng dẫn how to run multiple subprocess python - cách chạy nhiều quy trình con python

Là một phần của nhóm hoạt động CNTT để tự động hóa các tác vụ cho các máy chủ khác nhau, đôi khi chúng tôi chống lại một số vấn đề trong việc thực hiện các quy trình ở quy mô lớn.

Các vấn đề khi chúng tôi thực hiện các nhiệm vụ lặp đi lặp lại

Một số quá trình mất rất nhiều thời gian thực hiện. Ví dụ: nếu chúng ta có 100 nhiệm vụ trong hàng đợi và mỗi nhiệm vụ mất khoảng 5 phút để hoàn thành, thì sẽ mất khoảng 500 phút để hoàn thành tất cả chúng.

Giải pháp là gì?

Một trong những giải pháp hiệu quả tôi tìm thấy là xử lý song song hoặc thực hiện các quy trình đồng thời. Điều này chỉ có thể được sử dụng nếu chúng ta có một máy chủ hoặc tài nguyên có sức mạnh tính toán cao.

Làm thế nào chúng ta có thể thực hiện điều này trong Python?

Trong Python, chúng tôi có một mô -đun để xử lý song song. Asyncio là một thư viện để viết mã đồng thời bằng cách sử dụng cú pháp Async/Await. Tìm tài liệu ở đây. Asyncio thường là một phù hợp hoàn hảo cho mã mạng có cấu trúc cấp cao và IO.concurrent code using the async/await syntax. Find the documentation here . Asyncio is often a perfect fit for IO-bound and high-level structured network code.

Asyncio cung cấp một tập hợp các API cấp cao cho:high-level APIs to:

• Chạy đồng thời Python Coroutines và có toàn quyền kiểm soát việc thực hiện của họ
• Thực hiện mạng IO và IPC
• Kiểm soát các quy trình con
• Phân phối các nhiệm vụ qua hàng đợi
• Đồng bộ hóa mã đồng thời

import asyncio
async def main():
    print('Hello ...')
    await asyncio.sleep(1)
    print('... World!')
# reference https://docs.python.org/3/library/asyncio.html
# Python 3.7+
asyncio.run(main())

Hoàn thành mã để đồng bộ hóa quá trình phụ

Note:
 Requires Python 3.6.
“””import sys
import time
import platform
import asyncio
from pprint import pprintasync def run_command(*args):
 “””Run command in subprocess.Example from:
 http://asyncio.readthedocs.io/en/latest/subprocess.html
 “””
 # Create subprocess
 process = await asyncio.create_subprocess_exec(
 *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
 )# Status
 print(“Started: %s, pid=%s” % (args, process.pid), flush=True)# Wait for the subprocess to finish
 stdout, stderr = await process.communicate()# Progress
 if process.returncode == 0:
 print(
 “Done: %s, pid=%s, result: %s”
 % (args, process.pid, stdout.decode().strip()),
 flush=True,
 )
 else:
 print(
 “Failed: %s, pid=%s, result: %s”
 % (args, process.pid, stderr.decode().strip()),
 flush=True,
 )# Result
 result = stdout.decode().strip()# Return stdout
 return resultasync def run_command_shell(command):
 “””Run command in subprocess (shell).Note:
 This can be used if you wish to execute e.g. “copy”
 on Windows, which can only be executed in the shell.
 “””
 # Create subprocess
 process = await asyncio.create_subprocess_shell(
 command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
 )# Status
 print(“Started:”, command, “(pid = “ + str(process.pid) + “)”, flush=True)# Wait for the subprocess to finish
 stdout, stderr = await process.communicate()# Progress
 if process.returncode == 0:
 print(“Done:”, command, “(pid = “ + str(process.pid) + “)”, flush=True)
 else:
 print(
 “Failed:”, command, “(pid = “ + str(process.pid) + “)”, flush=True
 )# Result
 result = stdout.decode().strip()# Return stdout
 return resultdef make_chunks(l, n):
 “””Yield successive n-sized chunks from l.Note:
 Taken from https://stackoverflow.com/a/312464
 “””
 if sys.version_info.major == 2:
 for i in xrange(0, len(l), n):
 yield l[i : i + n]
 else:
 # Assume Python 3
 for i in range(0, len(l), n):
 yield l[i : i + n]def run_asyncio_commands(tasks, max_concurrent_tasks=0):
 “””Run tasks asynchronously using asyncio and return results.If max_concurrent_tasks are set to 0, no limit is applied.
Note:
 By default, Windows uses SelectorEventLoop, which does not support
 subprocesses. Therefore ProactorEventLoop is used on Windows.
 https://docs.python.org/3/library/asyncio-eventloops.html#windows
 “””
 all_results = []if max_concurrent_tasks == 0:
 chunks = [tasks]
 num_chunks = len(chunks)
 else:
 chunks = make_chunks(l=tasks, n=max_concurrent_tasks)
 num_chunks = len(list(make_chunks(l=tasks, n=max_concurrent_tasks)))if asyncio.get_event_loop().is_closed():
 asyncio.set_event_loop(asyncio.new_event_loop())
 if platform.system() == “Windows”:
 asyncio.set_event_loop(asyncio.ProactorEventLoop())
 loop = asyncio.get_event_loop()chunk = 1
 for tasks_in_chunk in chunks:
 print(
 “Beginning work on chunk %s/%s” % (chunk, num_chunks), flush=True
 )
 commands = asyncio.gather(*tasks_in_chunk) # Unpack list using *
 results = loop.run_until_complete(commands)
 all_results += results
 print(
 “Completed work on chunk %s/%s” % (chunk, num_chunks), flush=True
 )
 chunk += 1loop.close()
 return all_resultsdef main():
 “””Main program.”””
 start = time.time()if platform.system() == “Windows”:
 # Commands to be executed on Windows
 commands = [[“hostname”]]
 else:
 # Commands to be executed on Unix
 commands = [[“du”, “-sh”, “/var/tmp”], [“hostname”]]tasks = []
 for command in commands:
 tasks.append(run_command(*command))# # Shell execution example
 # tasks = [run_command_shell(‘copy c:/somefile d:/new_file’)]# # List comprehension example
 # tasks = [
 # run_command(*command, get_project_path(project))
 # for project in accessible_projects(all_projects)
 # ]results = run_asyncio_commands(
 tasks, max_concurrent_tasks=20
 ) # At most 20 parallel tasks
 print(“Results:”)
 pprint(results)end = time.time()
 rounded_end = “{0:.4f}”.format(round(end — start, 4))
 print(“Script ran in about %s seconds” % (rounded_end), flush=True)if __name__ == “__main__”:
 main()