Java中常用的集合框架

时间：2018-02-05 10:44:22 阅读：186 评论：0 收藏：0 [点我收藏+]

标签：return 线程安全第一个没有 throws rds 实现 sdn cal

BlockingQueue

A线程可以知道b线程的存在

是一个接口并非一个具体实现：

ArrayBlockingQueue

ArrayBlockingQueue的内部元素都放置在一个对象数组中：final Object[] items;

Offer():当队列已经满了，会立即返回false

Put():如果队列满了会一直等待

Pool()：弹出元素，如果为空返回null

Take():弹出元素，如果为空等待到有元素即可。

Take方法：

public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();
            return extract();
        } finally {
            lock.unlock();
        }
    }
    
private void insert(E x) {
        items[putIndex] = x;
        putIndex = inc(putIndex);
        ++count;
        notEmpty.signal();
    }

Put方法：

public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length)
                notFull.await();
            insert(e);
        } finally {
            lock.unlock();
        }
    }

/**
     * Extracts element at current take position, advances, and signals.
     * Call only when holding lock.
     */
    private E extract() {
        final Object[] items = this.items;
        E x = this.<E>cast(items[takeIndex]);
        items[takeIndex] = null;
        takeIndex = inc(takeIndex);
        --count;
        notFull.signal();
        return x;
    }

LinkedBlockingQueue(锁分离)

两把不同的锁
/** Lock held by take, poll, etc */
    private final ReentrantLock takeLock = new ReentrantLock();
 
    /** Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();
 
    /** Lock held by put, offer, etc */
    private final ReentrantLock putLock = new ReentrantLock();
 
    /** Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();

 
Take函数
public E take() throws InterruptedException {
        E x;
        int c = -1;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();//不能有两个线程同时取数据
        try {
            while (count.get() == 0) {//如果没有数据，一直等待（因为是lockInterruptibly，可中断）
                notEmpty.await();
            }
            x = dequeue();//取得第一个数据
            c = count.getAndDecrement();//数量-1，原子操作，因为会和put同时访问count。
            if (c > 1)
                notEmpty.signal();//通知其他take操作
        } finally {
            takeLock.unlock();//释放锁
        }
        if (c == capacity)
            signalNotFull();//通知put操作，已有空余空间
        return x;
    }

 
Put函数
public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        // Note: convention in all put/take/etc is to preset local var
        // holding count negative to indicate failure unless set.
        int c = -1;
        Node<E> node = new Node(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();//上锁不能有两个线程同时进行put函数
        try {
            /*
             * Note that count is used in wait guard even though it is
             * not protected by lock. This works because count can
             * only decrease at this point (all other puts are shut
             * out by lock), and we (or some other waiting put) are
             * signalled if it ever changes from capacity. Similarly
             * for all other uses of count in other wait guards.
             */
            while (count.get() == capacity) {//当队列已经满了以后，等待
                notFull.await();
            }
            enqueue(node);//插入数据
            c = count.getAndIncrement();//更新总数
            if (c + 1 < capacity)
                notFull.signal();//有足够的空间，通知其他线程
        } finally {
            putLock.unlock();//释放锁
        }
        if (c == 0)
            signalNotEmpty();//释放成功后，通知take函数取数据
    }