マルチスレッド

基本概念は

プロセスプロセス

プロセスは、比較的独立したメモリ空間とシステムリソースを持つ実行中のプログラムインスタンスです。オペレーティングシステムは、CPU 上で実行される複数のプロセスをスケジュールする責任がある。

Threadsより

スレッドはプロセス内の実行単位です。1つのプロセスは、ヒープメモリとプロセス内のリソースの一部を共有する複数のスレッドを含むことができるが、各スレッドは独自のプログラムカウンタ、仮想マシンスタックなどのスレッドプライベートデータを持つ。

複数のスレッドは通常、シングルコアCPUでは交互に実行され、マルチコアCPUでは実際に並列に実行される。スレッドの具体的なスケジューリングはオペレーティングシステムとJVMの協力によって行われ、プログラムは固定された実行順序に依存しません。

並行と並行

並行性は、複数のタスクが一定期間にわたって交互に進行することを意味する。
並列性は、複数のタスクが異なるプロセッサコアによって同時に実行されることを意味する。

スレッドの作成→

Threadクラスの継承

Threadサブクラスを定義し、run()をオーバーライドし、start()を呼び出してスレッドを起動します。

public class MyThread extends Thread {
    public MyThread(String name) {
        super(name);
    }

    @Override
    public void run() {
        for (int i = 1; i <= 10; i++) {
            System.out.println(getName() + " 执行第 " + i + " 次任务");
        }
    }
}

public class Main {
    public static void main(String[] args) {
        Thread thread1 = new MyThread("A");
        Thread thread2 = new MyThread("B");
        Thread thread3 = new MyThread("C");

        thread1.start();
        thread2.start();
        thread3.start();
    }
}

run()の直接呼び出しは通常のメソッド呼び出しであり、新しい実行スレッドは作成されません。Threadオブジェクトはstart()を1 回だけ呼び出すことができます。

Runnableインターフェイスの実装

スレッドオブジェクトからタスクを分離するのが一般的な方法です。タスククラスはRunnableを実現し，タスクオブジェクトをThreadに渡す.

public class TicketTask implements Runnable {
    private final String stationName;

    public TicketTask(String stationName) {
        this.stationName = stationName;
    }

    @Override
    public void run() {
        for (int i = 1; i <= 10; i++) {
            System.out.println(
                    stationName + "，窗口 "
                            + Thread.currentThread().getName()
                            + " 执行第 " + i + " 次任务"
            );
        }
    }
}

public class Main {
    public static void main(String[] args) {
        Runnable task1 = new TicketTask("哈西站");
        Runnable task2 = new TicketTask("哈尔滨站");

        new Thread(task1, "A").start();
        new Thread(task1, "B").start();
        new Thread(task2, "C").start();
    }
}

Runnableを使用する利点は以下のとおりです。

タスククラスは他のクラスから継承できます。
同じタスクオブジェクトを複数のスレッドに渡すことができ、タスクの状態を簡単に共有できます。
タスクはスレッドのライフサイクルから分離されており、スレッドプール管理が容易です。

タスクオブジェクトを共有すると、その中の可変フィールドも共有されるため、スレッドセーフを考慮する必要があります。

スレッドライフサイクル

Java Thread.Stateは6つのスレッド状態を定義する。

NEWスレッドオブジェクトは作成されましたが、startがまだ呼び出されていません。
RUNNABLEスレッドが実行中であるか、すでに実行条件が整っており、CPUスケジュールを待っている。
BLOCKED：あるsynchronizedモニタに入るのを待ちます。
WAITING他のスレッドが特定の操作を実行するのを無期限に待機します。たとえば、waitやjoinなどです。
TIMED_WAITING指定された時間の間待機します。たとえば、sleep、タイムアウト付きのwait、joinなどです。
TERMINATED：run 正常終了，もしくは異常未捕獲で終了.

Java 状態には個別のRUNNING列挙はなく、実行中のスレッドもRUNNABLEに属します。

スレッドの一般的な方法

currentThread()

現在実行中のコードのスレッドオブジェクトを返します。

System.out.println(Thread.currentThread().getName());

sleep()

現在のスレッドを少なくとも指定時間スリープさせ、TIMED_WAITINGに移行します。実際のリカバリ時間が指定された時間よりも後になる場合があります。

try {
    Thread.sleep(1000);
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
}

割り込みを受信すると、sleep()はInterruptedExceptionをスローし、割り込みフラグをクリアします。処理を続行できない場合は、例外のみを印刷するのではなく、通常、中断状態を再開してください。

タイムタスクは無限ループ延長時間sleep()ではなく、ScheduledExecutorServiceを推奨します。

ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
scheduler.scheduleAtFixedRate(
        () -> System.out.println("执行备份任务"),
        0,
        24,
        TimeUnit.HOURS
);

毎日現地時間に固定された実行が必要な場合は、タイムゾーンと夏時間を考慮して、現在の時間に基づいて最初の遅延を計算する必要があります。

yield()

Thread.yield()は、現在のスレッドが実行機会を割譲する意思があることをスケジューラに通知するが、スケジューラはこの通知を無視することができる。信頼性の高いスレッドコラボレーションを実現するためには使用できません。

join()

現在のスレッドは、別のスレッドのjoin()を呼び出した後、ターゲットスレッドが終了するのを待ちます。

public class Main {
    public static void main(String[] args) {
        Runnable task = () -> System.out.println(
                Thread.currentThread().getName() + " 完成任务"
        );

        Thread thread1 = new Thread(task, "A");
        Thread thread2 = new Thread(task, "B");
        Thread thread3 = new Thread(task, "C");

        thread1.start();
        thread2.start();
        thread3.start();

        try {
            thread1.join();
            thread2.join();
            thread3.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return;
        }

        System.out.println("所有任务执行结束");
    }
}

デーモンスレッド

setDaemon(true)を呼び出すと、スレッドが起動する前にデーモンとして設定されます。

Thread worker = new Thread(() -> {
    while (!Thread.currentThread().isInterrupted()) {
        // 提供后台服务
    }
});
worker.setDaemon(true);
worker.start();

JVMにデーモンスレッドしか残っていない場合、JVMは終了できます。デーモンスレッドは通常のスレッドにバインドしてガードしたり、必要な保存やクリーンアップを実行するために依存したりすることはできません。

スレッドの優先順位

setPriority() Thread.MIN_PRIORITY ~ Thread.MAX_PRIORITYの範囲で、スレッド優先度のヒントを設定します。オペレーティングシステムやJVMによって優先順位の取り扱いが異なり、実行順序や公平性を保証することはできません。

スレッドセーフと競合条件

複数のスレッドが同時に可変データを読み書きし共有する場合、操作が適切に同期されていない場合、結果は予測不可能な実行順序に依存する可能性があります。

例えば、money += 1000は不可分の単一ステップではなく、通常は読み取り、計算、書き込みバックを含む。

synchronized 关键字

synchronizedでは、オブジェクトモニタを使用して相互排他性と可視性を実現します。同期ゾーンに入る前に、スレッドは対応するモニターを取得する必要があります。同期ゾーンを終了すると、モニターを解放します。

同期インスタンスメソッド{{どうきいんすたんすめそっど}}

同期インスタンスメソッドは、現在のオブジェクト、つまりthisをロックします。

public class Account {
    private int money = 20_000;

    public synchronized void deposit(int amount) {
        money += amount;
        System.out.println("存款后余额：" + money);
    }

    public synchronized void withdraw(int amount) {
        if (money < amount) {
            throw new IllegalStateException("余额不足");
        }
        money -= amount;
        System.out.println("取款后余额：" + money);
    }
}

同じオブジェクトをロックする同期コード間だけが相互排他的です。2つのスレッドが2つの異なるAccountオブジェクトのメソッドを呼び出す場合、同じロックを使用しません。

コードブロックの同期

同期コードブロックは、ロックの範囲を絞り込み、ロックオブジェクトを明示的に指定します。

public void deposit(int amount) {
    synchronized (this) {
        money += amount;
        System.out.println("存款后余额：" + money);
    }
}

ロックオブジェクトは、関連するすべてのスレッドがアクセスできる安定したオブジェクトでなければなりません。

同期静的メソッド{{どうきstaticめそっど}}

同期静的メソッドは、現在のクラスに対応するClassオブジェクトをロックします。

public static synchronized void updateGlobalCount() {
    count++;
}

以下と同じロックを使用しています：

synchronized (Account.class) {
    count++;
}

インスタンス同期メソッドは静的同期メソッドとは異なるモニタを使用し、デフォルトでは互いにブロックしません。

sleepとロック

スレッドが同期領域でsleep()を呼び出したときにすでに保持しているモニタを解放しないため、ロック中に不必要に長い待ち時間や時間のかかるIOを行うべきではありません。

synchronizedの性能

最新のJVMではロックに様々な最適化が行われており、実装ポリシーはJDKのバージョンによって異なります。synchronizedを固定された“ヘビー級ロック”とみなすことはできず、ビジネスロジックを書くために特定のバージョンのバイアスや軽量ロックなどの内部フェーズに依存してはいけません。

CASは比較と交換に基づくアトミック更新の考え方であり、AtomicIntegerのようなオリジナルのサブクラスは部分的なロックフリー操作を実装するためにこれまたは同等のメカニズムを使用する。CASはすべてのロックを置き換えることはできず、複雑な複合状態には適切な同期スキームが依然として必要です。

デッドロック

デッドロックは、複数のスレッドが互いのリソースを待機し、関連するすべてのスレッドが実行を続行できなくなることです。

public class DeadLockTask implements Runnable {
    private static final Object LOCK_A = new Object();
    private static final Object LOCK_B = new Object();
    private final boolean firstA;

    public DeadLockTask(boolean firstA) {
        this.firstA = firstA;
    }

    @Override
    public void run() {
        if (firstA) {
            synchronized (LOCK_A) {
                synchronized (LOCK_B) {
                    System.out.println("A -> B");
                }
            }
        } else {
            synchronized (LOCK_B) {
                synchronized (LOCK_A) {
                    System.out.println("B -> A");
                }
            }
        }
    }
}

デッドロックの回避

すべてのスレッドは同じ順序で複数のロックを取得します。
ロックのネストと保持時間を短縮します。
外部呼び出し、ネットワークIO、または制御不能なブロッキング操作をロックしないでください。
タイムアウトに合わせてLock.tryLock()を使用して、フェッチに失敗した場合にすでに保持しているロックを解放して再試行する。
スレッドダンプ、監視ツール、およびThreadMXBeanを使用してデッドロックを検出します。

通常synchronizedロック待ちは割込みによって直接キャンセルできないため，“デッドロック発見後にスレッドを割込み”がデッドロックを解消するとは限らない.

wait、notify、およびnotifyAll

これら3つのメソッドはObjectクラスで定義されており、同じオブジェクトモニタに基づくスレッドコラボレーション用です。

ルールの利用

呼び出し前にオブジェクトのモニターを保持する必要があります。そうでない場合は、IllegalMonitorStateExceptionがスローされます。
wait()は、現在のスレッドを待機状態にし、オブジェクトモニタを解放します。
notify()は、このモニタ待機セット内のスレッドをウェイクアップします。具体的なスレッドは不明です。
notifyAll()はこのモニタで待機しているすべてのスレッドをウェイクアップします。
ウェイクアップされたスレッドは、wait()から戻る前に、モニタを再競合する必要があります。

スレッドが誤ってウェイクアップしたり、ロックを取り戻す前に条件が再び変更される可能性があるため、条件判定にはwhileを使用してください。

waitとsleepの违い

対照	`wait()`	`sleep()`
所属するクラス	`Object`	`Thread`
モニターが必要ですか？	はい。	いいえ、いいえ
モニターを解放するか	はい。	いいえ、いいえ
目覚めの方法。	通知、中断、タイムアウト	タイムアウトまたは中断

生産者と消費者

次に、容量に制限のあるリソースカウントを使用して、スレッドコラボレーションを示します。

public class Resource {
    private final int capacity = 20;
    private int count;

    public synchronized void produce() {
        while (count == capacity) {
            try {
                wait();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            }
        }

        count++;
        System.out.println("生产一个，当前数量：" + count);
        notifyAll();
    }

    public synchronized void consume() {
        while (count == 0) {
            try {
                wait();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            }
        }

        count--;
        System.out.println("消费一个，当前数量：" + count);
        notifyAll();
    }
}

Resource resource = new Resource();

Runnable producer = () -> {
    for (int i = 0; i < 100; i++) {
        resource.produce();
    }
};

Runnable consumer = () -> {
    for (int i = 0; i < 100; i++) {
        resource.consume();
    }
};

new Thread(producer, "生产者 1").start();
new Thread(producer, "生产者 2").start();
new Thread(consumer, "消费者 1").start();
new Thread(consumer, "消费者 2").start();

実際のプロジェクトでは、待機、通知、容量制御をカプセル化した生産者コンシューマモデルが一般的に使用されるBlockingQueueです。

BlockingQueue<String> queue = new ArrayBlockingQueue<>(20);
queue.put("task");
String task = queue.take();

スレッドプールThread pool

スレッドの作成と破棄が頻繁に行われると、システムリソースが消費される。スレッドプールワーカースレッドの再利用、同時実行数の制限、タスクキューと拒否ポリシーの統合管理。

少数の明確なライフサイクルを持つスレッドは直接作成できます。多数の短いタスクを継続的に処理する必要があるサーバプログラムは、通常、スレッドプールを使用します。

ThreadPoolExecutor

コアパラメータ

ThreadPoolExecutorの主な構造パラメータは以下の通りである。

corePoolSize：コアスレッド数.
maximumPoolSize：作成可能なスレッドの最大数。
keepAliveTime：コア以外のアイドルスレッドの生存時間。
unit：生存時間の単位。
workQueueコアスレッドがビジーになった後に保留中のタスクを保存するキュー。
threadFactoryワーカースレッドを作成するファクトリ。
handler：スレッド数が上限に达し、キューが満杯になったの拒否ポリシー.

タスクの送信順序

execute()をコールしてタスクを発行する場合の主な処理は次のとおりです。

現在のワーカースレッドがコアスレッド数より少ない場合、コアスレッドを作成してタスクを実行します。
コアスレッドが動作している間、タスクをキューに追加しようとする。
キューがいっぱいで、スレッド数が最大スレッド数より少ない場合、タスクを実行するために非コア·スレッドを作成します。
キューが満杯で、スレッド数がすでに最大値に达したに拒否ポリシーが実行されます。
非コアスレッドは、生存時間を超えてアイドル状態になるとリサイクルできます。デフォルトでは、コア·スレッドのタイムアウトが有効になっていない限り、コア·スレッドはアイドル状態で再利用されません。

拒否ポリシー

AbortPolicy RejectedExecutionExceptionを投げます。
CallerRunsPolicy：タスクをコミットした呼び出しスレッドがタスクを実行し、バックプレッシャーを提供します。
DiscardOldestPolicy：キュー内で最も待機しているタスクを破棄し、現在のタスクをコミットしようとする。
DiscardPolicy：現在のタスクを直接破棄します。

タスクを破棄すると、ビジネスデータが失われる可能性があります。タスクの重要性に応じてポリシーを選択し、監視情報を記録する必要があります。

カスタム·スレッド·プール例

public class TestTask implements Runnable {
    private final int number;

    public TestTask(int number) {
        this.number = number;
    }

    @Override
    public void run() {
        try {
            Thread.sleep(1000);
            System.out.println(
                    Thread.currentThread().getName()
                            + " 执行任务 " + number
            );
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

ThreadPoolExecutor executor = new ThreadPoolExecutor(
        2,
        5,
        30,
        TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(2),
        Executors.defaultThreadFactory(),
        new ThreadPoolExecutor.CallerRunsPolicy()
);

for (int i = 0; i < 10; i++) {
    executor.execute(new TestTask(i));
}

executor.shutdown();

shutdown()は新しいタスクを受信しなくなりましたが、すでに送信されたタスクは継続します。shutdownNow()は、実行中のタスクを中断し、まだ開始されていないキュータスクを返しますが、タスクがすぐに停止する保証はありません。

Executorsが提供する一般的なスレッドプール

Executorsは便利なファクトリ方式を提供するが、一部の方式では無限キューや近似無限スレッド数を使用する。サーバプログラムは内部パラメータを認識し、負荷に応じてスレッドプールを明示的に構成する必要があります。

FixedThreadPool

固定数のワーカースレッドを作成します。通常は無限キューを使用して余分なタスクを保存します。

ExecutorService executor = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10; i++) {
    executor.execute(new TestTask(i));
}
executor.shutdown();

タスクのコミット速度が処理速度を上回ると、無限キューが成長し続け、大量のメモリを消費する可能性があります。

CachedThreadPool

タスクの量に応じてスレッドが作成および再利用され、アイドルスレッドは通常約60秒間保持されます。最大スレッド数は非常に大きく、実行時間が短く、コミット量が制御可能なタスクに適しており、並行性の上限がない外部リクエストには適していません。

ExecutorService executor = Executors.newCachedThreadPool();
executor.execute(new TestTask(1));
executor.shutdown();

ScheduledThreadPool

遅延タスクおよび周期タスクに使用されます。

ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
executor.schedule(
        new TestTask(1),
        5,
        TimeUnit.SECONDS
);
executor.shutdown();

周期タスクにはscheduleAtFixedRate()またはscheduleWithFixedDelay()を使用できますが、タスク実行時間の扱い方は異なります。

SingleThreadExecutor

常に単一のワーカースレッドを使用して、キューの順序でタスクを実行します。ワーカースレッドが例外的に終了すると、スレッドプールは代替スレッドを作成し、後続のタスクを続行します。

ExecutorService executor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 5; i++) {
    executor.execute(new TestTask(i));
}
executor.shutdown();

タスクのシリアル実行が必要なシナリオに適していますが、無限キューはタスクをバックログできます。

WorkStealingPool

newWorkStealingPool()はForkJoinPoolをベースにしており、ワーカースレッドは他のキューからタスクを盗むことができ、比較的独立したコンピューティングタスクに適している。

ExecutorService executor = Executors.newWorkStealingPool(5);
List<Future<?>> futures = new ArrayList<>();

for (int i = 0; i < 10; i++) {
    futures.add(executor.submit(new TestTask(i)));
}

for (Future<?> future : futures) {
    try {
        future.get();
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        break;
    } catch (ExecutionException e) {
        e.getCause().printStackTrace();
    }
}

executor.shutdown();

タスクの実行順序は固定されません。メインスレッドにsleep()を長時間放置してスレッドプールタスクの終了を待たせないでください。Future、awaitTermination()、またはその他の明示的な同期方法を使用してください。