نحوه استفاده از ForkJoinPool در جاوا

از ForkJoinPool برای تجزیه وظایف محاسباتی فشرده و اجرای موازی آنها برای عملکرد بهتر برنامه جاوا استفاده کنید.

• تجمیع موضوع با ForkJoinPool
• الگوریتم سرقت کار
• کلاس‌های اصلی ForkJoinPool
• استفاده از RecursiveAction
• کار بازگشتی
• زمان استفاده از ForkJoinPool
• خلاصه

ForkJoinPool یک کلاس جاوا قدرتمند است که برای پردازش وظایف محاسباتی فشرده استفاده می شود. این کار با تقسیم وظایف به وظایف فرعی کوچکتر و سپس اجرای موازی آنها کار می کند. این مجموعه رشته با استفاده از یک استراتژی تقسیم کن و حکومت کن، کار می‌کند که به آن امکان می‌دهد وظایف را همزمان اجرا کند، توان عملیاتی را افزایش داده و زمان پردازش را کاهش دهد.

یکی از ویژگی های منحصر به فرد ForkJoinPool الگوریتم سرقت کار است که برای بهینه سازی عملکرد استفاده می کند. هنگامی که یک Worker Thread وظایف محول شده خود را به پایان می‌رساند، وظایف را از رشته‌های دیگر می‌دزدد و اطمینان حاصل می‌کند که همه رشته‌ها به طور موثر کار می‌کنند و هیچ منابع کامپیوتری هدر نمی‌رود.

ForkJoinPool به‌طور گسترده در جریان‌های موازی جاوا و CompletableFutures استفاده می‌شود و به توسعه‌دهندگان اجازه می‌دهد تا وظایف را همزمان با سهولت انجام دهند. علاوه بر این، سایر زبان‌های JVM مانند Kotlin و Akka از این چارچوب برای ساخت برنامه‌های پیام‌محور استفاده می‌کند که نیاز به همزمانی و انعطاف‌پذیری بالایی دارند.

تجمیع موضوعات با ForkJoinPool

کلاس ForkJoinPool کارگران را ذخیره می کند، که فرآیندهایی هستند که روی هر هسته CPU از دستگاه اجرا می شوند. هر یک از این فرآیندها در یک deque ذخیره می‌شوند که مخفف صف دو طرفه است. به محض اینکه کار یک رشته کارگر تمام شود، شروع به سرقت وظایف از سایر کارگران می کند.

ابتدا، روند انجام کار وجود خواهد داشت. این بدان معنی است که یک کار بزرگ به وظایف کوچکتر تقسیم می شود که می توانند به صورت موازی اجرا شوند. پس از تکمیل تمام وظایف فرعی، آنها مجدداً ملحق می شوند. کلاس ForkJoinPool سپس یک نتیجه را ارائه می دهد، همانطور که در شکل ۱ نشان داده شده است.

شکل ۱. ForkJoinPool در عمل

وقتی کار در یک ForkJoinPool ارسال می‌شود، فرآیند به فرآیندهای کوچک‌تر تقسیم می‌شود و به یک صف مشترک هدایت می‌شود.

هنگامی که متد fork() فراخوانی شد، وظایف به صورت موازی فراخوانی می شوند تا زمانی که شرط پایه درست باشد. هنگامی که پردازش فورک شد، روش join() تضمین می‌کند که رشته‌ها برای یکدیگر منتظر می‌مانند تا فرآیند نهایی شود.

همه وظایف در ابتدا به یک صف اصلی ارسال می‌شوند و این صف اصلی وظایف را به رشته‌های کارگر هدایت می‌کند. توجه داشته باشید که وظایف با استفاده از استراتژی LIFO (آخرین ورود، اولین خروج) که مشابه ساختار داده پشته.

نکته مهم دیگر این است که ForkJoinPool از deque ها برای ذخیره وظایف استفاده می کند. این قابلیت استفاده از LIFO یا FIFO (اول وارد، اولین خروج) را می دهد که برای الگوریتم سرقت کار ضروری است.

شکل ۲. ForkJoinPool از Deques برای ذخیره وظایف استفاده می کند

الگوریتم سرقت کار

سرقت کار در ForkJoinPool یک الگوریتم موثر است که استفاده کارآمد از منابع رایانه را با متعادل کردن حجم کار در تمام رشته‌های موجود در استخر امکان‌پذیر می‌سازد.

وقتی رشته‌ای بیکار می‌شود، به جای غیرفعال ماندن، سعی می‌کند تا کارهایی را از رشته‌های دیگری که هنوز مشغول کار محول شده خود هستند، بدزدد. این فرآیند استفاده از منابع محاسباتی را به حداکثر می‌رساند و تضمین می‌کند که هیچ رشته‌ای بیش از حد بارگذاری نمی‌شود در حالی که بقیه بیکار می‌مانند.

مفهوم کلیدی پشت الگوریتم سرقت کار این است که هر رشته وظایف خاص خود را دارد که به ترتیب LIFO آنها را اجرا می کند.

وقتی یک رشته کارهای خود را تمام می‌کند و بی‌کار می‌شود، با پیروی از یک استراتژی FIFO، مانند ساختار داده صف. این به رشته بیکار اجازه می دهد تا کارهایی را که برای طولانی ترین زمان منتظر مانده اند انجام دهد، و زمان کلی انتظار را کاهش می دهد و توان عملیاتی را افزایش می دهد.

در نمودار زیر، Thread 2 با جمع‌بندی آخرین عنصر از Deque Thread 1، وظیفه‌ای را از Thread 1 می‌دزدد و سپس کار را اجرا می‌کند. کار دزدیده شده معمولاً قدیمی‌ترین کار در دک است، که تضمین می‌کند که حجم کار به طور مساوی بین تمام رشته‌های موجود در استخر توزیع می‌شود.

شکل ۳. تصویری از الگوریتم سرقت کار ForkJoinPool

به طور کلی، الگوریتم کار دزدی ForkJoinPool یک ویژگی قدرتمند است که می تواند عملکرد برنامه های موازی را با اطمینان از استفاده کارآمد از همه منابع محاسباتی موجود، به طور قابل توجهی بهبود بخشد.

کلاس های اصلی ForkJoinPool

بیایید نگاهی گذرا به کلاس‌های اصلی که پردازش با استفاده از ForkJoinPool را پشتیبانی می‌کنند بیاندازیم.

• ForkJoinPool: یک Thread Pool برای استفاده از چارچوب ForkJoin ایجاد می کند. به طور مشابه با دیگر استخرهای نخ کار می کند. مهمترین روش از این کلاس commonPool() است که مخزن رشته ForkJoin را ایجاد می کند.
• RecursiveAction: وظیفه اصلی این کلاس محاسبه اقدامات بازگشتی است. به یاد داشته باشید که در روش compute()، مقداری بر نمی‌گردانیم. این به این دلیل است که بازگشت در روش compute() رخ می دهد.
• RecursiveTask: این کلاس مشابه RecursiveAction کار می کند، با این تفاوت که روش compute() مقداری را برمی گرداند. li>

استفاده از RecursiveAction

برای استفاده از قابلیت‌های RecursiveAction باید آن را به ارث ببریم و روش compute() را لغو کنیم. سپس، وظایف فرعی را با منطقی که می‌خواهیم پیاده‌سازی کنیم، ایجاد می‌کنیم.

در مثال کد زیر، عددی را که دو برابر هر عدد در آرایه است به صورت موازی و بازگشتی محاسبه می کنیم. ما محدود به محاسبه دو در دو عنصر آرایه به صورت موازی هستیم.

همانطور که می بینید، روش fork() متد compute() را فراخوانی می کند. به محض اینکه کل آرایه دارای مجموع هر یک از عناصر خود باشد، فراخوانی بازگشتی متوقف می شود. هنگامی که تمام عناصر آرایه به صورت بازگشتی جمع شدند، نتیجه را نشان می دهیم.

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;

public class ForkJoinDoubleAction {

  public static void main(String[] args) {
    ForkJoinPool forkJoinPool = new ForkJoinPool();
    int[] array = {1, 5, 10, 15, 20, 25, 50};
    DoubleNumber doubleNumberTask = new DoubleNumber(array, 0, array.length);

    // Invokes compute method
    forkJoinPool.invoke(doubleNumberTask);
    System.out.println(DoubleNumber.result);
  }
}

class DoubleNumber extends RecursiveAction {

  final int PROCESS_THRESHOLD = 2;
  int[] array;
  int startIndex, endIndex;
  static int result;

  DoubleNumber(int[] array, int startIndex, int endIndex) {
    this.array = array;
    this.startIndex = startIndex;
    this.endIndex = endIndex;
  }

  @Override
  protected void compute() {
    if (endIndex - startIndex <= PROCESS_THRESHOLD) {
      for (int i = startIndex; i < endIndex; i++) {
        result += array[i] * 2;
      }
    } else {
      int mid = (startIndex + endIndex) / 2;
      DoubleNumber leftArray = new DoubleNumber(array, startIndex, mid);
      DoubleNumber rightArray = new DoubleNumber(array, mid, endIndex);

      // Invokes the compute method recursively
      leftArray.fork();
      rightArray.fork();

      // Joins results from recursive invocations
      leftArray.join();
      rightArray.join();
    }
  }
}

خروجی از این محاسبه ۲۵۲ است.

نکته مهمی که باید از RecursiveAction به خاطر بسپارید این است که مقداری را بر نمی گرداند. همچنین می‌توان با استفاده از استراتژی تفرقه بینداز و حکومت کن برای افزایش عملکرد، فرآیند را شکست داد.

این کاری است که ما در فهرست ۱ انجام دادیم، به جای محاسبه دو برابری هر عنصر آرایه، این کار را به صورت موازی با تقسیم آرایه به قطعات انجام دادیم.

همچنین مهم است که توجه داشته باشید که RecursiveAction زمانی مؤثرتر است که برای کارهایی استفاده شود که می‌توانند به طور مؤثر به زیرمشکلات کوچک‌تر تقسیم شوند.

بنابراین، RecursiveAction و ForkJoinPool باید برای کارهای محاسباتی فشرده استفاده شوند که در آن موازی سازی کار می تواند منجر به بهبود عملکرد قابل توجهی شود. در غیر این صورت، به دلیل ایجاد و مدیریت موضوعات، عملکرد حتی بدتر خواهد شد.

کار بازگشتی

در این مثال بعدی، اجازه دهید یک برنامه ساده را بررسی کنیم که به صورت بازگشتی در وسط شکسته می شود تا زمانی که به شرایط پایه برسد. در این مورد، ما از کلاس RecursiveTask استفاده می کنیم.

تفاوت RecursiveAction و RecursiveTask در این است که با RecursiveTask، می‌توانیم مقداری را در compute() روش.

import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public class ForkJoinSumArrayTask extends RecursiveTask<Integer> {

  private final List<Integer> numbers;

  public ForkJoinSumArrayTask(List<Integer> numbers) {
    this.numbers = numbers;
  }

  @Override
  protected Integer compute() {
    if (numbers.size() <= 2) {
      return numbers.stream().mapToInt(e -> e).sum();
    } else {
      int mid = numbers.size() / 2;
      List<Integer> list1 = numbers.subList(0, mid);
      List<Integer> list2 = numbers.subList(mid, numbers.size());
 
      ForkJoinSumArrayTask task1 = new ForkJoinSumArrayTask(list1);
      ForkJoinSumArrayTask task2 = new ForkJoinSumArrayTask(list2);

      task1.fork();

      return task1.join() + task2.compute();
    }
  }

  public static void main(String[] args) {
    ForkJoinPool forkJoinPool = new ForkJoinPool();

    List<Integer> numbers = List.of(1, 3, 5, 7, 9);
    int output = forkJoinPool.invoke(new ForkJoinSumArrayTask(numbers));

    System.out.println(output);
  }
}

در اینجا، ما به صورت بازگشتی آرایه را در وسط تجزیه می کنیم تا زمانی که به شرایط پایه برسد.

هنگامی که آرایه اصلی شکسته شد، list1 و list2 را به ForkJoinSumArrayTask می فرستیم، سپس task1< /code> که متد compute() و قسمت دیگر آرایه را به صورت موازی اجرا می کند.