Java多线程编程实战—利用Condition来实现阻塞队列

[TOC]

欢迎关注作者简书
csdn传送门

  假设有一个固定大小的缓冲区（缓冲区是队列，遵守FIFO），支持存和取方法。如果缓冲区为空时尝试取数据，那么线程将阻塞，直到缓冲区有一个可用的数据；如果缓冲区满了，这时尝试写数据，那么线程将被阻塞直到有可用空间。我们希望有两个独立的等待集(Condition阻塞队列)，一个放置存数据的等待线程，一个放置取数据的等待线程，这样，当缓冲区有空位置的时侯，可以到“存数据的线程等待集”中唤醒一个线程存数据；当缓冲区从一无所有到有数据存入时，可以到“取数据的线程等待集”中唤醒一个线程取数据。这个关系不能搞错！
Condition是由Lock创建的，每个Lock锁可以创建多个Condition对象。只有同一把锁的Condition对象可以相互影响.

---

package com.concurrency.chapter16;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.IntStream;

/**
 * @program: springcloud-zzx
 *
 * @description:
 *
 * @author: zhouzhixiang
 *
 * @create: 2019-05-15 19:18
 */
public class ConditionDemo {
    static class BoudedBuffer {
        final Lock lock = new ReentrantLock();
        final Condition notEmpty = lock.newCondition();
        final Condition notFull = lock.newCondition();
        private Object[] buffer = new Object[5];
        private int input_index, take_index, count;

        public void put(Object x) {
            lock.lock();
            try {
                System.out.println(Thread.currentThread().getName()+" ready to put .....");
                while (count == buffer.length) {
                    System.err.println("put data to buffer["+input_index+"] : error : buffer full cannot insert ....");
                    notFull.await();
                }
                buffer[input_index] = x;
                System.out.println("put data to buffer["+input_index+"] : success : buffer insert data" + String.valueOf(x));
                if (++input_index == buffer.length) {
                    input_index = 0;
                }
                count++;
                notEmpty.signal();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }

        public Object take() {
            lock.lock();
            try {
                System.out.println(Thread.currentThread().getName()+" ready to take .....");
                while (count == 0) {
                    System.err.print("take from buffer["+take_index+"] : error : buffer is empty now");
                    notEmpty.await();
                }
                System.out.println("take from buffer["+take_index+"] : success : take data is " + String.valueOf(buffer[take_index]));
                if (++take_index == buffer.length) {
                    take_index = 0;
                }
                count--;
                notFull.signal();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
            return buffer[take_index];
        }

    }

    public static void main(String[] args) {
        final BoudedBuffer bf = new BoudedBuffer();
        IntStream.rangeClosed(1, 10).forEach(i -> {
            new Thread(new Runnable() {
                @Override public void run() {
                    bf.put("zhouzhixiang["+i+"]");
                }
            }).start();
        });

        IntStream.rangeClosed(1, 5).forEach(i -> {
            new Thread(new Runnable() {
                @Override public void run() {
                    bf.take();
                }
            }).start();
        });

    }
}

Thread-0 ready to put .....
put data to buffer[0] : error : buffer full cannot insert ....
put data to buffer[0] : success : buffer insert datazhouzhixiang[1]
put data to buffer[0] : error : buffer full cannot insert ....
Thread-5 ready to put .....
put data to buffer[0] : error : buffer full cannot insert ....
put data to buffer[1] : success : buffer insert datazhouzhixiang[6]
put data to buffer[0] : error : buffer full cannot insert ....
Thread-1 ready to put .....
put data to buffer[0] : error : buffer full cannot insert ....
put data to buffer[2] : success : buffer insert datazhouzhixiang[2]
Thread-2 ready to put .....
put data to buffer[3] : success : buffer insert datazhouzhixiang[3]
Thread-3 ready to put .....
put data to buffer[4] : success : buffer insert datazhouzhixiang[4]
Thread-4 ready to put .....
Thread-6 ready to put .....
Thread-7 ready to put .....
Thread-9 ready to put .....
Thread-8 ready to put .....
Thread-10 ready to take .....
take from buffer[0] : success : take data is zhouzhixiang[1]
put data to buffer[0] : success : buffer insert datazhouzhixiang[5]
Thread-12 ready to take .....
take from buffer[1] : success : take data is zhouzhixiang[6]
Thread-11 ready to take .....
take from buffer[2] : success : take data is zhouzhixiang[2]
put data to buffer[1] : success : buffer insert datazhouzhixiang[7]
Thread-13 ready to take .....
take from buffer[3] : success : take data is zhouzhixiang[3]
Thread-14 ready to take .....
take from buffer[4] : success : take data is zhouzhixiang[4]
put data to buffer[2] : success : buffer insert datazhouzhixiang[8]
put data to buffer[3] : success : buffer insert datazhouzhixiang[10]
put data to buffer[4] : success : buffer insert datazhouzhixiang[9]

本文参考 使用 ReentrantLock 和 Condition 实现一个阻塞队列

<center>欢迎加入Java猿社区！</center>
<center>免费领取我历年收集的所有学习资料哦！</center>

欢迎加入Java猿社区.png