كيفية إنشاء عدة خيوط وفهم التزامن في جافا

يصف هذا المقال كيفية تحقيق برمجة متزامنة في جافا، إذ يُغطي مبادئ البرمجة المتوازية والثبات والخيوط وإطار العمل التنفيذي (تجمُعات الخيوط thread pools)، إلى جانب واجهات Futures وCompletableFuture القابلة للاستدعاء وإطار عمل fork-join.

التزامن

يعرف التزامن بأنه القدرة على تشغيل عدة برامجٍ أو عدة أجزاءٍ من البرنامج على التوازي، فعندما نتمكن من إجراء مهمةٍ تستغرق وقتًا بالتزامن أو على التوازي مع مهمةٍ أخرى، فسيرفع هذا من الإنتاجية وتفاعلية البرنامج.

يمتلك أي حاسوبٍ حديثٍ عدة وحدات معالجةٍ مركزية CPU أو عدة نوىً ضمن الوحدة الواحدة، ومن الممكن أن تكون القدرة على زيادة هذه النوى المتعددة هي مفتاح النجاح لتطبيقٍ يمتلك حجمًا ضخمًا من المُستخدمين.

العملية Process مقابل الخيوط Threads

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

يُدعى الخيط بالعملية البسيطة ويمتلك مكدسًا خاصًا به، ومع ذلك يستطيع الوصول للبيانات المشتركة للخيوط الأخرى ضمن نفس العملية، كما يمتلك كل خيطٍ ذاكرة تخزينٍ مؤقتٍ خاصة، وعندما يقرأ الخيط بيانات مشتركة يحفظها في ذاكرته المؤقتة، بحيث يُمكن للخيط إعادة قراءة البيانات المشتركة.

يعمل تطبيق جافا افتراضيًا ضمن عمليةٍ واحدةٍ، ولتحقيق عملية المعالجة المتوازية أو السلوك غير المتزامن؛ فيجب أن يعمل تطبيق الجافا ضمن عدة خيوط.

التحسينات والمشاكل المرافقة للتزامن

حدود مكاسب التزامن

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

يُمكن حساب الربح النظري باستخدام القاعدة التالية المُشار إليها بقانون Amdahl.

إذا كانت "F" هي النسبة المئوية لأجزاء البرنامج التي يُمكن أن تعمل على التوازي و"N" هو عدد العمليات، فإن ربح الأداء الأعظمي هو:

 1/(F+((1-F)/N))

مشاكل التزامن

تمتلك الخيوط مكدساتٍ خاصة بها، إلا أنها لا تستطيع الوصول إلى البيانات المُشاركة، ولذلك توجد مشكلتان مرتبطتان بالوصول ومجال الرؤية؛ حيث تظهر مشكلة مجال الرؤية عندما يقرأ الخيط "أ" البيانات المُشاركة التي يُغيرها الخيط "ب" فيما بعد، دون معرفة الخيط "أ" بهذا التغيير؛ بينما تظهر مشكلة الوصول عند محاولة عدة خيوطٍ الوصول وتغيير نفس البيانات المُشاركة في الوقت ذاته.

تؤدي مشكلتا الوصول ومجال الرؤية إلى:

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

التزامن في جافا

العمليات والخيوط

يعمل برنامج جافا افتراضيًا ضمن عمليةٍ خاصةٍ به وضمن خيطٍ واحد، حيث تُعد الخيوط جزءًا من لغة جافا وتتعامل جافا معها باستخدام الشيفرة البرمجية Thread، ويستطيع تطبيق جافا إنشاء خيوطٍ جديدةٍ باستخدام هذا الصنف، كما يوفِّر الإصدار 1.5 من جافا دعمًا متقدمًا للتزامن باستخدام حزمة java.util.concurrent.

الأقفال ومزامنة الخيوط

توفّر لغة جافا أقفالًا لحماية أجزاءٍ مُعينةٍ من الشيفرة البرمجية لتنفيذها من قِبل عدة خيوطٍ بنفس الوقت، والطريقة الأسهل لقفل طريقة method أو صنف Class جافا هي من خلال تعريف هذه الطريقة أو الصنف باستخدام الكلمة المفتاحية synchronized.

تضمن الكلمة المفتاحية synchronized في جافا ما يلي:

• إمكانية تنفيذ كتلةٍ من الشيفرة البرمجية في نفس الوقت من قِبل خيطٍ واحدٍ فقط.
• يستطيع أي خيطٍ الاطلاع على التعديلات السابقة للكتلة المتزامنة من الشيفرة البرمجية والمحمية بنفس القفل عند دخوله إليها.

عملية التزامن ضرورية لضمان الوصول الحصري والمتبادل إلى الكتل البرمجية وضمان اتصالٍ موثوقٍ بين الخيوط.

تستطيع استخدام الكلمة المفتاحية synchronized عند تعريف الطريقة لتضمن أن خيطًا واحدًا فقط يستطيع استخدام هذه الطريقة بنفس الوقت، وبالتالي عند محاولة خيطٍ آخر لاستخدام هذه الطريقة، فسوف ينتظر حتى ينتهي الخيط الأول من استخدام هذه الطريقة.

public synchronized void critial() {
    // some thread critical stuff
    // here
}

يمكن أيضًا استخدام الكلمة المفتاحية synchronized لحماية كتل من الشيفرة البرمجية ضمن الطريقة method، حيث تكون الكتلة محميةً بمفتاحٍ من الممكن أن يكون محرفًا أو كائنًا، ويُدعى هذا المفتاح بالقفل، حيث لا يُمكن للشيفرات البرمجية المحمية بنفس القفل الوصول إلا من قِبل خيطٍ واحدٍ بنفس الوقت، فعلى سبيل المثال تضمن هيكلية البيانات التالية وصول خيطٍ واحدٍ فقط إلى داخل كتلتي الطريقتين ()add و()next.

package de.vogella.pagerank.crawler;

import java.util.ArrayList;
import java.util.List;

/**
 * Data structure for a web crawler. Keeps track of the visited sites and keeps
 * a list of sites which needs still to be crawled.
 *
 * @author Lars Vogel
 *
 */
public class CrawledSites {
    private List<String> crawledSites = new ArrayList<String>();
    private List<String> linkedSites = new ArrayList<String>();

    public void add(String site) {
        synchronized (this) {
            if (!crawledSites.contains(site)) {
                linkedSites.add(site);
            }
        }
    }

    /**
     * Get next site to crawl. Can return null (if nothing to crawl)
     */
    public String next() {
        if (linkedSites.size() == 0) {
            return null;
        }
        synchronized (this) {
            // Need to check again if size has changed
            if (linkedSites.size() > 0) {
                String s = linkedSites.get(0);
                linkedSites.remove(0);
                crawledSites.add(s);
                return s;
            }
            return null;
        }
    }

}

الذاكرة المتطايرة

يضمن التصريح عن أي متحولٍ باستخدام الكلمة المفتاحية volatile قراءة الخيط لآخر قيمة مكتوبة في الحقل، حيث أن الكلمة المفتاحية volatile لا تُطبق أي قفلٍ على المتغير.

بدءًا من الإصدار 5 لجافا، فقد صارت قيم المتغيرات الغير متطايرة -التي تُعدل من قبل نفس الخيط- تُحدَّث عند الكتابة على متغيرٍ متطايرٍ، كما يُمكن استخدام هذه الخاصية لتحديث قيمٍ ضمن متغيرٍ مرجعي مثل المتغير المتطاير "person"، ففي هذه الحالة لا بُدّ من استخدام متغير مؤقتٍ "person" واستخدام أداة ضبط لتهيئة متغير، ثم تعيين قيمة متغيرٍ مؤقتٍ للمتغير النهائي، وسوف يُغير هذا عنوان المتغير ويجعل القيم مرئيةً للخيوط الأخرى.

نموذج الذاكرة في جافا

يصف هذا النموذج عملية الاتصال بين ذاكرة الخيوط والذاكرة الرئيسية للتطبيق، فهو يُحدد شروط انتشار التغييرات في الذاكرة بين الخيوط، والحالات التي يُحدِث فيها الخيط ذاكرته الخاصة من الذاكرة الرئيسية، والعمليات الذرية وترتيب هذه العمليات.

العمليات الذرية

العملية الذرية هي عملية تُطبق كوحدة عملٍ منفردة دون أي تدخل من عملياتٍ أخرى، حيث تضمن مواصفات لغة جافا أن تكون عملية قراءة أو كتابة متغير ذريةً، إلا في حال كان نوع المتغير long أو double، وتكون العمليات على المتغيرات من نوع long أو short ذريةً فقط إذا صُرح عنها باستخدام الكلمة المفتاحية volatile.

بفرض أنه قد صُرِّح عن المتغير i على أنه صحيح int، فسوف تكون الزيادة i++‎ عمليةً غير ذرية في جافا، وهذا ينطبق على القيم الرقمية الأخرى مثل long. تقرأ العملية i++‎ أولًا القيمة المُخزنة حاليًا في i (عمليةٌ ذريةٌ)، ثم تُضيف قيمة واحد لها (عمليةٌ ذريةٌ)، ولكن يُحتمل أن تتغير القيمة بين عمليتي الكتابة والقراءة.

توفّر لغة جافا بدءًا من الإصدار 1.5 متغيراتٍ ذريةٍ مثل AtomicInteger، أو AtomicLong، والتي توفّر توابعًا مثل:

• getAndDecrement()‎
• getAndIncrement()‎
• getAndSet()‎

وجميعها ذرية.

تحديثات الذاكرة في الشيفرة البرمجية المتزامنة

يضمن نموذج الذاكرة في جافا أن يستطيع كل خيط يدخل كتلة شيفرة برمجية متزامنة، الاطلاع على جميع التعديلات التي حدثت تحت حماية نفس القفل.

الثبات والنسخ الوقائية

الثبات

يمكن تجنب مشاكل التزامن بطريقةٍ بسيطةٍ من خلال مشاركة البيانات الثابتة فقط بين الخيوط، وهي البيانات التي لا يُمكن تغييرها، كما يجب التصريح عن صنف مع كل الحقول الخاصة به على أنها نهائية final لجعله ثابتًا، ويجب التأكد من عدم تسرُّب أي مرجع للحقول أثناء عملية البناء، لذلك يجب أن يكون الحقل:

• خاص private.
• لا يمتلك طريقةً ضابطة.
• أن يُنسخ ضمن الباني إذا كان كائنًا متغيرًا لتجنب تغيير البيانات من خارج الصنف.
• لا يجب إعادة قيمته مباشرةً أو التعامل معه مباشرةً.
• لا يتغير وإذا حصل تغيير فيجب ألا يكون ظاهرًا للخارج.

يُمكن أن يتضمن الصنف الثابت بعض البيانات المتغيرة والتي تُستخدم لإدارة حالته، ولكن لا يُمكن تغيير هذا الصنف أو أي سمةٍ ضمنه من خارج الصنف، إذ يجب أن يُنشئ هذا الصنف نُسخًا وقائيةً عن عناصر الحقول الثابتة التي تُمرر له من الخارج مثل المصفوفة، وذلك لضمان عدم إمكانية أي كائنٍ خارجيٍ تغيير البيانات.

النسخ الوقائية

يجب حماية الأصناف من أي شيفرة استدعاء، إذ من الممكن لهذه الشيفرة تغيير البيانات بطريقةٍ لا تتوقعها حتى من أجل البيانات الثابتة التي لا تتوقع أن تغير قيمتها من خارج الصنف. ولحماية الصنف، يجب نسخ البيانات المُستقبلة وإعادة نُسخٍ فقط عن البيانات إلى شيفرة الاستدعاء.

يُنشئ المثال التالي نسخةً من اللائحة ArrayList ويُعيد النسخة فقط وبالتالي لا يستطيع عميل هذا الصنف إزالة العناصر من اللائحة.

package de.vogella.performance.defensivecopy;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class MyDataStructure {
    List<String> list = new ArrayList<String>();

    public void add(String s) {
        list.add(s);
    }

    /**
     * Makes a defensive copy of the List and return it
     * This way cannot modify the list itself
     *
     * @return List<String>
     */
    public List<String> getList() {
        return Collections.unmodifiableList(list);
    }
}

الخيوط في جافا

إن أساس التزامن في جافا هو الصنف java.lang.Threads، حيث يُنفّذ Thread كائنًا من النوع java.lang.Runnable، وتتضمن واجهة Runnable تعريفًا للطريقة ()run التي تُستدعى من قِبل الكائن Thread وتحتوي العمل الواجب تنفيذه، لذلك فإن Runnable هو مهمة يجب تنفيذها، أما Thread فهو العامل الذي يُنفذ هذه المهمة.

يوضح المثال التالي مهمة Runnable بحساب مجموع مجال مُعطى من الأرقام، لذا أنشِئ مشروع جافا وسمِّه de.vogella.concurrency.threads لهذا المثال.

package de.vogella.concurrency.threads;

/**
 * MyRunnable will count the sum of the number from 1 to the parameter
 * countUntil and then write the result to the console.
 * <p>
 * MyRunnable is the task which will be performed
 *
 * @author Lars Vogel
 *
 */
public class MyRunnable implements Runnable {
    private final long countUntil;

    MyRunnable(long countUntil) {
        this.countUntil = countUntil;
    }

    @Override
    public void run() {
        long sum = 0;
        for (long i = 1; i < countUntil; i++) {
            sum += i;
        }
        System.out.println(sum);
    }
}

يوضح المثال التالي استخدام صنفي Thread وRunnable.

package de.vogella.concurrency.threads;

import java.util.ArrayList;
import java.util.List;

public class Main {

    public static void main(String[] args) {
        // We will store the threads so that we can check if they are done
        List<Thread> threads = new ArrayList<Thread>();
        // We will create 500 threads
        for (int i = 0; i < 500; i++) {
            Runnable task = new MyRunnable(10000000L + i);
            Thread worker = new Thread(task);
            // We can set the name of the thread
            worker.setName(String.valueOf(i));
            // Start the thread, never call method run() direct
            worker.start();
            // Remember the thread for later usage
            threads.add(worker);
        }
        int running = 0;
        do {
            running = 0;
            for (Thread thread : threads) {
                if (thread.isAlive()) {
                    running++;
                }
            }
            System.out.println("We have " + running + " running threads. ");
        } while (running > 0);

    }
}

سلبيات الصنف Thread:

• يُؤثر إنشاء خيطٍ جديدٍ بعض الشيء على الأداء.
• سوف يُسبب إنشاء العديد من الخيوط تراجعًا في الأداء لأن وحدة المعالجة المركزية CPU سوف تضطر للتبديل بين هذه الخيوط.
• لا تستطيع التحكم بعدد الخيوط بسهولة لذلك سوف تواجه أخطاءً بسبب امتلاء الذاكرة الناتج عن العدد الكبير للخيوط.

اقتباس

تُقدم حزمة java.util.concurrent دعمًا مُحسّنًا للتزامن مقابل الاستخدام المُباشر للخيوط، وهذه الحزمة مشروحة في القسم التالي.

تجمعات الخيوط Thread Pools مع الإطار المنفذ Executer Framework

تُدير تجمُعات الخيوط مجموعةً من الخيوط العاملة، حيث تتضمن رتل عمل يحتفظ بالمهمات التي تنتظر أن تُنفذ، ويُمكن وصف تجمُّعات الخيوط على أنها مجموعةٌ من كائنات Runnable (رتل عمل) واتصالٌ من الخيوط العاملة، حيث تعمل هذه الخيوط باستمرار وتتحقق من استدعاء العمل لعملٍ جديد، فإذا كان عملًا جديدًا يجب تنفيذه، فستُنفذ هذا الكائن Runnable، حيث يوفّر صنف Thread نفسه طريقةً مثل (execute(Runnable r لإضافة كائن Runnable جديدٍ إلى رتل العمل.

يوفر الإطار المُنفذ مثالًا لتطبيق واجهة java.util.concurrent.Executor، مثل:

(Executor.newFixedThreadPool(int n

والذي سوف يُنشئ n خيط عامل، وتُضيف ExecutorService توابع دورة الحياة للمُنفذ، مما يسمح له بإطفاء المُنفذ وانتظار عملية الإغلاق.

اقتباس

إذا كنت تريد استخدام تجمع خيوط مع خيطٍ واحدٍ يُنفذ عدة كائنات Runnable، فتستطيع استخدام طريقة ()Executors.newSingleThreadExecutor.

أنشئ Runnable مرةً ثانية.

package de.vogella.concurrency.threadpools;

/**
 * MyRunnable will count the sum of the number from 1 to the parameter
 * countUntil and then write the result to the console.
 * <p>
 * MyRunnable is the task which will be performed
 *
 * @author Lars Vogel
 *
 */
public class MyRunnable implements Runnable {
    private final long countUntil;

    MyRunnable(long countUntil) {
        this.countUntil = countUntil;
    }

    @Override
    public void run() {
        long sum = 0;
        for (long i = 1; i < countUntil; i++) {
            sum += i;
        }
        System.out.println(sum);
    }
}

تستطيع الآن تشغيل كائنات Runnable باستخدام الإطار المُنفذ.

package de.vogella.concurrency.threadpools;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {
    private static final int NTHREDS = 10;

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(NTHREDS);
        for (int i = 0; i < 500; i++) {
            Runnable worker = new MyRunnable(10000000L + i);
            executor.execute(worker);
        }
        // This will make the executor accept no new threads
        // and finish all existing threads in the queue
        executor.shutdown();
        // Wait until all threads are finish
        executor.awaitTermination();
        System.out.println("Finished all threads");
    }
}

تستطيع استخدام صنف java.util.concurrent.Callable عند الحاجة لإعادة قيمةٍ معينة من الخيوط.

البرمجة غير المتزامنة

يُفضل تنفيذ أي عملية تستهلك وقتًا بصورةٍ غير متزامنة حيث توجد مقاربتان للتعامل مع المهام بصورةٍ غير متزامنة ضمن تطبيق جافا، وهما:

• حجب منطق التطبيق إلى حين إكمال المهمة.
• استدعاء منطق التطبيق حالما تكتمل المهمة وتُدعى هذه بالمقاربة دون إعاقة.

يدعم CompletableFuture وهو توسّع عن واجهة Future، الاستدعاءات غير المتزامنة، فهو يستعمل واجهة CompletionStage التي توفر توابعًا تسمح بربط الاستدعاءات التي سوف تُنفذ عند إكمالها، كما يُضيف تقنيات قياسية لتنفيذ شيفرة تطبيق برمجية عند إكمال مهمةٍ ما، وهذا يتضمن طرقًا متعددةً من أجل دمج المهام. يدعم CompletableFuture أيضًا المقاربتين بإعاقة ودون إعاقة، إضافةً إلى الاستدعاءات العادية. ويٌمكن تنفيذ الاستدعاء في خيطٍ آخر عند تنفيذ CompletableFuture في الخيط ذاته.

يُوضح المثال التالي كيفية إنشاء CompletableFuture أساسي.

CompletableFuture.supplyAsync(this::doSomething);

يُشغّل CompletableFuture.supplyAsync المهمة بصورةٍ غير متزامنة ضمن تجمُّع الخيط الافتراضية لجافا، ويكون الخيار مُتاحًا بالسماح للمنفذ المُخصص بتعريف ThreadPool.

package snippet;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class CompletableFutureSimpleSnippet {
    public static void main(String[] args) {
        long started = System.currentTimeMillis();

        // configure CompletableFuture
        CompletableFuture<Integer> futureCount = createCompletableFuture();

            // continue to do other work
            System.out.println("Took " + (started - System.currentTimeMillis()) + " milliseconds" );

            // now its time to get the result
            try {
              int count = futureCount.get();
                System.out.println("CompletableFuture took " + (started - System.currentTimeMillis()) + " milliseconds" );

               System.out.println("Result " + count);
             } catch (InterruptedException | ExecutionException ex) {
                // Exceptions from the future should be handled here
            }
    }

    private static CompletableFuture<Integer> createCompletableFuture() {
        CompletableFuture<Integer> futureCount = CompletableFuture.supplyAsync(
                () -> {
                    try {
                        // simulate long running task
                        Thread.sleep(5000);
                    } catch (InterruptedException e) { }
                    return 20;
                });
        return futureCount;
    }

}

يُمكن استخدام thenApply لتعريف استدعاء يُنفذ حالما ينتهي CompletableFuture.supplyAsync، ويوضح المثال التالي استخدام طريقة thenApply.

package snippet;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class CompletableFutureCallback {
    public static void main(String[] args) {
        long started = System.currentTimeMillis();

        CompletableFuture<String>  data = createCompletableFuture()
                .thenApply((Integer count) -> {
                    int transformedValue = count * 10;
                    return transformedValue;
                }).thenApply(transformed -> "Finally creates a string: " + transformed);

            try {
                System.out.println(data.get());
            } catch (InterruptedException | ExecutionException e) {

            }
    }

    public static CompletableFuture<Integer> createCompletableFuture() {
        CompletableFuture<Integer>  result = CompletableFuture.supplyAsync(() -> {
            try {
                // simulate long running task
                Thread.sleep(5000);
            } catch (InterruptedException e) { }
            return 20;
        });
        return result;
    }

}

تستطيع أيضًا تشغيل CompletableFuture مُؤخر بدءًا من الإصدار 9 لجافا.

CompletableFuture<Integer> future = new CompletableFuture<>();
 future.completeAsync(() -> {
       System.out.println("inside future: processing data...");
       return 1;
 }, CompletableFuture.delayedExecutor(3, TimeUnit.SECONDS))
 .thenAccept(result -> System.out.println("accept: " + result));

خوارزميات دون إعاقة

يوفر الإصدار 5.0 لجافا دعمًا لعملياتٍ ذريةٍ إضافية، وهذا يسمح بتطوير خوارزميات دون إعاقة لا تتطلب تزامنًا، ولكن تعتمد على تعليمات عتاد صلب ذري منخفضة المستوى مثل الموازنة والتبديل compare-and-swap أو اختصارًا CAS، حيث تتحقق عمليتا الموازنة والتبديل فيما إذا كان المتغير يحتوي على قيمةٍ ما، وفي حال احتوائه على تلك القيمة سوف تُنفذ العملية.

تكون الخوارزميات دون إعاقة عادةً أسرع من خوارزميات الإعاقة لأن تزامن الخيوط يظهر في مستويات أدنى (عتاد صلب).

يُنشئ المثال التالي عدّادا دون إعاقة يزداد باستمرار، وهذا المثال موجود ضمن مشروع project يُدعى:
 

 de.volgella.concurrency.nonblocking.counter.

package de.vogella.concurrency.nonblocking.counter;

import java.util.concurrent.atomic.AtomicInteger;

public class Counter {
    private AtomicInteger value = new AtomicInteger();
    public int getValue(){
        return value.get();
    }
    public int increment(){
        return value.incrementAndGet();
    }

    // Alternative implementation as increment but just make the
    // implementation explicit
    public int incrementLongVersion(){
        int oldValue = value.get();
        while (!value.compareAndSet(oldValue, oldValue+1)){
             oldValue = value.get();
        }
        return oldValue+1;
    }

}

واختبارًا.

package de.vogella.concurrency.nonblocking.counter;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class Test {
        private static final int NTHREDS = 10;

        public static void main(String[] args) {
            final Counter counter = new Counter();
            List<Future<Integer>> list = new ArrayList<Future<Integer>>();

            ExecutorService executor = Executors.newFixedThreadPool(NTHREDS);
            for (int i = 0; i < 500; i++) {
                Callable<Integer> worker = new  Callable<Integer>() {
                    @Override
                    public Integer call() throws Exception {
                        int number = counter.increment();
                        System.out.println(number );
                        return number ;
                    }
                };
                Future<Integer> submit= executor.submit(worker);
                list.add(submit);

            }


            // This will make the executor accept no new threads
            // and finish all existing threads in the queue
            executor.shutdown();
            // Wait until all threads are finish
            while (!executor.isTerminated()) {
            }
            Set<Integer> set = new HashSet<Integer>();
            for (Future<Integer> future : list) {
                try {
                    set.add(future.get());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
            if (list.size()!=set.size()){
                throw new RuntimeException("Double-entries!!!");
            }

        }


}

إن الجزء المثير للاهتمام هو كيفية تنفيذ طريقة ()incrementAndGet، فهي تستخدم عملية CAS.

public final int incrementAndGet() {
        for (;;) {
            int current = get();
            int next = current + 1;
            if (compareAndSet(current, next))
                return next;
        }
    }

يستفيد JDK من الخوارزميات بدون إعاقة باستمرار لرفع الأداء لجميع المطورين، كما أن عملية تطوير خوارزميات دون إعاقة صحيحة ليس بالأمر السهل، ويمكنك الاطلاع على المزيد من التفاصيل عن الخوارزميات بدون إعاقة من ibm.com.

إطار عمل Fork-Join ضمن جافا 7

طرح الإصدار 7 من جافا آليةً جديدةً تفرعيةً لتنفيذ المهمات المكثفة، وأُطلق الاسم fork-join على إطار العمل التابع لها، حيث يسمح إطار العمل هذا على توزيع مهمةٍ ما على عدة عاملين ثم انتظار النتيجة.

اقتباس

يمكن تنزيل الحزمة jsr166y للإصدار 6 لجافا.

أنشئ المشروع de.vogella.performance.forkjoin لتجريبها، وإن كنت تستخدم نسخةً مختلفةً عن جافا 7، فعليك إضافة jsr166y.jar إلى مسار الأصناف.

أنشئ أولًا الحزمة algorithm ثم الصنف التالي.

package algorithm;

import java.util.Random;

/**
 *
 * This class defines a long list of integers which defines the problem we will
 * later try to solve
 *
 */
public class Problem {
    private final int[] list = new int[2000000];

    public Problem() {
        Random generator = new Random(19580427);
        for (int i = 0; i < list.length; i++) {
            list[i] = generator.nextInt(500000);
        }
    }

    public int[] getList() {
        return list;
    }

}

بعد ذلك عرّف الصنف Solver كما هو موضح في المثال التالي.

package algorithm;

import java.util.Arrays;

import jsr166y.forkjoin.RecursiveAction;

public class Solver extends RecursiveAction {
    private int[] list;
    public long result;

    public Solver(int[] array) {
        this.list = array;
    }

    @Override
    protected void compute() {
        if (list.length == 1) {
            result = list[0];
        } else {
            int midpoint = list.length / 2;
            int[] l1 = Arrays.copyOfRange(list, 0, midpoint);
            int[] l2 = Arrays.copyOfRange(list, midpoint, list.length);
            Solver s1 = new Solver(l1);
            Solver s2 = new Solver(l2);
            forkJoin(s1, s2);
            result = s1.result + s2.result;
        }
    }
}

الآن عرّف صنفًا بسيطًا للتجريب وسمِّه Test.

package testing;

import jsr166y.forkjoin.ForkJoinExecutor;
import jsr166y.forkjoin.ForkJoinPool;
import algorithm.Problem;
import algorithm.Solver;

public class Test {

    public static void main(String[] args) {
        Problem test = new Problem();
        // check the number of available processors
        int nThreads = Runtime.getRuntime().availableProcessors();
        System.out.println(nThreads);
        Solver mfj = new Solver(test.getList());
        ForkJoinExecutor pool = new ForkJoinPool(nThreads);
        pool.invoke(mfj);
        long result = mfj.getResult();
        System.out.println("Done. Result: " + result);
        long sum = 0;
        // check if the result was ok
        for (int i = 0; i < test.getList().length; i++) {
            sum += test.getList()[i];
        }
        System.out.println("Done. Result: " + sum);
    }
}

التوقف التام أو الجمود Deadlock

يتضمن التطبيق المتزامن خطر حدوث جمود، حيث نقول عن مجموعةٍ من العمليات أنها تسبب جمودًا إذا كانت جميع هذه العمليات تنتظر حصول حدثٍ معينٍ لعمليةٍ أخرى موجودةٍ في نفس المجموعة، فمثلًا إذا كان الخيط "أ" ينتظر إغلاق الكائن "د" المحجوز من قِبل الخيط "ب"، والخيط "ب" ينتظر إغلاق الكائن "ت" المحجوز من قِبل العملية "أ"، فعندها سوف تُقفل هاتان العمليتان ولا يُمكنهما متابعة عملياتهما.

يُمكن تشبيه هذه الحادثة بالازدحام المروري، حيث تحتاج السيارات (الخيوط) الوصول لشارعٍ مُعين (موارد)، وهذا الشارع محجوزٌ حاليًا من قِبل سيارةٍ أخرى (قفل).

ترجمة -وبتصرّف- للمقال Java concurrency (multi-threading) - Tutorial لصاحبه Lars Vogel.

اقرأ أيضًا

• الأحداث غير المتزامنة: حلقات الجس Polling Loops والمقاطعات Interrupts في المعالج
• الدليل السريع للغة البرمجة Java
• الأصناف المتداخلة Nested Classes في جافا
• المتغيرات والأنواع البسيطة في لغة جافا