개요

자바의 실행자(Executor) 프레임워크는 스레드에 대해 그리고 시스템에서 스레드가 사용하는 자원에 대한 새로운 차원의 제어를 가능하게 한다.

Executor 프레임워크에 관련된 클래스를 사용하면 시스템이 스레드 수를 관리하거나 더 이상 필요하지 않은 스레드를 취소 할 수 있다.

• 스레드에서 실행되기를 기다리는 태스크 수를 제어하는 작업자 스레드 풀과 큐를 설정한다.
• 비정상적으로 종료되는 스레드를 초래하는 에러를 확인한다.
• 완료되는 스레드를 기다리고 스레드로부터 결과를 가져온다.
• 스레드의 일괄 처리를 실행하고 고정된 순서로 결과를 가져온다.
• 사용자가 더 빨리 결과를 확인할 수 있도록 알맞은 시간에 백그라운드 스레드를 시작한다.

• 태스크 큐잉
• 태스크 실행 순서
• 태스크 실행 유형 (직렬 또는 동시)

private static class SerialExecutor implements Executor {
        final ArrayDeque mTasks = new ArrayDeque();
        Runnable mActive;
        public synchronized void execute(final Runnable r) {
            mTasks.offer(new Runnable() {
                @Override
                public void run() {
                    try {
                        r.run();

                    } finally {
                        scheduleNext();
                    }
                }
            });
            if (mActive == null) {
                scheduleNext();
            }
        }

        protected synchronized void scheduleNext() {
            if((mActive = mTasks.poll()) != null) {
                THREAD_POOL_EXECUTOR.execute(mActive);
            }
        }
    }

• 작업자 스레드는 실행할 다음 태스크를 기다리기 위해서 살아 있다. 매 스레드를 생성하고 파괴할 때 생기는 오버헤드를 줄일 수 있다.
• 스레드의 최대 개수로 정의된다. 백그라운드 스레드 때문에 발생하는 과부하를 막는다.
• 작업자 스레드의 생명주기는 스레드 풀 생명주기에 의해 제어된다.

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        Executors.newFixedThreadPool(10, new LowPriorityThreadFactory());
    }


    static class LowPriorityThreadFactory implements ThreadFactory {
        private static int count = 1;
        private static String TAG = "LowPriorityThreadFactory";


        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setName("LowPrio " + count++);
            t.setPriority(4);
            t.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
                @Override
                public void uncaughtException(Thread t, Throwable throwable) {
                    Log.d(TAG, "Thread = " +t.getName() + ", error = " + throwable.getMessage());
                }
            });
            return t;
        }
    }

public class TaskTrackingThreadPool extends ThreadPoolExecutor {
        private AtomicInteger mTaskCount = new AtomicInteger(0);

        public TaskTrackingThreadPool() {
            super(3,3,0L,TimeUnit.SECONDS, new LinkedBlockingQueue());
        }

        @Override
        protected void beforeExecute(Thread t, Runnable r) {
            super.beforeExecute(t, r);
            mTaskCount.getAndIncrement();
        }

        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            super.afterExecute(r, t);
            mTaskCount.getAndIncrement();
        }

        public int getNbrOfTasks() {
            return mTaskCount.get();
        }
    }

 BlockingQueue preloadedQueue = new LinkedBlockingQueue();
        final String[] alphabat = {"Alpha", "Beta", "Gamma", "Delta", "Epsilon", "Zeta"};
        for (int i = 0; i < alphabat.length; i++) {
            final int j = i;
            preloadedQueue.add(new Runnable() {
                @Override
                public void run() {
                    //긴 동작들
                }
            });
        }
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 1, TimeUnit.SECONDS, preloadedQueue);
        executor.prestartAllCoreThreads();

• 핵심 풀 크기에 도달하지 않는 경우, 새로운 스레드가 생성될 수 있다.
• 핵심 풀 크기에 도달했고 큐에 빈 슬롯이 있는 경우, 태스크는 큐에 추가될 수 있다.
• 핵심 풀 크기에 도달했으나 큐가 가득 찬 경우, 태스크는 거부된다.

@UiThread
    public void onButtonClick(View v) {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor();
        threadPoolExecutor.execute(new Runnable() {
            @Override
            public void run() {
                List> tasks = new ArrayList>();
                tasks.add(new Callable() {
                    @Override
                    public String call() throws Exception {
                        return getFirstDataFromNetwork();
                    }
                });
                tasks.add(new Callable() {
                    @Override
                    public String call() throws Exception {
                        return getSecondDataFromNetwork();
                    }
                });

                ExecutorService executorService = Executors.newFixedThreadPool(2);
                try {
                    List> futures = executorService.invokeAll(tasks);
                    String mashedData = mashupResult(futures);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                executorService.shutdown();
            }
        });
    }

    private String getFirstDataFromNetwork() {
        // 네트워크 호출
        return "1";
    }

    private String getSecondDataFromNetwork() {
        //네트워크 호출
        return "2";
    }

    private String mashupResult(List> futures) throws InterruptedException {
        for (Future future : futures) {
            // data = futures.get() 으로 결과 얻음
        }
        return "data";
    }

    private LinearLayout layoutImages;

    private class ImageDownloadTask implements Callable {
        @Override
        public Bitmap call() throws Exception {
            return downloadRemoteImage();
        }

        private Bitmap downloadRemoteImage() {
            // 이미지 다운로드 실행
            return null;
        }
    }

    private class DownloadCompletionService extends ExecutorCompletionService {
        private ExecutorService mExecutor;

        public DownloadCompletionService(ExecutorService executor) {
            super(executor);
            mExecutor = executor;
        }

        public void shutdown() {
            mExecutor.shutdown();
        }

        public boolean isTerminated() {
            return mExecutor.isTerminated();
        }
    }

    private class ConsumerThread extends Thread {
        private DownloadCompletionService mEcs;

        private ConsumerThread(DownloadCompletionService ecs) {
            this.mEcs = ecs;
        }

        @Override
        public void run() {
            super.run();
            try {
                while (!mEcs.isTerminated()) {
                    Future future = mEcs.poll(1, TimeUnit.SECONDS);
                    if (future != null) {
                        addImage(future.get());
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    }

    private void addImage(final Bitmap image) {
        runOnUiThread(new Runnable() {
            @Override
            public void run() {
                ImageView iv = new ImageView(MainActivity.this);
                iv.setImageBitmap(image);
                layoutImages.addView(iv);
            }
        });
    }

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        DownloadCompletionService ecs = new DownloadCompletionService(Executors.newCachedThreadPool());
        new ConsumerThread(ecs).start();
        for (int i = 0; i < 5; i++) {
            ecs.submit(new ImageDownloadTask());
        }
        ecs.shutdown();

9.5 마치며

Executor 프레임 워크는 안드로이드 비동기 기술의 토대를 제공한다. 해당 기법을 이용하면 더 정교한 방법으로 동시성을 관리하게 해준다. 다음시간에는 안드로이드 대표 기술인 AsyncTask에 대해서 배워보자