java.util.concurrent.ForkJoinWorkerThread에 대한 Java 코드 예제

이 글에서는 클래스에 대해 논의java.util.concurrent.ForkJoinWorkerThread`하고 강력한 멀티 스레드 응용 프로그램을 빌드 할 때 자신의 코드에서 클래스를 사용하는 방법에 대해 설명합니다.

ForkJoinTasks를 실행하는 ForkJoinPool에서 관리하는 스레드입니다. 이 클래스는 기능 추가를 위해서 서브 클래스만 가능합니다. 스케줄링 또는 실행을 처리하는 재정의 가능한 메소드는 없습니다. 그러나 기본 작업 처리 루프를 둘러싼 초기화 및 종료 방법을 재정의 할 수 있습니다. 이러한 서브 클래스를 작성하는 경우, ForkJoinPool에서 {@linkplain ForkJoinPool#ForkJoinPool use it}에 사용자 정의 ForkJoinPool.ForkJoinWorkerThreadFactory를 제공해야합니다.

1. ForkJoinWorkerThread 클래스

Java 7의 가장 흥미로운 기능 중 하나는_Fork / Join 프레임 워크_입니다.그것은의 구현입니다Executor 그리고ExecutorService 당신에게 실행 가능 인터페이스 Callable 와Runnable 그들을 실행 스레드를 관리하지 않고 작업을.

이 실행 프로그램은 더 작은 부분으로 나눌 수있는 작업을 실행하기위한 것입니다.주요 구성 요소는 다음과 같습니다.

• ForkJoinTask 클래스에의해 구현 된 특별한 종류의 작업.
• 작업을 하위 작업 (포크작업)으로 나누고해당 하위 작업이 완료 될 때까지 기다리는 (조인작업) 두 가지 작업.
• 풀 스레딩의 사용을 최적화하는 작업 스털링 알고리즘을 나타내는 알고리즘입니다.작업이 하위 작업을 기다리는 경우 작업을 실행중인 스레드는 다른 스레드를 실행하는 데 사용됩니다.

Fork / Join 프레임 워크의 주요 클래스는ForkJoinPool 클래스입니다. 내부적으로 다음 두 가지 요소가 있습니다.

• 실행 대기중인 작업 대기열
• 작업을 실행하는 스레드 풀

2. 일부 코드 실행

WorkerThread.java

package com.javacodegeeks.examples.forkjoinworkerthread.threads;

//~--- JDK imports ------------------------------------------------------------

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinWorkerThread;

\*
 * This class implements a custom thread for the Fork/Join framework. It extends
 * the ForkJoinWorkerThread that is the default implementation of the threads
 * that executes the tasks in the Fork/Join Framework. This custom thread counts
 * the number of tasks executed in it
 */
public class WorkerThread extends ForkJoinWorkerThread {
    private static ThreadLocal<Integer> taskCounter = new ThreadLocal();
    public WorkerThread(ForkJoinPool pool) {
        super(pool);
    }
    @Override
    protected void onStart() {
        super.onStart();
        System.out.printf("WorkThread %d: Initializing task counter.\\n", this.getId());
        taskCounter.set(0);
    }
    @Override
    protected void onTermination(Throwable exception) {
        System.out.printf("WorkerThread %d: %d\\n", getId(), taskCounter.get());
        super.onTermination(exception);
    }
    public void addTask() {
        int counter = taskCounter.get().intValue();
        counter++;
        taskCounter.set(counter);
    }
}

WorkerThreadFactory.java

package com.javacodegeeks.examples.forkjoinworkerthread.factories;

//~--- non-JDK imports --------------------------------------------------------
import com.javacodegeeks.examples.forkjoinworkerthread.threads.WorkerThread;
//~--- JDK imports ------------------------------------------------------------

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinPool.ForkJoinWorkerThreadFactory;
import java.util.concurrent.ForkJoinWorkerThread;

/*
 * Factory to be used by the Fork/Join framework to create the worker threads. Implements
 * the ForkJoinWorkerThreadFactory interface
 */
public class WorkerThreadFactory implements ForkJoinWorkerThreadFactory {
    @Override
    public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
        return new WorkerThread(pool);
    }
}

ARecursiveTask.java

package com.javacodegeeks.examples.forkjoinworkerthread.tasks;

//~--- non-JDK imports --------------------------------------------------------

import com.javacodegeeks.examples.forkjoinworkerthread.threads.WorkerThread;

//~--- JDK imports ------------------------------------------------------------

import java.util.concurrent.ExecutionException;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;

/*
 * Task that will be executed in the Fork/Join framework. It calculates
 * the sum of all array elements
 */

public class ARecursiveTask extends RecursiveTask {
    private static final long serialVersionUID = -4702976772011326493L;

    // Array to be summed
    private int\[\] intArray;

    // Start and end positions of the part of the array to be summed by this task
    private int start, end;

    public ARecursiveTask(int\[\] array, int start, int end) {
        this.intArray = array;
        this.start    = start;
        this.end      = end;
    }

    @Override
    protected Integer compute() {
        Integer      ret;
        WorkerThread thread = (WorkerThread) Thread.currentThread();
        thread.addTask();

        if (end - start > 100) {
            int            mid   = (start + end) / 2;
            ARecursiveTask task1 = new ARecursiveTask(intArray, start, mid);
            ARecursiveTask task2 = new ARecursiveTask(intArray, mid, end);
            invokeAll(task1, task2);
            ret = addResults(task1, task2);
        } else {
            int add = 0;
            for (int i = start; i < end; i++) {
                add += intArray\[i\];
            }

            ret = new Integer(add);
        }

        try {
            TimeUnit.MILLISECONDS.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return ret;
    }

    private Integer addResults(ARecursiveTask task1, ARecursiveTask task2) {
        int value;

        try {
            value = task1.get().intValue() + task2.get().intValue();
        } catch (InterruptedException e) {
            e.printStackTrace();
            value = 0;
        } catch (ExecutionException e) {
            e.printStackTrace();
            value = 0;
        }

        return new Integer(value);
    }

}

이전 코드에서 사용 된 방법을 설명하겠습니다

• protected void onStart() \– 시공 후 작업을 처리하기 전에 내부 상태를 초기화합니다.이 메소드를 대체하는 경우 메소드 시작시 super.onStart ()를 호출해야합니다.초기화에는주의가 필요합니다.이 스레드가 작업 처리를 시작하기 전에 다른 스레드의 액세스 시도가 올바르게 작동하도록하려면 대부분의 필드에 올바른 기본값이 있어야합니다.
• protected void onTermination(Throwable exception) \– 이 작업자 스레드 종료와 관련된 정리를 수행합니다.이 메소드를 대체하는 경우 대체 된 메소드의 끝에서 super.onTermination을 호출해야합니다.

명령의 출력

com.javacodegeeks.examples.forkjoinworkerthread.App

다음과 유사해야합니다.

WorkThread 8: Initializing task counter.
WorkThread 9: Initializing task counter.
WorkThread 10: Initializing task counter.
WorkThread 11: Initializing task counter.
WorkerThread 10: 543
WorkerThread 9: 448
WorkerThread 8: 513
WorkerThread 11: 543
Main: Result: 100000
Main: End of the program

3.이 학습서의 Eclipse 프로젝트를 다운로드하십시오.

ForkJoinWorkerThread 클래스사용 설정 방법의 예입니다.

당신은 여기서 예제의 전체 소스 코드를 다운로드 할 수 있습니다 : forkjoinworkerthread.zip