[Java]鋳型にはめるマルチスレッド

#はじめに
バッチ処理などで、時間のかかる似たような処理をマルチスレッドを使用して、
時間短縮させたいことありますよね。

例えば、単純化して考えるために、
1処理5分かかるものを10処理することを考えます。
1スレッドだと 5 * 10 = 50分かかりますが、
10スレッドだと5分で終わりそうな気がします。

そういった場合にスレッドセーフな処理で記述するというのは、
なかなか骨が折れます。
次回以降、似たような場面で心理的負荷を下げるために、
鋳型のようなものを作成しておきたいと思って書きました。

##実装例
とりあえずこの形ならスレッドセーフに書けるという大雑把な枠組みです。
欲しい情報の肝以外は省略しているつもりなので、
自分の必要な処理を埋め込んでいくと、
完成するような「鋳型」をイメージしています。

MultiThreadExecute.java

/**
* マルチスレッド化するクラス
* ここでスレッドを作成し実行させる
*/
public class MultiThreadExecute {
	public static void main(String[] args) {
		MultiThreadExecute logic = new MultiThreadExecute();
		int status = logic.run();
	}

	/** スレッド作成数 */
	private static const THREAD_COUNT = 5;
	
	protected int run() {
	
		// 処理対象データ取得
		int dataSize = ...(省略)...;
		
		// 並行実行処理
		// 終端処理を含めるためスレッド数分領域を増やしておく
		final BlockingQueue<DataDto> taskQueue =
            new LinkedBlockingQueue<DataDto>(dataSize + THREAD_COUNT);
            
		// キューに追加
		for (DataDto data : (省略。どこかから取得)) {
			taskQueue.add(data);
		}
		
		final OnExecListener listener = new OnExecListener();
		final ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
        for (int i = 0; i < THREAD_COUNT; i++) {
            // 終端ということが分かるようにしておく
            taskQueue.add(空のDataDtoなどを入れる);

            // 子スレッドの実行
            final LogicTask task = new LogicTask(taskQueue, listener);
            executor.execute(task);
        }

		// スレッドの終了を待機
        executor.shutdown();
        while (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
        }
		
		// 処理件数などは以下のように取得
		// listener.getTotalCount()
	}
}

DataDto.java

public class DataDto {
	// 処理をするために渡す引数を格納
}

Listener.java

/**
 * 処理終了検知
 */
private static interface Listener {
	/**
	* 処理後に欲しい情報を設計し、作成する
	* interfaceをきっちり定義することで、
	* 欲しい情報を柔軟に対応できるようにする
	*/
	void execute();
}

OnExecListener.java

/**
 * 処理終了リスナー
 */
private static class OnExecListener implements Listener {

	private int totalCount = 0;

    @Override
    public void execute(){
        synchronized (this) {
            totalCount++;
        }
    }

    synchronized int getTotalCount() {
        return totalCount;
    }
}

LogicTask.java

private static class LogicTask implements Runnable {
	private final BlockingQueue<DataDto> taskQueue;
	private final Listener listener;
	
	/**
	* コンストラクタ
	*/
	LogicTask(BlockingQueue<DataDto> taskQueue, Listener listener) {
		this.taskQueue = taskQueue;
		this.listener = listener;
	}

	/**
	* タスクの実行
	*/
	@Override
	public void run() {
		try {
		    while (true) {
		        if (終了判定) {
		            break;
		        }
		        
		        final DataDto data = taskQueue.take();
		        
		        // 時間のかかる処理を実行
		        exec(data);
		    }
		} catch (// 省略) {
		    // 省略
		}
	}
	
	private void exec(DataDto data){
		// 省略
		return;
	}
}

#思ったこと

• スレッドセーフにマルチスレッド処理を記述するのは気を遣う
• 自分なりの「鋳型」を増やしていくことで1回きりにしない！