تسريع تطبيقات بايثون باستخدام multiprocessing و multithreading

إذا بدأت مشاريعك في بايثون تكبر وتزداد العمليات الحسابية أو عمليات الإدخال والإخراج (قراءة الملفات، التعامل مع الشبكة، قواعد البيانات…) فغالبًا ستصل لنقطة تشعر فيها أن الأداء أصبح عنق زجاجة. هنا يظهر دور multiprocessing multithreading Python كأهم أدوات الاستفادة من تعدد الأنوية في المعالج، وتحسين استغلال وقت التنفيذ بشكل ذكي.

في هذا المقال سنتعرف على:

• مفهوم تعدد الخيوط (Multithreading) وتعدد العمليات (Multiprocessing) في بايثون.
• متى تستخدم الخيوط ومتى تحتاج إلى العمليات.
• تأثير الـ GIL على الأداء.
• أمثلة عملية مبسطة توضح الفكرة.
• مشكلات شائعة وكيفية تجنبها.

إذا كنت مهتمًا أكثر بجانب البرمجة غير المتزامنة (async/await) في بايثون، يمكنك الإطلاع أيضًا على مقالنا: البرمجة غير المتزامنة في بايثون: تحسين الأداء باستخدام async و await، لأنه يكمل الصورة الكاملة لأدوات تحسين الأداء في بايثون.

لماذا بايثون بطيئة أحيانًا؟ فهم GIL أولًا

قبل الحديث عن multiprocessing multithreading Python، يجب فهم نقطة مهمة في تنفيذ بايثون وهي ما يُسمى بـ:

Global Interpreter Lock (GIL)

الـ GIL هو "قفل" على مستوى مفسر بايثون، يضمن أن خيطًا (Thread) واحدًا فقط هو من ينفذ كود بايثون في لحظة معينة داخل عملية واحدة، حتى لو كان لديك أكثر من نواة في المعالج.

النتيجة:

• الخيوط في بايثون لا تستطيع تنفيذ كود بايثون CPU-bound (يتطلب حسابات مكثفة) بالتوازي الحقيقي داخل نفس العملية.
• لكنها مفيدة جدًا عندما يكون الكود I/O-bound مثل طلبات HTTP، التعامل مع الملفات، قواعد البيانات… لأن الخيط ينتظر الـ I/O بينما خيط آخر يمكنه العمل.

من هنا يأتي الفارق الجوهري:

• Multithreading: أفضل غالبًا مع المهام I/O-bound.
• Multiprocessing: أفضل مع المهام CPU-bound (حسابات كثيفة، معالجة صور، عمليات علمية…)، لأنه يتجاوز قيود الـ GIL باستخدام عدة عمليات مستقلة.

متى تستخدم multithreading في بايثون؟

استخدام الخيوط (Threads) مناسب عندما يكون عنق الزجاجة في انتظار الإدخال/الإخراج وليس في الحساب نفسه. أمثلة:

• إرسال عدد كبير من الطلبات إلى API خارجي.
• تحميل ملفات من الإنترنت بالتوازي.
• قراءة أو كتابة عدد كبير من الملفات الصغيرة.
• عمليات تفاعل مع قواعد البيانات عبر الشبكة.

في هذه السيناريوهات، الخيط يقضي وقتًا طويلاً في الانتظار، وليس في استهلاك الـ CPU، فيمكن تشغيل عدة خيوط في نفس الوقت للاستفادة من الوقت الضائع.

مثال بسيط على multithreading في بايثون

المثال التالي يوضح استخدام threading لتنفيذ مهام I/O-bound (محاكاة عبر time.sleep):

import threading
import time

def download_file(file_id):
    print(f"بدء تحميل الملف {file_id}")
    time.sleep(2)  # محاكاة عملية تحميل تستغرق وقت
    print(f"انتهاء تحميل الملف {file_id}")

threads = []

start = time.time()

for i in range(5):
    t = threading.Thread(target=download_file, args=(i,))
    t.start()
    threads.append(t)

for t in threads:
    t.join()

end = time.time()
print(f"الوقت الكلي: {end - start:.2f} ثانية")

في هذا المثال:

• نطلق 5 خيوط لتحميل 5 "ملفات" بالتوازي.
• بدلًا من تنفيذ 5 مرات متتالية لـ time.sleep(2) (أي ~10 ثوانٍ)، سيتم الانتهاء تقريبًا في حوالي 2–3 ثوانٍ فقط، لأن الانتظار يتم بالتوازي.

متى تستخدم multiprocessing في بايثون؟

إذا كان تطبيقك يقوم بعمليات حسابية معقدة (معالجة صور، تحليل بيانات ضخمة، خوارزميات ذكاء اصطناعي، تشفير، ضغط ملفات…) فغالبًا هو CPU-bound. عندها:

الخيوط لن تعطيك تسريعًا حقيقيًا بسبب الـ GIL.

هنا يأتي دور multiprocessing حيث يقوم بإنشاء عمليات (Processes) مستقلة بدلاً من الخيوط. كل عملية تمتلك:

• مفسر بايثون خاص بها بدون مشاركة الـ GIL مع العمليات الأخرى.
• مساحة ذاكرة مستقلة.

وبالتالي يمكن استغلال تعدد الأنوية بالكامل لتنفيذ كود بايثون المتعلق بالحسابات في توازي حقيقي.

مثال بسيط على multiprocessing في بايثون

لنفرض أن لدينا دالة تقوم بحساب عملية ثقيلة على CPU (محاكاة عن طريق حلقة كبيرة):

from multiprocessing import Process, cpu_count
import time

def cpu_heavy_task(n):
    total = 0
    for i in range(10_000_000):
        total += i * n
    return total

def run_in_process(n):
    start = time.time()
    result = cpu_heavy_task(n)
    end = time.time()
    print(f"العملية {n} انتهت في {end - start:.2f} ثانية")

if __name__ == "__main__":
    processes = []
    start_all = time.time()

    for i in range(4):  # تشغيل 4 عمليات
        p = Process(target=run_in_process, args=(i+1,))
        p.start()
        processes.append(p)

    for p in processes:
        p.join()

    end_all = time.time()
    print(f"الوقت الكلي: {end_all - start_all:.2f} ثانية")

النقاط المهمة هنا:

• نستخدم if __name__ == "__main__" وهو ضروري عند استخدام multiprocessing خصوصًا على ويندوز حتى لا يحدث استدعاء ذاتي لا نهائي للسكريبت.
• كل Process يعمل في نواة (Core) مختلفة إن أمكن، وبالتالي يتم توزيع الحمل على الأنوية المتاحة.

في كثير من الأحيان، تشغيل 4 عمليات على معالج رباعي الأنوية يمكن أن يقلل وقت التنفيذ إلى ما يقارب ربع الوقت (مع اعتبار بعض تكلفة إنشاء العمليات).

استخدام Pool في multiprocessing لتبسيط الكود

بدلًا من إنشاء العمليات يدويًا، يمكن استخدام multiprocessing.Pool لتوزيع قائمة من المهام على مجموعة من العمليات بسهولة.

from multiprocessing import Pool, cpu_count
import time

def cpu_heavy_task(n):
    total = 0
    for i in range(8_000_000):
        total += i * n
    return total

if __name__ == "__main__":
    numbers = [1, 2, 3, 4, 5, 6, 7, 8]
    start = time.time()

    with Pool(processes=cpu_count()) as pool:
        results = pool.map(cpu_heavy_task, numbers)

    end = time.time()
    print("النتائج:", results)
    print(f"الوقت الكلي: {end - start:.2f} ثانية")

المزايا:

• pool.map توزع العناصر في numbers على العمليات المتاحة تلقائيًا.
• cpu_count() تستخدم عدد الأنوية المتوفرة على الجهاز.

مقارنة سريعة: Multithreading vs Multiprocessing في Python

مشكلات شائعة مع multithreading في بايثون

1. ظروف السباق (Race Conditions)

عندما تحاول عدة خيوط تعديل نفس المتغير أو نفس البنية في نفس اللحظة، قد تحصل نتائج غير متوقعة.

import threading

counter = 0

def increment():
    global counter
    for _ in range(100000):
        counter += 1

threads = []

for _ in range(5):
    t = threading.Thread(target=increment)
    t.start()
    threads.append(t)

for t in threads:
    t.join()

print("القيمة النهائية:", counter)

توقع أن تكون القيمة 500000، لكن غالبًا ستحصل على رقم أقل بسبب تداخل التحديثات.

الحل: استخدام أقفال (Locks).

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

# بقية الكود كما هو

2. الـ Deadlock

يحدث عندما تنتظر خيوط أقفالًا يحتفظ بها خيوط أخرى، ولا يتحرك أي منها. لتجنب ذلك:

• قلل استخدام الأقفال قدر الإمكان.
• حافظ على ترتيب ثابت لاكتساب الأقفال.
• استخدم تراكيب بيانات آمنة من الخيوط مثل queue.Queue عندما تحتاج إلى تبادل البيانات.

مشكلات شائعة مع multiprocessing في بايثون

1. صعوبة مشاركة البيانات

كل عملية لها ذاكرة مستقلة، فلا يمكنك ببساطة مشاركة قائمة أو قاموس بين العمليات كما في الخيوط. الأدوات المتاحة:

• multiprocessing.Queue: لتمرير الرسائل بين العمليات.
• multiprocessing.Value و multiprocessing.Array: لمشاركة قيم بسيطة.
• Manager: لتوفير كائنات مشتركة مثل list و dict.

from multiprocessing import Process, Manager

def worker(shared_list):
    shared_list.append("عنصر جديد")

if __name__ == "__main__":
    with Manager() as manager:
        lst = manager.list()
        processes = []

        for _ in range(5):
            p = Process(target=worker, args=(lst,))
            p.start()
            processes.append(p)

        for p in processes:
            p.join()

        print("المحتوى:", list(lst))

2. مشكلة pickling للكائنات

عند تمرير دوال أو كائنات بين العمليات، يعتمد multiprocessing على آلية pickle. لذلك:

• الدوال يجب أن تكون معرفة في المستوى الأعلى (top-level) وليس داخل دالة أخرى أو داخل كلاس (في بعض الأنظمة).
• بعض الكائنات غير قابلة للـ pickling، مما يسبب أخطاء عند محاولة تمريرها إلى العمليات.

3. تكلفة الإنشاء العالية

إنشاء عملية جديدة مكلف مقارنة بإنشاء خيط جديد، لذلك:

• لا تستخدم عددًا ضخمًا من العمليات لمهام صغيرة جدًا.
• حاول تجميع المهام أو استخدام Pool لإعادة استخدام مجموعة محدودة من العمليات.

اختيار الأداة المناسبة في multiprocessing multithreading Python

باختصار:

• إذا كانت مشكلتك I/O-bound: استخدم threading أو فكر في البرمجة غير المتزامنة async/await (خاصة في تطبيقات الويب أو التعامل الكثيف مع الشبكة). يمكنك قراءة المزيد عن ذلك في مقال: البرمجة غير المتزامنة في بايثون.
• إذا كانت مشكلتك CPU-bound: استخدم multiprocessing أو استعن بمكتبات تعتمد على C/NumPy والتي تستفيد من الأنوية داخليًا، أو فكر في نقل الجزء الثقيل إلى لغة أخرى أسرع مثل C++ أو C# (راجع مقالتنا مقارنة بين بايثون وسي شارب).

نصائح عملية لتحسين الأداء في مشاريع بايثون

• ابدأ أولًا بقياس الأداء (profiling) قبل محاولة التسريع، حتى تعرف عنق الزجاجة الحقيقي.
• لا تستخدم multithreading لمهام CPU-bound متوقعة؛ لن تحصل على تسريع حقيقي بسبب الـ GIL.
• جرّب concurrent.futures كواجهة أبسط لـ ThreadPool و ProcessPool.
• احرص على كتابة كود قابل للاختبار، واستخدم أدوات مثل pytest لاختبار الوحدات في بايثون لضمان عدم كسر منطق البرنامج عند إضافة التوازي.
• راقب الأخطاء والمشاكل في بيئة الإنتاج؛ أدوات مثل Sentry لمراقبة وتتبع أخطاء بايثون تفيد كثيرًا عندما تبدأ في استخدام الخيوط والعمليات.

خلاصة

تسريع تطبيقات بايثون لا يعني دائمًا تغيير اللغة أو إعادة كتابة المشروع من الصفر. في كثير من الحالات، يمكن لتقنيات multiprocessing multithreading Python أن تمنحك دفعة كبيرة في الأداء إذا استخدمت في المكان الصحيح:

• multithreading للمهام I/O-bound مع انتباه لمشاكل الـ race conditions.
• multiprocessing للمهام CPU-bound مع الأخذ في الاعتبار تكلفة إنشاء العمليات وصعوبة مشاركة البيانات.

ابدأ بتحديد نوع حمل تطبيقك، ثم اختر الأداة المناسبة، وطبّقها على أجزاء صغيرة، مع اختبار دقيق وقياس مستمر للأداء، حتى تصل للتوازن الأفضل بين السرعة وتعقيد الكود.