zoukankan      html  css  js  c++  java

  • java生产者消费者问题代码分析

    作者要的是一个生产者生成，接着必须有一个消费者消费，那这不是需要单线程吗？或者使用1个大小的阻塞队列。所以只谈论问题本身，不谈论好不好。

     

    具体代码：

    Java代码 复制代码  收藏代码
    1. import java.util.concurrent.locks.Condition;   
    2.   
    3. import java.util.concurrent.locks.Lock;   
    4.   
    5. import java.util.concurrent.locks.ReentrantLock;   
    6.   
    7.   
    8. //生产/消费者模式   
    9.   
    10. public class Basket {   
    11.   
    12.     Lock lock = new ReentrantLock();   
    13.   
    14.   
    15.     // 产生Condition对象   
    16.   
    17.     Condition produced = lock.newCondition();   
    18.   
    19.     Condition consumed = lock.newCondition();   
    20.   
    21.     boolean available = false;   
    22.   
    23.   
    24.     public void produce() throws InterruptedException {   
    25.   
    26.         lock.lock();   
    27.   
    28.   
    29.         try {   
    30.   
    31.             if (available) {   
    32.   
    33.                 produced.await(); // 放弃lock进入睡眠   
    34.   
    35.             }   
    36.   
    37.   
    38.             System.out.println("Apple produced.");   
    39.   
    40.   
    41.             available = true;   
    42.   
    43.   
    44.             consumed.signal(); // 发信号唤醒等待这个Condition的线程   
    45.   
    46.         } finally {   
    47.   
    48.             lock.unlock();   
    49.   
    50.         }   
    51.   
    52.     }   
    53.   
    54.   
    55.     public void consume() throws InterruptedException {   
    56.   
    57.         lock.lock();   
    58.   
    59.   
    60.         try {   
    61.   
    62.             if (!available) {   
    63.   
    64.                 consumed.await(); // 放弃lock进入睡眠   
    65.   
    66.             }   
    67.   
    68.   
    69.             /* 吃苹果 */  
    70.   
    71.             System.out.println("Apple consumed.");   
    72.   
    73.   
    74.             available = false;   
    75.   
    76.   
    77.             produced.signal(); // 发信号唤醒等待这个Condition的线程   
    78.   
    79.         } finally {   
    80.   
    81.             lock.unlock();   
    82.   
    83.         }   
    84.   
    85.     }   
    86.   
    87. }  
    import java.util.concurrent.locks.Condition;

import java.util.concurrent.locks.Lock;

import java.util.concurrent.locks.ReentrantLock;


//生产/消费者模式

public class Basket {

    Lock lock = new ReentrantLock();


    // 产生Condition对象

    Condition produced = lock.newCondition();

    Condition consumed = lock.newCondition();

    boolean available = false;


    public void produce() throws InterruptedException {

        lock.lock();


        try {

            if (available) {

                produced.await(); // 放弃lock进入睡眠

            }


            System.out.println("Apple produced.");


            available = true;


            consumed.signal(); // 发信号唤醒等待这个Condition的线程

        } finally {

            lock.unlock();

        }

    }


    public void consume() throws InterruptedException {

        lock.lock();


        try {

            if (!available) {

                consumed.await(); // 放弃lock进入睡眠

            }


            /* 吃苹果 */

            System.out.println("Apple consumed.");


            available = false;


            produced.signal(); // 发信号唤醒等待这个Condition的线程

        } finally {

            lock.unlock();

        }

    }

}

     

    Java代码 复制代码  收藏代码
    1. import java.util.concurrent.ExecutorService;   
    2. import java.util.concurrent.Executors;   
    3.   
    4. //测试用类   
    5. public class ConditionTester {   
    6.   
    7.     public static void main(String[] args) throws InterruptedException {   
    8.         final Basket basket = new Basket();   
    9.   
    10.         // 定义一个producer   
    11.         Runnable producer = new Runnable() {   
    12.             public void run() {   
    13.                 try {   
    14.                     basket.produce();   
    15.                 } catch (InterruptedException ex) {   
    16.                     ex.printStackTrace();   
    17.                 }   
    18.             }   
    19.         };   
    20.   
    21.         // 定义一个consumer   
    22.         Runnable consumer = new Runnable() {   
    23.             public void run() {   
    24.                 try {   
    25.                     basket.consume();   
    26.                 } catch (InterruptedException ex) {   
    27.                     ex.printStackTrace();   
    28.                 }   
    29.             }   
    30.         };   
    31.   
    32.         // 各产生10个consumer和producer   
    33.         ExecutorService service = Executors.newCachedThreadPool();   
    34.   
    35.         for (int i = 0; i < 4; i++)   
    36.             service.submit(consumer);   
    37.   
    38.         Thread.sleep(2000 * 2);   
    39.   
    40.         for (int i = 0; i < 4; i++)   
    41.             service.submit(producer);   
    42.   
    43.         service.shutdown();   
    44.     }   
    45. }  
    import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

//测试用类
public class ConditionTester {

    public static void main(String[] args) throws InterruptedException {
        final Basket basket = new Basket();

        // 定义一个producer
        Runnable producer = new Runnable() {
            public void run() {
                try {
                    basket.produce();
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        };

        // 定义一个consumer
        Runnable consumer = new Runnable() {
            public void run() {
                try {
                    basket.consume();
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        };

        // 各产生10个consumer和producer
        ExecutorService service = Executors.newCachedThreadPool();

        for (int i = 0; i < 4; i++)
            service.submit(consumer);

        Thread.sleep(2000 * 2);

        for (int i = 0; i < 4; i++)
            service.submit(producer);

        service.shutdown();
    }
}

     

    原因分析：

    1、假设前面有2个producer(此时available=true)

    1.1、一个在等待lock

    1.2、一个await

    2、consumer生成内容后，available=false，produced.signal(); 最后lock.unlock();

    3.1、因为lock.unlock所以会触发一个lock获取到锁（虽然signal也会触发等待这个条件的其他线程，但是多线程大家都知道什么时候触发这是不确定的），如果此时正好是[1.1]那么因为available=false，执行完释放锁

    3.2、produced.signal()所以会触发一个await的producer；

     

    解决方案：

    只要保证[3.1]还是需要await即可解决问题

     

    所以加一个 AtomicInteger producedAwaitCounter = new AtomicInteger(0); 统计当前等待的生产者，如果当前available=false，但已经有生产者生成了内容，那么先等待消费者消费了再说

     

                if (available || producedAwaitCounter.get() > 0) {

                    producedAwaitCounter.incrementAndGet();

                    produced.await(); // 放弃lock进入睡眠

                    producedAwaitCounter.decrementAndGet();

                }

     

    当然最简单的是使用：自旋，原理可以自己分析下：

                while (available) {

                    produced.await(); // 放弃lock进入睡眠

                }

     

    Java代码 复制代码  收藏代码
    1. package com.sishuok.es.test;   
    2.   
    3. import java.util.concurrent.atomic.AtomicInteger;   
    4. import java.util.concurrent.locks.Condition;   
    5.   
    6. import java.util.concurrent.locks.Lock;   
    7.   
    8. import java.util.concurrent.locks.ReentrantLock;   
    9.   
    10.   
    11. //生产/消费者模式   
    12.   
    13. public class Basket {   
    14.   
    15.     Lock lock = new ReentrantLock(true);   
    16.   
    17.   
    18. // 产生Condition对象   
    19.   
    20.     Condition produced = lock.newCondition();   
    21.   
    22.     Condition consumed = lock.newCondition();   
    23.   
    24.     boolean available = false;   
    25.     AtomicInteger producedAwaitCounter = new AtomicInteger(0);   
    26.   
    27.   
    28.     public void produce() throws InterruptedException {   
    29.   
    30.         lock.lock();   
    31.   
    32.         try {   
    33.   
    34.             if (available || producedAwaitCounter.get() > 0) {   
    35.                 producedAwaitCounter.incrementAndGet();   
    36.                 produced.await(); // 放弃lock进入睡眠   
    37.                 producedAwaitCounter.decrementAndGet();   
    38.             }   
    39.   
    40.             System.out.println("Apple produced.");   
    41.   
    42.   
    43.             available = true;   
    44.   
    45.   
    46.             consumed.signal(); // 发信号唤醒等待这个Condition的线程   
    47.   
    48.         } finally {   
    49.             lock.unlock();   
    50.         }   
    51.   
    52.     }   
    53.   
    54.   
    55.     public void consume() throws InterruptedException {   
    56.   
    57.         lock.lock();   
    58.   
    59.   
    60.         try {   
    61.   
    62.             if (!available) {   
    63.                 consumed.await(); // 放弃lock进入睡眠   
    64.             }   
    65.   
    66.             /* 吃苹果 */  
    67.   
    68.             System.out.println("Apple consumed.");   
    69.   
    70.   
    71.             available = false;   
    72.   
    73.             produced.signal(); // 发信号唤醒等待这个Condition的线程   
    74.         } finally {   
    75.             lock.unlock();   
    76.         }   
    77.   
    78.     }   
    79.   
    80. }  
    package com.sishuok.es.test;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;

import java.util.concurrent.locks.Lock;

import java.util.concurrent.locks.ReentrantLock;


//生产/消费者模式

public class Basket {

    Lock lock = new ReentrantLock(true);


// 产生Condition对象

    Condition produced = lock.newCondition();

    Condition consumed = lock.newCondition();

    boolean available = false;
    AtomicInteger producedAwaitCounter = new AtomicInteger(0);


    public void produce() throws InterruptedException {

        lock.lock();

        try {

            if (available || producedAwaitCounter.get() > 0) {
                producedAwaitCounter.incrementAndGet();
                produced.await(); // 放弃lock进入睡眠
                producedAwaitCounter.decrementAndGet();
            }

            System.out.println("Apple produced.");


            available = true;


            consumed.signal(); // 发信号唤醒等待这个Condition的线程

        } finally {
            lock.unlock();
        }

    }


    public void consume() throws InterruptedException {

        lock.lock();


        try {

            if (!available) {
                consumed.await(); // 放弃lock进入睡眠
            }

            /* 吃苹果 */

            System.out.println("Apple consumed.");


            available = false;

            produced.signal(); // 发信号唤醒等待这个Condition的线程
        } finally {
            lock.unlock();
        }

    }

}

     
  • 相关阅读:
    Unity HDRP BentNormal的理解
    c语言变长数组VLA的变通实现
    中间件目录索引：redis，git，grpc等
    MYSQL插入脚本
    Polly是一个.NET弹性和瞬态故障处理库
    grpc的.net core使用
    基于PaddleOCR实现AI发票识别的Asp.net Core应用
    Clean Architecture For RazorPage 实现多语言和本地化
    easyui-datagrid 主从表(一对多)表结构，明细在前端存json，一键保存至数据库
    下拉框级联
  • 原文地址：https://www.cnblogs.com/snake-hand/p/3155482.html
Copyright © 2011-2022 走看看