1. Lý do sử dụng các nhóm tiến trình và nhóm luồng
Việc khởi tạo tiến trình hay luồng đều tiêu tốn tài nguyên hệ thống, trong đó luồng thường ít tốn kém hơn tiến trình. Tuy nhiên, không thể tạo ra số lượng tiến trình/luồng vô hạn vì tài nguyên phần cứng máy tính có giới hạn. Mục tiêu là tối ưu hóa việc sử dụng phần cứng trong khi vẫn đảm bảo hệ thống hoạt động ổn định.
Các loại tác vụ:
- Tác vụ nặng về I/O (I/O-bound): Các tác vụ dành phần lớn thời gian để chờ đợi hoạt động nhập/xuất, ví dụ như đọc/ghi tập tin, giao tiếp mạng, truy vấn cơ sở dữ liệu. Với các tác vụ này, việc sử dụng luồng (threading) thường hiệu quả do Python GIL (Global Interpreter Lock) không ảnh hưởng khi luồng đang chờ I/O.
- Tác vụ nặng về tính toán (CPU-bound): Các tác vụ tiêu tốn nhiều chu kỳ CPU cho các phép toán số học và logic phức tạp. Trong trường hợp này, việc sử dụng tiến trình (multiprocessing) là cần thiết để tận dụng hết các lõi CPU, vượt qua giới hạn của GIL.
Khái niệm nhóm (Pool):
Một nhóm (pool) là một cơ chế quản lý một tập hợp các tiến trình hoặc luồng đã được khởi tạo sẵn. Khi có tác vụ mới, thay vì tạo một tiến trình/luồng mới, tác vụ sẽ được phân bổ cho một tài nguyên sẵn có trong nhóm. Điều này giúp:
- Giảm chi phí khởi tạo và hủy tài nguyên.
- Hạn chế số lượng tiến trình/luồng hoạt động đồng thời, bảo vệ tài nguyên hệ thống.
- Tăng hiệu quả tổng thể cho các tác vụ lặp đi lặp lại hoặc số lượng lớn.
Kể từ Python 3.2, thư viện concurrent.futures đã được giới thiệu, và Python 3.4 bổ sung asyncio vào thư viện chuẩn, với cú pháp async/await từ Python 3.5 trở đi (dành cho coroutines).
2. Sử dụng concurrent.futures
Thư viện chuẩn concurrent.futures (có sẵn từ Python 3.2+) cung cấp hai lớp chính: ThreadPoolExecutor (nhóm luồng) và ProcessPoolExecutor (nhóm tiến trình). Chúng cung cấp một tầng trừu tượng cao hơn so với việc sử dụng trực tiếp threading và multiprocessing, đơn giản hóa việc triển khai các nhóm luồng/tiến trình.
Các hàm chính để tương tác với nhóm:
pool.submit(func, *args, **kwargs): Gửi một tác vụ (hàmfuncvới các đối số) vào nhóm để thực thi. Hàm này trả về một đối tượngFuture.future.result(timeout=None): Lấy kết quả trả về từ tác vụ liên quan đến đối tượngFuture. Nếu tác vụ chưa hoàn thành, phương thức này sẽ chặn cho đến khi có kết quả.future.add_done_callback(fn): Gắn một hàm callback (fn) sẽ được gọi khi tác vụ hoàn thành. Hàm callback này sẽ nhận đối tượngFuturelàm đối số duy nhất.
2.1. Nhóm tiến trình với concurrent.futures.ProcessPoolExecutor
ProcessPoolExecutor cho phép thực thi các tác vụ song song trên các tiến trình riêng biệt, phù hợp cho các tác vụ nặng về tính toán. Mặc định, nó sử dụng số lượng tiến trình bằng với số lõi CPU của máy.
Ví dụ cơ bản:
import concurrent.futures
import os
import time
import random
def process_data_task(task_id):
"""Một tác vụ giả lập xử lý dữ liệu nặng."""
pid = os.getpid()
print(f"[{pid}] Bắt đầu xử lý tác vụ: {task_id}")
processing_time = random.randint(1, 3)
time.sleep(processing_time)
print(f"[{pid}] Hoàn thành xử lý tác vụ: {task_id} sau {processing_time}s")
return f"Kết quả từ {task_id}"
if __name__ == '__main__':
# Tạo một nhóm tiến trình với 3 tiến trình làm việc.
# Các tiến trình này được tạo một lần và tái sử dụng.
with concurrent.futures.ProcessPoolExecutor(max_workers=3) as process_pool:
print("--- Gửi 5 tác vụ vào nhóm tiến trình ---")
futures_list = []
for i in range(1, 6):
# submit() là một hoạt động bất đồng bộ, nó không chặn luồng chính.
future_obj = process_pool.submit(process_data_task, f"Job-{i}")
futures_list.append(future_obj)
# Chủ yếu để đợi các tác vụ hoàn thành (ví dụ thực tế sẽ dùng callback hoặc result() có điều kiện)
# Trong ví dụ này, chúng ta sẽ để nó chạy tự do cho đến khi main process kết thúc
# process_pool.shutdown() được gọi tự động khi thoát khỏi khối 'with'
print("Tất cả tác vụ đã được gửi. Luồng chính tiếp tục hoạt động.")
print("Chương trình chính đã kết thúc.")
2.2. Nhóm luồng với concurrent.futures.ThreadPoolExecutor
ThreadPoolExecutor cho phép thực thi các tác vụ đồng thời trên các luồng, phù hợp cho các tác vụ nặng về I/O. Tất cả các luồng trong một ThreadPoolExecutor sẽ có cùng PID với tiến trình chính.
Ví dụ cơ bản:
import concurrent.futures
import os
import time
import random
def fetch_web_data(url_name):
"""Một tác vụ giả lập tải dữ liệu từ web."""
pid = os.getpid()
thread_id = concurrent.futures.threading.current_thread().name # Lấy tên luồng
print(f"[{pid}/{thread_id}] Bắt đầu tải dữ liệu từ: {url_name}")
download_time = random.randint(1, 2)
time.sleep(download_time) # Giả lập chờ I/O
print(f"[{pid}/{thread_id}] Hoàn thành tải dữ liệu từ: {url_name} sau {download_time}s")
return f"Nội dung từ {url_name}"
if __name__ == '__main__':
# Tạo một nhóm luồng với 2 luồng làm việc.
# Mặc định thường là 5 lần số lõi CPU.
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_pool:
print("--- Gửi 4 tác vụ vào nhóm luồng ---")
for i in range(1, 5):
thread_pool.submit(fetch_web_data, f"site_{i}.com")
# thread_pool.shutdown() được gọi tự động khi thoát khỏi khối 'with'
print("Tất cả tác vụ đã được gửi. Tiến trình chính tiếp tục.")
print("Chương trình chính hoàn tất.")
2.3. Lấy kết quả từ tác vụ (đồng bộ và bất đồng bộ)
Lấy kết quả đồng bộ (chặn):
Gọi .result() trực tiếp trên đối tượng Future sẽ chặn luồng hiện tại cho đến khi tác vụ hoàn thành và trả về kết quả.
import concurrent.futures
import time
import random
def perform_calculation(value):
"""Thực hiện một phép tính và trả về bình phương của giá trị."""
print(f"Đang tính toán cho: {value}")
time.sleep(random.uniform(0.5, 1.5)) # Ngủ ngẫu nhiên
result_val = value * value
print(f"Hoàn thành tính toán cho: {value}, kết quả: {result_val}")
return result_val
if __name__ == '__main__':
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
print("--- Gửi tác vụ và lấy kết quả đồng bộ ---")
for num in range(1, 6):
future = executor.submit(perform_calculation, num)
# Gọi .result() ngay lập tức sẽ khiến vòng lặp trở nên đồng bộ.
# Mỗi lần submit, chương trình sẽ đợi cho đến khi tác vụ đó hoàn thành.
print(f"Kết quả đồng bộ cho {num}: {future.result()}")
print("Tất cả tính toán đã hoàn tất.")
Lấy kết quả bất đồng bộ (sử dụng callback):
Để tránh chặn luồng chính, chúng ta có thể sử dụng phương thức add_done_callback(). Callback sẽ được gọi khi tác vụ hoàn thành, nhận đối tượng Future làm đối số.
import concurrent.futures
import time
import random
def heavy_computation(number):
"""Một hàm giả lập tính toán nặng."""
print(f"Bắt đầu tính toán nặng cho: {number}")
time.sleep(random.randint(1, 3))
res = number ** 3
print(f"Hoàn thành tính toán nặng cho: {number}, kết quả: {res}")
return res
def process_final_result(future_obj):
"""Hàm callback để xử lý kết quả khi tác vụ hoàn thành."""
try:
final_data = future_obj.result() # Lấy kết quả từ Future
print(f"*** Callback nhận được kết quả: {final_data} ***")
except Exception as exc:
print(f"*** Callback nhận được lỗi: {exc} ***")
if __name__ == '__main__':
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
print("--- Gửi tác vụ và xử lý kết quả bất đồng bộ bằng callback ---")
for item in range(10, 15):
# Gắn callback ngay sau khi gửi tác vụ
executor.submit(heavy_computation, item).add_done_callback(process_final_result)
# Luồng chính tiếp tục chạy mà không bị chặn
print("Tất cả tác vụ đã được gửi. Tiến trình chính đang chờ các callback.")
print("Chương trình chính đã thoát khỏi nhóm. Các callback có thể vẫn đang chạy.")
3. Sử dụng multiprocessing.Pool
Module multiprocessing cung cấp lớp Pool, cho phép tạo một nhóm các tiến trình để phân phối công việc. Nó có các phương thức khác nhau để gửi tác vụ, cung cấp các lựa chọn linh hoạt cho việc xử lý song song.
3.1. Các phương thức gửi tác vụ trong multiprocessing.Pool
3.1.1. apply_async() (bất đồng bộ, không chặn)
Phương thức này gửi một tác vụ duy nhất vào nhóm và trả về một đối tượng AsyncResult ngay lập tức. Luồng gọi không bị chặn. Nó hỗ trợ nhiều đối số và có thể chỉ định hàm callback và error callback.
func: Hàm cần thực thi.args: Tuple các đối số vị trí chofunc.kwds: Dictionary các đối số từ khóa chofunc.callback: Hàm sẽ được gọi với kết quả củafunckhi tác vụ hoàn thành. Chỉ nhận một đối số.error_callback: Hàm sẽ được gọi nếufuncphát sinh ngoại lệ. Chỉ nhận một đối số (ngoại lệ).
import multiprocessing
import time
import os
def process_file_segment(segment_id):
"""Tác vụ xử lý một phân đoạn tập tin."""
print(f"[{os.getpid()}] Bắt đầu xử lý phân đoạn: {segment_id}")
time.sleep(1.2)
print(f"[{os.getpid()}] Hoàn thành phân đoạn: {segment_id}")
return f"Phân đoạn {segment_id} đã xử lý."
def after_task_completion(result_str):
"""Hàm callback, chạy trong tiến trình chính."""
time.sleep(0.5) # Giả lập công việc nhẹ
print(f"[{os.getpid()}] Callback nhận kết quả: '{result_str}'")
if __name__ == '__main__':
# Tạo nhóm tiến trình với 3 tiến trình
with multiprocessing.Pool(processes=3) as process_pool:
print("--- Gửi tác vụ với apply_async ---")
async_results = []
for j in range(1, 6):
res = process_pool.apply_async(
func=process_file_segment,
args=(f"segment-{j}",),
callback=after_task_completion
)
async_results.append(res)
# Đóng nhóm và chờ tất cả tác vụ hoàn thành
process_pool.close()
process_pool.join()
print("Tất cả các tác vụ và callback đã hoàn tất.")
3.1.2. map_async() (bất đồng bộ, không chặn)
map_async() tương tự như hàm map() tích hợp sẵn nhưng thực hiện bất đồng bộ. Nó nhận một hàm và một iterable, áp dụng hàm đó cho từng phần tử của iterable. Kết quả được trả về dưới dạng một danh sách (list) sau khi tất cả các tác vụ hoàn thành. Chỉ chấp nhận một đối số cho hàm tác vụ.
func: Hàm cần thực thi (chỉ nhận một đối số).iterable: Đối tượng có thể lặp.chunksize: Số lượng phần tử sẽ được gửi trong một "khối" đến mỗi tiến trình con.callback,error_callback: Tương tự nhưapply_async, nhưngcallbacknhận danh sách kết quả.
import multiprocessing
import time
import os
def calculate_complex_value(value_input):
"""Một tác vụ tính toán giá trị phức tạp."""
pid = os.getpid()
print(f"[{pid}] Đang tính toán cho giá trị: {value_input}")
time.sleep(random.uniform(0.8, 1.8))
result_val = value_input * 10 + 5
print(f"[{pid}] Hoàn thành tính toán cho giá trị: {value_input}")
return result_val
def handle_map_results(final_results_list):
"""Callback nhận danh sách kết quả."""
time.sleep(0.3)
print(f"[{os.getpid()}] Callback nhận được TẤT CẢ kết quả: {final_results_list}")
if __name__ == '__main__':
with multiprocessing.Pool(processes=3) as process_pool:
print("--- Gửi tác vụ với map_async ---")
# Gửi một iterable các giá trị, mỗi giá trị sẽ được xử lý bởi calculate_complex_value
process_pool.map_async(
func=calculate_complex_value,
iterable=[10, 20, 30, 40, 50],
callback=handle_map_results
)
process_pool.close()
process_pool.join()
print("Chương trình chính hoàn tất sau map_async.")
3.1.3. starmap_async() (bất đồng bộ, không chặn)
starmap_async() tương tự như map_async() nhưng được thiết kế để áp dụng một hàm cho các đối số được cung cấp dưới dạng các tuple trong một iterable. Điều này cho phép hàm tác vụ nhận nhiều đối số.
import multiprocessing
import time
import os
def process_coordinates(x, y, label):
"""Tác vụ xử lý tọa độ."""
print(f"[{os.getpid()}] Xử lý {label}: ({x}, {y})")
time.sleep(0.7)
return f"Kết quả từ {label}: {x*y}"
if __name__ == '__main__':
with multiprocessing.Pool(processes=2) as process_pool:
print("--- Gửi tác vụ với starmap_async ---")
coordinates_data = [
(10, 20, "Point A"),
(5, 15, "Point B"),
(8, 12, "Point C")
]
# Mỗi tuple trong coordinates_data sẽ được "unpack" thành các đối số cho process_coordinates
process_pool.starmap_async(func=process_coordinates, iterable=coordinates_data)
process_pool.close()
process_pool.join()
print("Chương trình chính hoàn tất sau starmap_async.")
3.1.4. imap() (có thứ tự)
imap() trả về một iterator, cho phép chúng ta nhận kết quả ngay khi chúng có sẵn, nhưng theo thứ tự ban đầu của các phần tử trong iterable đầu vào. Điều này có nghĩa là nếu tác vụ cho phần tử thứ 2 hoàn thành trước tác vụ cho phần tử thứ nhất, bạn vẫn phải chờ kết quả của phần tử thứ nhất trước khi nhận được kết quả của phần tử thứ 2.
import multiprocessing
import time
import random
import os
def sequenced_task(item_idx):
"""Tác vụ có thể mất thời gian ngẫu nhiên."""
print(f"[{os.getpid()}] Bắt đầu tác vụ {item_idx}")
time.sleep(random.uniform(0.5, 2.0)) # Thời gian ngẫu nhiên
print(f"[{os.getpid()}] Hoàn thành tác vụ {item_idx}")
return f"Dữ liệu đã xử lý cho {item_idx}"
if __name__ == '__main__':
with multiprocessing.Pool(processes=3) as process_pool:
print("--- Gửi tác vụ với imap và lấy kết quả có thứ tự ---")
# Kết quả sẽ được nhận theo thứ tự 1, 2, 3, ... dù tác vụ nào hoàn thành trước.
for result_item in process_pool.imap(func=sequenced_task, iterable=range(1, 6)):
print(f"Nhận được kết quả (theo thứ tự): {result_item}")
process_pool.close()
process_pool.join()
print("Chương trình chính hoàn tất sau imap.")
3.1.5. imap_unordered() (không thứ tự)
imap_unordered() cũng trả về một iterator, nhưng các kết quả được trả về ngay khi chúng hoàn thành, không quan tâm đến thứ tự ban đầu trong iterable. Điều này có thể nhanh hơn imap() nếu thứ tự không quan trọng.
import multiprocessing
import time
import random
import os
def unordered_task(task_key):
"""Tác vụ với thời gian hoàn thành ngẫu nhiên."""
print(f"[{os.getpid()}] Khởi động tác vụ {task_key}")
time.sleep(random.uniform(0.5, 2.0))
print(f"[{os.getpid()}] Kết thúc tác vụ {task_key}")
return f"Dữ liệu hoàn tất cho {task_key}"
if __name__ == '__main__':
with multiprocessing.Pool(processes=3) as process_pool:
print("--- Gửi tác vụ với imap_unordered và lấy kết quả không thứ tự ---")
# Kết quả sẽ được nhận ngay khi tác vụ hoàn thành, không cần theo thứ tự đầu vào.
for result_unordered in process_pool.imap_unordered(func=unordered_task, iterable=range(1, 6)):
print(f"Nhận được kết quả (không thứ tự): {result_unordered}")
process_pool.close()
process_pool.join()
print("Chương trình chính hoàn tất sau imap_unordered.")
3.1.6. So sánh các phương thức
apply_async()vsmap_async():apply_async(): Gửi từng tác vụ riêng lẻ, mỗi tác vụ có thể nhận nhiều đối số. Kết quả được xử lý riêng lẻ thông qua callback. Nếu một tác vụ lỗi, các tác vụ khác không bị ảnh hưởng.map_async(): Gửi một iterable các phần tử, mỗi phần tử được áp dụng cho một hàm nhận một đối số duy nhất. Callback nhận một danh sách kết quả tổng hợp. Nếu một tác vụ trong batch lỗi, toàn bộ batch có thể bị lỗi.
map_async()vsimap()vàimap_unordered():map_async(): Chờ tất cả các tác vụ hoàn thành rồi mới trả về một danh sách kết quả duy nhất.imap()vàimap_unordered(): Trả về một iterator, cho phép xử lý kết quả ngay khi chúng có sẵn.imap(): Đảm bảo thứ tự kết quả giống với thứ tự đầu vào.imap_unordered(): Trả về kết quả theo thứ tự hoàn thành của chúng, không theo thứ tự đầu vào.
starmap_async()vsmap_async():starmap_async(): Cho phép hàm tác vụ nhận nhiều đối số thông qua việc "unpack" các tuple từ iterable.map_async(): Chỉ cho phép hàm tác vụ nhận một đối số duy nhất.
3.2. Xử lý lỗi với error_callback
Trong các phương thức _async (như apply_async, map_async, starmap_async), bạn có thể chỉ định một hàm error_callback. Hàm này sẽ được gọi nếu tác vụ con phát sinh một ngoại lệ chưa được xử lý.
3.2.1. apply_async với error_callback
Nếu một tác vụ con trong apply_async phát sinh lỗi, error_callback sẽ được gọi với đối tượng ngoại lệ làm đối số.
import multiprocessing
import time
import os
def check_data_integrity(data_id):
"""Tác vụ kiểm tra tính toàn vẹn, có thể gây lỗi."""
pid = os.getpid()
print(f"[{pid}] Đang kiểm tra dữ liệu ID: {data_id}")
time.sleep(0.8)
if data_id == 3:
raise ValueError(f"Lỗi dữ liệu nghiêm trọng cho ID {data_id}!")
print(f"[{pid}] Kiểm tra thành công ID: {data_id}")
return f"ID {data_id} OK"
def on_success_log(result):
"""Callback khi tác vụ thành công."""
print(f"[{os.getpid()}] Callback thành công: {result}")
def on_error_notify(exc):
"""Callback khi tác vụ gặp lỗi."""
print(f"[{os.getpid()}] Callback lỗi: Phát hiện ngoại lệ: {type(exc).__name__} - {exc}")
if __name__ == '__main__':
with multiprocessing.Pool(processes=2) as process_pool:
print("--- Kiểm tra apply_async với error_callback ---")
for k in range(1, 5):
process_pool.apply_async(
func=check_data_integrity,
args=(k,),
callback=on_success_log,
error_callback=on_error_notify
)
process_pool.close()
process_pool.join()
print("Chương trình chính đã hoàn tất xử lý lỗi.")
Lưu ý: Nếu bạn bắt ngoại lệ bên trong hàm tác vụ (sử dụng try...except) và trả về một giá trị (hoặc None), thì error_callback sẽ không được gọi. Thay vào đó, callback thông thường sẽ nhận kết quả đó.
3.2.2. map_async với error_callback
Trong trường hợp map_async, nếu bất kỳ tác vụ nào trong iterable phát sinh lỗi không được xử lý, error_callback sẽ được gọi. Tuy nhiên, toàn bộ nhóm kết quả (list) sẽ không được truyền tới callback thông thường.
import multiprocessing
import time
import os
def data_processing_unit(val):
"""Một đơn vị xử lý dữ liệu có thể thất bại."""
pid = os.getpid()
print(f"[{pid}] Đang xử lý giá trị: {val}")
time.sleep(0.5)
if val % 4 == 0: # Ví dụ, gây lỗi nếu giá trị chia hết cho 4
raise TypeError(f"Lỗi kiểu dữ liệu cho giá trị {val}!")
return val * 2
def collect_all_outputs(results_list):
"""Callback khi tất cả tác vụ map_async thành công."""
print(f"[{os.getpid()}] Tất cả kết quả map_async: {results_list}")
def report_map_error(exc_type):
"""Callback khi map_async gặp lỗi."""
print(f"[{os.getpid()}] Báo cáo lỗi map_async: {exc_type}")
if __name__ == '__main__':
with multiprocessing.Pool(processes=2) as process_pool:
print("--- Kiểm tra map_async với error_callback ---")
process_pool.map_async(
func=data_processing_unit,
iterable=[1, 2, 3, 4, 5, 6], # 4 sẽ gây lỗi
callback=collect_all_outputs,
error_callback=report_map_error
)
process_pool.close()
process_pool.join()
print("Chương trình chính đã hoàn tất map_async (có thể có lỗi).")
3.3. Lấy kết quả từ AsyncResult
Đối tượng AsyncResult (được trả về bởi apply_async, map_async, v.v.) có các phương thức hữu ích để kiểm tra trạng thái và lấy kết quả.
ready(): Trả vềTruenếu tác vụ đã hoàn thành (thành công hoặc thất bại).successful(): Trả vềTruenếu tác vụ đã hoàn thành mà không có ngoại lệ. Chỉ có ý nghĩa nếuready()làTrue.get(timeout=None): Trả về kết quả của tác vụ. Nếu tác vụ chưa hoàn thành, nó sẽ chặn cho đến khi có kết quả. Nếu tác vụ phát sinh ngoại lệ,get()sẽ ném lại ngoại lệ đó.
Ví dụ: Lấy kết quả và xử lý lỗi
import multiprocessing
import os
import time
def process_complex_calculation(task_num):
"""Tác vụ tính toán phức tạp, có thể phát sinh lỗi."""
pid = os.getpid()
print(f"[{pid}] Bắt đầu tính toán cho tác vụ: {task_num}")
time.sleep(1)
if task_num == 3:
raise ZeroDivisionError(f"Không thể xử lý tác vụ {task_num}: Lỗi chia cho 0 giả định!")
return f"Tính toán cho {task_num} hoàn tất."
if __name__ == '__main__':
with multiprocessing.Pool(processes=2) as process_pool:
print("--- Lấy kết quả từ AsyncResult với xử lý lỗi ---")
futures_tasks = []
for i_task in range(1, 5):
futures_tasks.append(process_pool.apply_async(func=process_complex_calculation, args=(i_task,)))
process_pool.close()
process_pool.join()
print("\n=== Đánh giá kết quả ===")
for future_res in futures_tasks:
print(f"Tác vụ đã sẵn sàng? {future_res.ready()}")
if future_res.ready():
try:
result_data = future_res.get()
print(f" Thành công: {future_res.successful()}, Kết quả: {result_data}")
except Exception as e:
print(f" Thành công: {future_res.successful()}, Lỗi: {type(e).__name__} - {e}")
else:
print(" Tác vụ chưa hoàn thành (không nên xảy ra sau join).")
print("Chương trình chính hoàn tất kiểm tra AsyncResult.")
Lưu ý về Callback và get(): Hàm get() trên đối tượng AsyncResult sẽ trả về kết quả gốc từ hàm tác vụ ban đầu, không phải là giá trị trả về từ bất kỳ hàm callback nào mà bạn đã gắn vào. Callback chỉ đơn thuần là một hành động phụ trợ được kích hoạt khi tác vụ hoàn thành.
import multiprocessing
import os
def original_data_generator(idx):
"""Hàm tạo dữ liệu gốc."""
print(f"[{os.getpid()}] Tạo dữ liệu gốc cho {idx}")
return f"Dữ liệu gốc {idx}"
def callback_modifier(raw_data):
"""Hàm callback sửa đổi dữ liệu."""
modified = f"[{os.getpid()}] (Callback) Đã chỉnh sửa: {raw_data}"
print(modified)
return modified # Giá trị trả về này KHÔNG ảnh hưởng đến .get()
if __name__ == '__main__':
with multiprocessing.Pool(processes=1) as process_pool:
print("--- Minh họa get() và Callback ---")
# Callback được gắn, nhưng get() vẫn trả về kết quả gốc
future_obj = process_pool.apply_async(func=original_data_generator, args=(1,), callback=callback_modifier)
process_pool.close()
process_pool.join()
print(f"Kết quả từ .get(): {future_obj.get()}")
print("Minh họa đã kết thúc.")
3.4. Nhóm luồng với multiprocessing.pool.ThreadPool
multiprocessing.pool.ThreadPool cung cấp giao diện tương tự như multiprocessing.Pool nhưng sử dụng các luồng thay vì tiến trình. Nó cũng có các phương thức apply_async, map_async, v.v., hoạt động theo cách tương tự.
import multiprocessing.pool
import os
import time
import random
def network_request_task(request_id):
"""Một tác vụ giả lập yêu cầu mạng."""
pid = os.getpid()
thread_name = multiprocessing.current_process().name # Dù là ThreadPool, nó vẫn là một process name
print(f"[{pid}/{thread_name}] Bắt đầu yêu cầu: {request_id}")
time.sleep(random.uniform(0.5, 1.5))
print(f"[{pid}/{thread_name}] Hoàn thành yêu cầu: {request_id}")
return f"Phản hồi từ {request_id}"
def log_network_response(response_data):
"""Callback ghi log phản hồi."""
print(f"[{os.getpid()}] (Callback Thread) Phản hồi đã nhận: {response_data}")
if __name__ == '__main__':
# Tạo một nhóm luồng từ multiprocessing.pool
with multiprocessing.pool.ThreadPool(processes=4) as thread_pool_mp:
print("--- Sử dụng multiprocessing.pool.ThreadPool ---")
mp_thread_futures = []
for req_idx in range(1, 7):
mp_thread_futures.append(
thread_pool_mp.apply_async(
func=network_request_task,
args=(f"Request-{req_idx}",),
callback=log_network_response
)
)
thread_pool_mp.close()
thread_pool_mp.join()
print("\n=== Kiểm tra kết quả ThreadPool ===")
for future_mp_thread in mp_thread_futures:
try:
print(f" Kết quả cuối cùng: {future_mp_thread.get()}")
except Exception as e:
print(f" Lỗi khi lấy kết quả: {e}")
print("Chương trình chính hoàn tất với ThreadPool của multiprocessing.")
4. So sánh concurrent.futures và multiprocessing.Pool
Cả hai module đều cung cấp các công cụ mạnh mẽ để thực hiện các tác vụ song song/đồng thời, nhưng có những khác biệt chính:
- Mức độ trừu tượng:
concurrent.futurescung cấp một API cấp cao hơn, đơn giản hơn để sử dụng. Nó thống nhất giao diện cho cả nhóm luồng và nhóm tiến trình. - Tính linh hoạt của
multiprocessing.Pool: Module này, đặc biệt là lớpPool, cung cấp nhiều phương thức gửi tác vụ (apply_async,map_async,starmap_async,imap,imap_unordered) với các đặc điểm khác nhau về xử lý thứ tự và đối số. Điều này mang lại sự kiểm soát chi tiết hơn trong một số trường hợp. concurrent.futureslà lựa chọn hiện đại: Đối với hầu hết các trường hợp sử dụng,concurrent.futureslà lựa chọn được khuyến nghị do tính dễ sử dụng và cú pháp nhất quán. Nó là một sự "bao bọc" hiệu quả cho các tính năng cốt lõi củathreadingvàmultiprocessing.