マルチスレッド

[GitEngHar]

https://github.com/GitEngHar/learnJava/issues/1 上記に紐づく issue

検証ソースコード

[GitEngHar]

## テンプレ用
### aaa

#### コード

◆aaa
```java

実行結果

[GitEngHar]

マルチスレッドについて

• 意識すること
  • 並列処理の各終了時間が均等であること
  • スレッド間でお互いに町が発生する処理が存在しないこと

マルチスレッドによる問題

以下問題を解決する条件のことを スレッドセーフ という

1. スレッド数が増加するため、スレッドを管理するメモリの消費が大きくなる
2. 単位時間当たりの処理量(スループット)が低下する
3. 作業ファイルでデグレーションが発生する
4. 端末/サーバが占有される場合。方スレッドの処理時間が待ち時間となる

更に現場で恐れられる理由

マルチスレッドを利用することは現場で恐れられているという
発生するエラーの多くが再現性がとれないもの
デッドロック問題のようにタイミングが重なって起きる問題で発生するからである
その為、安易にスレッド処理を作成してはならない

Synchronized を用いた同期処理

synchronized は修飾子になっており、これを利用したメソッド内の実行はシングルスレッドのみとなり
複数スレッドで呼び出す際は同時実行できず、先に呼び出した処理が終了するまで待つ

synchronized void sycs(){}

• クラス単位 (クラスないメソッドで同期を取る場合は以下に配慮する)

  • 別のクラスは同期しない
  • 同クラス内のみで同期処理となる

• メソッド単位

  • map.get , map.putによる無限ループ処理は各単体での処理ではなく一連の処理の流れで発生するため、同期処理の範囲に配慮する

synchronized は スコープに配慮し利用するべき
スコープ内はシングルスレッドになるため、多用や広範囲での利用はスループットの低下となる

[GitEngHar]

デグレーションによる計算不良

inclementでの検証

◆100万回の計算

public class Incrementer implements Runnable{
    private String name;
    private HolderInt holders;

    public Incrementer(String name,HolderInt holder){
        this.name = name;
        this.holders = holder;
    }
    public void run(){
        System.out.printf("[%s] started %n",name);
        for(int i=0;i<1000000;i++){
            holders.increment();
        }
        System.out.printf("[%s] started %n",name);
    }
}

◆ インクリメントの実行

public class HolderInt {
    private int intNum = 0;

    public int getNum(){
        return intNum;
    }

    public void increment(){
        this.intNum++;
    }
}

◆ 100万回 計算 を 2スレッドで並列実行

import java.util.List;

public class ResourceLock implements Runnable

{
    private String name;
    private List<String> fromList;
    private List<String> toList;

    public ResourceLock(String name,List<String> fromList, List<String> toList){
        this.name = name;
        this.fromList = fromList;
        this.toList = toList;
    }

    public void run(){
        String str = null;
        try{
            // %n は改行
            System.out.printf("[%s] started.%n",name);
            Thread.sleep(500L);
            System.out.printf("[%s] attempt to lock fmList(%s).%n",name,fromList);
            synchronized (fromList){
                System.out.printf("[%s] fmList(%s) was locked.%n",name,fromList);
                str = fromList.get(0);
                System.out.printf("[%s] %s <- fmList(%s).%n",name,str,fromList);
                Thread.sleep(500L);

                System.out.printf("[%s] attempt to lock toList(%s).%n",name,toList);
                synchronized (toList){
                    System.out.printf("[%s] toList(%s) was locked.%n",name,toList);
                    toList.add(str);
                    System.out.printf("[%s] %s -> toList(%s).%n",name,str,toList);
                }
            }
        }catch(InterruptedException e){
            e.printStackTrace();
        } finally{
            System.out.printf("[%s] finished.%n",name);
        }
    }
}

実行結果

結果が200万にとどかない

[thread2] started 
[thread1] started
[thread2] started 
[thread1] started
result:1956006

[GitEngHar]

Tips 他パターン

• HashMap
  • put や put と remove を繰り返すと無限ループが発生する可能性がある
  • これがサーバで発生すると 常時20%のCPU消費率だったものが、急激に100%に跳ね上がったりする
• ArrayList
  • iterator でリストを処理している最中に add や remove を対象リストに実施すると、Iterator処理直前に ConcurrentModificationException が発生する
  • iterator が フェイルファスト(リスト処理でエラーがあった場合に強制終了)
  • ArrayList は スレッドセーフではない
  • iterator は リストを逐次取得する処理だが、内部に配列の位置カーソルをもっており、処理中にこれを add や remove でずらされるとエラーが出るという仕組みである
• longの値を入れ替え
  • 1 と -1 の値をマルチスレッドで数百万回代入処理を行うと、32bitの 1 -1 の組み合わせ次第で 4294967295 等に値が変化する事象が見られる

上記を避けるために、使用するクラスがスレッドセーフ を扱うかどうかを確認する必要がある

[GitEngHar]

スレッドセーフを実現する

ステートレスにする

クラス変数 や インスタンス変数を持たないようにする
どうしても必要な場合にのみ、利用することでマルチスレッド問題を極力回避する

ステートレスにする手段

メソッドの引数を活用する

常に保持する必要のない変数はメソッドの引数に渡すようにする
◆ Before

public class Bedore(){
  String name;
  int old;
  public void readyName(){
    this.name = "haru";
    this.old = 12;
    hello(); 
  }
  public void hello(){
    System.out.printf("%s : %d old",this.name,this.old);
  }
}

◆ After

public class Bedore(){
  public void readyName(String name , int old){
    hello(name,old); 
  }
  public void hello(String name , int old){
    System.out.printf("%s : %d old",this.name,this.old);
  }
}

(ロジック非同期処理) 処理結果をインスタンス変数へ格納しアクセサメソッドを用いて取得

ロジック処理そのものを非同期処理として実行する

◆ ロジックの操作

public static void callbackSample(){
        ExecutorService executor = Executors.newSingleThreadExecutor();
        AsyncProcess proc = new AsyncProcess(
            new AsyncCallback() {
                public void notify(String message){
                    System.out.println("callback message" + message);
                    executor.shutdown();
                }
        });
        executor.execute(proc);
        System.out.println("AsyncProcess is started.");
    }

◆ ロジック処理

public class AsyncProcess implements Runnable {
    private AsyncCallback callback;

    public AsyncProcess(AsyncCallback callback){
        this.callback = callback;
    }

    public void run(){
        try{
            Thread.sleep(1000);
            this.callback.notify("Finished");
        }catch(InterruptedException e){
            e.printStackTrace();
        }
        System.out.println("AsyncProcess finished");
    }
}

◆ Interface

public interface AsyncCallback {
    void notify(String message);
}

Future を用いて get で処理を実行する

public static void futureSample(){
        ExecutorService exec = Executors.newSingleThreadExecutor();
        Future<String> future = exec.submit(new Callable<String>() {
            public String call(){
                try{
                    Thread.sleep(1000);
                }catch(InterruptedException ex){
                    return "Error";
                }
                return "Finished";
            }
        });

        System.out.println("Exec Service is fanished");
        try{
            String message = future.get();
            System.out.println("Exec is finished : message = " + message);
        }catch(InterruptedException | ExecutionException ex){
            ex.printStackTrace();
        }finally{
            exec.shutdown();
        }

    }

future.get 時に実行される