Javaマルチスレッドの生産者と消費者モデル

生産者と消費者モデル

ソフトウェア開発の過程で、データを生成するコードモジュールがあり、データを処理するコードモジュールがあり、生産者と消費者モデルでは、データを生成するモジュールを生産者と呼び、データを処理するモジュールを消費者と呼ぶ.もう1つのストレージエリアは生産者と消費者が接続されており、生産者が生産したデータはストレージエリアに置かれ、消費者がデータを取るのもストレージエリアから取りに行きます.このストレージエリアには最大容量制限があり、ストレージエリアのデータが最大容量に達した場合、生産者に生産データを一時停止させ、容量があって生産を行っています.ストレージにデータがない場合は、データが消費されるまで、消費者にデータの処理を一時停止させます.生産者と消費者モデルが解決しなければならない問題は、生産者と消費者の間でデータを処理する動的バランスの問題である.
例えば、今男の子(生産者)、女の子(消費者)、バスケット(ストレージエリア)、バスケットにりんご(データ)が入っています.かごに最大30個のりんごを入れることができると仮定すると、次の問題は男の子がかごにりんごを入れることと女の子がかごからりんごを取る動的なバランスの問題です.次にコードによって実装するが,生産者と消費者の実装には主に2つの方式があり,1つはwaitとnotify,1つはブロックキューBlockingQueueによって実装する.

方式1:waitとnotify方式

バスケット(ストレージ)コード:

package cn.znh.procon;

import java.util.ArrayList;
import java.util.List;

/**
 *  
 * 
 * @author znh
 *
 */
public class Basket {

    //  
    public static List basket = new ArrayList<>();

}

男の子(生産者)コード:

package cn.znh.procon;

/**
 *  
 * 
 * @author znh
 *
 */
public class Boy extends Thread {

    @Override
    public void run() {
        while (true) {
            synchronized (Basket.basket) {

                //  ， 
                if (Basket.basket.size() >= 30) {
                    try {
                        Basket.basket.wait();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }

                //  
                Basket.basket.add(" ");
                System.out.println(" ， " + Basket.basket.size() + " ");

                //  add ， ， 
                Basket.basket.notify();
            }

            //  ， 
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

女の子(消費者)コード:

package cn.znh.procon;

/**
 * 
 *  
 * 
 * @author znh
 *
 */
public class Girl extends Thread {

    @Override
    public void run() {
        while (true) {
            synchronized (Basket.basket) {

                //  ， 
                if (Basket.basket.size() <= 0) {
                    try {
                        Basket.basket.wait();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }

                //  
                Basket.basket.remove(0);
                System.out.println(" ， " + Basket.basket.size() + " ");

                //  remove ， ， 
                Basket.basket.notify();
            }

            //  ， 
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

実行コード:

public static void main(String[] args) {

        // 
        Boy boy = new Boy();
        Girl girl = new Girl();

        // 
        boy.start();
        girl.start();
    }

方式2:ブロックキュー方式

バスケット(ストレージ)コード:

//  ， 30
BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(30);

男の子(生産者)コード:

package cn.znh.procon.queue;

import java.util.concurrent.BlockingQueue;

/**
 *  
 * 
 * @author znh
 *
 */
public class Boy extends Thread {

    //  （ ）
    private BlockingQueue blockingQueue;

    public Boy(BlockingQueue blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    @Override
    public void run() {
        while (true) {

            //  
            try {
                blockingQueue.put(" ");
                System.out.println(" ， " + blockingQueue.size() + " ");
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }

            //  ， 
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

女の子(消費者)コード:

package cn.znh.procon.queue;

import java.util.concurrent.BlockingQueue;

/**
 * 
 *  
 * 
 * @author znh
 *
 */
public class Girl extends Thread {

    //  （ ）
    private BlockingQueue blockingQueue;

    public Girl(BlockingQueue blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    @Override
    public void run() {
        while (true) {

            //  
            try {
                blockingQueue.take();
                System.out.println(" ， " +blockingQueue.size() + " ");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            //  ， 
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

実行コード:

public static void main(String[] args) {

        //  ， 30
        BlockingQueue blockingQueue = new ArrayBlockingQueue<>(30);

        //  
        Boy boy = new Boy(blockingQueue);
        Girl girl = new Girl(blockingQueue);

        //  
        boy.start();
        girl.start();
    }

上記は2つの方式で実現された生産者と消費者モデルのコードであり、2つの方式の対比から見ると、それらが達成した効果は同じであり、データ処理のダイナミックバランスを維持しているが、方式2は方式1よりも簡単に使用されている.