1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

public class Main {
    private static int a = 0;
    public static void main(String[] args) throws InterruptedException {
        new Thread(() -> {
            while (a == 0);
            System.out.println("线程结束！");
        }).start();

        Thread.sleep(1000);
        System.out.println("正在修改a的值...");
        a = 1;   //很明显，按照我们的逻辑来说，a的值被修改那么另一个线程将不再循环
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

public interface Lock {
      //获取锁，拿不到锁会阻塞，等待其他线程释放锁，获取到锁后返回
    void lock();
      //同上，但是等待过程中会响应中断
    void lockInterruptibly() throws InterruptedException;
      //尝试获取锁，但是不会阻塞，如果能获取到会返回true，不能返回false
    boolean tryLock();
      //尝试获取锁，但是可以限定超时时间，如果超出时间还没拿到锁返回false，否则返回true，可以响应中断
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
      //释放锁
    void unlock();
      //暂时可以理解为替代传统的Object的wait()、notify()等操作的工具
    Condition newCondition();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

public interface Condition {
      //与调用锁对象的wait方法一样，会进入到等待状态，但是这里需要调用Condition的signal或signalAll方法进行唤醒（感觉就是和普通对象的wait和notify是对应的）同时，等待状态下是可以响应中断的
      void await() throws InterruptedException;
      //同上，但不响应中断（看名字都能猜到）
      void awaitUninterruptibly();
      //等待指定时间，如果在指定时间（纳秒）内被唤醒，会返回剩余时间，如果超时，会返回0或负数，可以响应中断
      long awaitNanos(long nanosTimeout) throws InterruptedException;
      //等待指定时间（可以指定时间单位），如果等待时间内被唤醒，返回true，否则返回false，可以响应中断
      boolean await(long time, TimeUnit unit) throws InterruptedException;
      //可以指定一个明确的时间点，如果在时间点之前被唤醒，返回true，否则返回false，可以响应中断
      boolean awaitUntil(Date deadline) throws InterruptedException;
      //唤醒一个处于等待状态的线程，注意还得获得锁才能接着运行
      void signal();
      //同上，但是是唤醒所有等待线程
      void signalAll();
}

1
2
3
4
5
6

public interface ReadWriteLock {
    //获取读锁
    Lock readLock();
      //获取写锁
    Lock writeLock();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

static final int WAITING   = 1;          // must be 1  
static final int CANCELLED = 0x80000000; // must be negative  
static final int COND      = 2;          // in a condition wait

/**  
 * Head of the wait queue, lazily initialized. 
 */
private transient volatile Node head;  
/**  
 * Tail of the wait queue. After initialization, modified only via casTail. */ private transient volatile Node tail;  
  
/**  
 * The synchronization state. */ 
private volatile int state;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

abstract static class Node {  
    volatile Node prev;       // initially attached via casTail  
    volatile Node next;       // visibly nonnull when signallable  
    Thread waiter;            // visibly nonnull when enqueued  
    volatile int status;      // written by owner, atomic bit ops by others  
  
    // methods for atomic operations    
    final boolean casPrev(Node c, Node v) {  // for cleanQueue  
        return U.weakCompareAndSetReference(this, PREV, c, v);  
    }  
    final boolean casNext(Node c, Node v) {  // for cleanQueue  
        return U.weakCompareAndSetReference(this, NEXT, c, v);  
    }  
    final int getAndUnsetStatus(int v) {     // for signalling  
        return U.getAndBitwiseAndInt(this, STATUS, ~v);  
    }  
    final void setPrevRelaxed(Node p) {      // for off-queue assignment  
        U.putReference(this, PREV, p);  
    }  
    final void setStatusRelaxed(int s) {     // for off-queue assignment  
        U.putInt(this, STATUS, s);  
    }  
    final void clearStatus() {               // for reducing unneeded signals  
        U.putIntOpaque(this, STATUS, 0);  
    }  
  
    private static final long STATUS  
        = U.objectFieldOffset(Node.class, "status");  
    private static final long NEXT  
        = U.objectFieldOffset(Node.class, "next");  
    private static final long PREV  
        = U.objectFieldOffset(Node.class, "prev");  
}

1
2
3
4

final void lock() {  
    if (!initialTryLock())  
        acquire(1);  
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

/**  
 * Acquires only if reentrant or queue is empty. */
final boolean initialTryLock() {  
    Thread current = Thread.currentThread();  
    int c = getState();  
    if (c == 0) {  
        if (!hasQueuedThreads() && compareAndSetState(0, 1)) {  
            setExclusiveOwnerThread(current);  
            return true;  
        }  
    } else if (getExclusiveOwnerThread() == current) {  
        if (++c < 0) // overflow  
            throw new Error("Maximum lock count exceeded");  
        setState(c);  
        return true;  
    }  
    return false;  
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

/**  
 * Queries whether any threads are waiting to acquire. Note that * because cancellations due to interrupts and timeouts may occur * at any time, a {@code true} return does not guarantee that any  
 * other thread will ever acquire. * * @return {@code true} if there may be other threads waiting to acquire  
 */
 public final boolean hasQueuedThreads() {  
    for (Node p = tail, h = head; p != h && p != null; p = p.prev)  
        if (p.status >= 0)  
            return true;  
    return false;  
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

/**  
 * Acquires in exclusive mode, ignoring interrupts.  Implemented * by invoking at least once {@link #tryAcquire},  
 * returning on success.  Otherwise the thread is queued, possibly * repeatedly blocking and unblocking, invoking {@link  
 * #tryAcquire} until success.  This method can be used  
 * to implement method {@link Lock#lock}.  
 * * @param arg the acquire argument.  This value is conveyed to  
 *        {@link #tryAcquire} but is otherwise uninterpreted and  
 *        can represent anything you like. */
public final void acquire(int arg) {  
    if (!tryAcquire(arg))  
        acquire(null, arg, false, false, false, 0L);  
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

/**  
 * Acquires only if thread is first waiter or empty */
protected final boolean tryAcquire(int acquires) {  
    if (getState() == 0 && !hasQueuedPredecessors() &&  
        compareAndSetState(0, acquires)) {  
        setExclusiveOwnerThread(Thread.currentThread());  
        return true;  
    }  
    return false;  
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

/**  
 * Queries whether any threads have been waiting to acquire longer * than the current thread. * * <p>An invocation of this method is equivalent to (but may be  
 * more efficient than): * <pre> {@code  
 * getFirstQueuedThread() != Thread.currentThread()  
 *   && hasQueuedThreads()}</pre>  
 *  
 * <p>Note that because cancellations due to interrupts and  
 * timeouts may occur at any time, a {@code true} return does not  
 * guarantee that some other thread will acquire before the current * thread.  Likewise, it is possible for another thread to win a * race to enqueue after this method has returned {@code false},  
 * due to the queue being empty. * * <p>This method is designed to be used by a fair synchronizer to  
 * avoid <a href="AbstractQueuedSynchronizer.html#barging">barging</a>.  
 * Such a synchronizer's {@link #tryAcquire} method should return  
 * {@code false}, and its {@link #tryAcquireShared} method should  
 * return a negative value, if this method returns {@code true}  
 * (unless this is a reentrant acquire).  For example, the {@code  
 * tryAcquire} method for a fair, reentrant, exclusive mode  
 * synchronizer might look like this: * * <pre> {@code  
 * protected boolean tryAcquire(int arg) {  
 *   if (isHeldExclusively()) { *     // A reentrant acquire; increment hold count *     return true; *   } else if (hasQueuedPredecessors()) { *     return false; *   } else { *     // try to acquire normally *   } * }}</pre>  
 *  
 * @return {@code true} if there is a queued thread preceding the  
 *         current thread, and {@code false} if the current thread  
 *         is at the head of the queue or the queue is empty * @since 1.7  
 */
public final boolean hasQueuedPredecessors() {  
    Thread first = null; Node h, s;  
    if ((h = head) != null && ((s = h.next) == null ||  
                               (first = s.waiter) == null ||  
                               s.prev == null))  
        first = getFirstQueuedThread(); // retry via getFirstQueuedThread  
    return first != null && first != Thread.currentThread();  
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

final int acquire(Node node, int arg, boolean shared,
                      boolean interruptible, boolean timed, long time) {
        Thread current = Thread.currentThread();
        byte spins = 0, postSpins = 0;   // retries upon unpark of first thread
        boolean interrupted = false, first = false;
        Node pred = null;               // predecessor of node when enqueued

        /*
         * Repeatedly:
         *  Check if node now first
         *    if so, ensure head stable, else ensure valid predecessor
         *  if node is first or not yet enqueued, try acquiring
         *  else if queue is not initialized, do so by attaching new header node
         *     resort to spinwait on OOME trying to create node
         *  else if node not yet created, create it
         *     resort to spinwait on OOME trying to create node
         *  else if not yet enqueued, try once to enqueue
         *  else if woken from park, retry (up to postSpins times)
         *  else if WAITING status not set, set and retry
         *  else park and clear WAITING status, and check cancellation
         */

        for (;;) {
            if (!first && (pred = (node == null) ? null : node.prev) != null &&
                !(first = (head == pred))) {
                if (pred.status < 0) {
                    cleanQueue();           // predecessor cancelled
                    continue;
                } else if (pred.prev == null) {
                    Thread.onSpinWait();    // ensure serialization
                    continue;
                }
            }
            if (first || pred == null) {
                boolean acquired;
                try {
                    if (shared)
                        acquired = (tryAcquireShared(arg) >= 0);
                    else
                        acquired = tryAcquire(arg);
                } catch (Throwable ex) {
                    cancelAcquire(node, interrupted, false);
                    throw ex;
                }
                if (acquired) {
                    if (first) {
                        node.prev = null;
                        head = node;
                        pred.next = null;
                        node.waiter = null;
                        if (shared)
                            signalNextIfShared(node);
                        if (interrupted)
                            current.interrupt();
                    }
                    return 1;
                }
            }
            Node t;
            if ((t = tail) == null) {           // initialize queue
                if (tryInitializeHead() == null)
                    return acquireOnOOME(shared, arg);
            } else if (node == null) {          // allocate; retry before enqueue
                try {
                    node = (shared) ? new SharedNode() : new ExclusiveNode();
                } catch (OutOfMemoryError oome) {
                    return acquireOnOOME(shared, arg);
                }
            } else if (pred == null) {          // try to enqueue
                node.waiter = current;
                node.setPrevRelaxed(t);         // avoid unnecessary fence
                if (!casTail(t, node))
                    node.setPrevRelaxed(null);  // back out
                else
                    t.next = node;
            } else if (first && spins != 0) {
                --spins;                        // reduce unfairness on rewaits
                Thread.onSpinWait();
            } else if (node.status == 0) {
                node.status = WAITING;          // enable signal and recheck
            } else {
                long nanos;
                spins = postSpins = (byte)((postSpins << 1) | 1);
                if (!timed)
                    LockSupport.park(this);
                else if ((nanos = time - System.nanoTime()) > 0L)
                    LockSupport.parkNanos(this, nanos);
                else
                    break;
                node.clearStatus();
                if ((interrupted |= Thread.interrupted()) && interruptible)
                    break;
            }
        }
        return cancelAcquire(node, interrupted, interruptible);
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

/**
 * Possibly repeatedly traverses from tail, unsplicing cancelled
 * nodes until none are found. Unparks nodes that may have been
 * relinked to be next eligible acquirer.
 */
private void cleanQueue() {
    for (;;) {                               // restart point
        for (Node q = tail, s = null, p, n;;) { // (p, q, s) triples
            if (q == null || (p = q.prev) == null)
                return;                      // end of list
            if (s == null ? tail != q : (s.prev != q || s.status < 0))
                break;                       // inconsistent
            if (q.status < 0) {              // cancelled
                if ((s == null ? casTail(q, p) : s.casPrev(q, p)) &&
                    q.prev == p) {
                    p.casNext(q, s);         // OK if fails
                    if (p.prev == null)
                        signalNext(p);
                }
                break;
            }
            if ((n = p.next) != q) {         // help finish
                if (n != null && q.prev == p) {
                    p.casNext(n, q);
                    if (p.prev == null)
                        signalNext(p);
                }
                break;
            }
            s = q;
            q = q.prev;
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

protected final boolean tryRelease(int releases) {  
    int c = getState() - releases;  
    if (getExclusiveOwnerThread() != Thread.currentThread())  
        throw new IllegalMonitorStateException();  
    boolean free = (c == 0);  
    if (free)  
        setExclusiveOwnerThread(null);  
    setState(c);  
    return free;  
}
/**
 * Wakes up the successor of given node, if one exists, and unsets its
 * WAITING status to avoid park race. This may fail to wake up an
 * eligible thread when one or more have been cancelled, but
 * cancelAcquire ensures liveness.
 */
private static void signalNext(Node h) {
    Node s;
    if (h != null && (s = h.next) != null && s.status != 0) {
        s.getAndUnsetStatus(WAITING);
        LockSupport.unpark(s.waiter);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

static final class ConditionNode extends Node
    implements ForkJoinPool.ManagedBlocker {
    ConditionNode nextWaiter;            // link to next waiting node
    // ......
}
/**
 * Condition implementation for a {@link AbstractQueuedSynchronizer}
 * serving as the basis of a {@link Lock} implementation.
 *
 * <p>Method documentation for this class describes mechanics,
 * not behavioral specifications from the point of view of Lock
 * and Condition users. Exported versions of this class will in
 * general need to be accompanied by documentation describing
 * condition semantics that rely on those of the associated
 * {@code AbstractQueuedSynchronizer}.
 *
 * <p>This class is Serializable, but all fields are transient,
 * so deserialized conditions have no waiters.
 */
public class ConditionObject implements Condition, java.io.Serializable {
    private static final long serialVersionUID = 1173984872572414699L;
    /** First node of condition queue. */
    private transient ConditionNode firstWaiter;
    /** Last node of condition queue. */
    private transient ConditionNode lastWaiter;
    // ......
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

/**
 * Implements interruptible condition wait.
 * <ol>
 * <li>If current thread is interrupted, throw InterruptedException.
 * <li>Save lock state returned by {@link #getState}.
 * <li>Invoke {@link #release} with saved state as argument,
 *     throwing IllegalMonitorStateException if it fails.
 * <li>Block until signalled or interrupted.
 * <li>Reacquire by invoking specialized version of
 *     {@link #acquire} with saved state as argument.
 * <li>If interrupted while blocked in step 4, throw InterruptedException.
 * </ol>
 */
public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    ConditionNode node = newConditionNode();
    if (node == null)
        return;
    int savedState = enableWait(node);
    LockSupport.setCurrentBlocker(this); // for back-compatibility
    boolean interrupted = false, cancelled = false, rejected = false;
    while (!canReacquire(node)) {
        if (interrupted |= Thread.interrupted()) {
            if (cancelled = (node.getAndUnsetStatus(COND) & COND) != 0)
                break;              // else interrupted after signal
        } else if ((node.status & COND) != 0) {
            try {
                if (rejected)
                    node.block();
                else
                    ForkJoinPool.managedBlock(node);
            } catch (RejectedExecutionException ex) {
                rejected = true;
            } catch (InterruptedException ie) {
                interrupted = true;
            }
        } else
            Thread.onSpinWait();    // awoke while enqueuing
    }
    LockSupport.setCurrentBlocker(null);
    node.clearStatus();
    acquire(node, savedState, false, false, false, 0L);
    if (interrupted) {
        if (cancelled) {
            unlinkCancelledWaiters(node);
            throw new InterruptedException();
        }
        Thread.currentThread().interrupt();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

/**
 * Adds node to condition list and releases lock.
 *
 * @param node the node
 * @return savedState to reacquire after wait
 */
private int enableWait(ConditionNode node) {
    if (isHeldExclusively()) {
        node.waiter = Thread.currentThread();
        node.setStatusRelaxed(COND | WAITING);
        ConditionNode last = lastWaiter;
        if (last == null)
            firstWaiter = node;
        else
            last.nextWaiter = node;
        lastWaiter = node;
        int savedState = getState();
        if (release(savedState))
            return savedState;
    }
    node.status = CANCELLED; // lock not held or inconsistent
    throw new IllegalMonitorStateException();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

/**
 * Returns true if a node that was initially placed on a condition
 * queue is now ready to reacquire on sync queue.
 * @param node the node
 * @return true if is reacquiring
 */
private boolean canReacquire(ConditionNode node) {
    // check links, not status to avoid enqueue race
    Node p; // traverse unless known to be bidirectionally linked
    return node != null && (p = node.prev) != null &&
        (p.next == node || isEnqueued(node));
}

1
2
3

final int getAndUnsetStatus(int v) {     // for signalling  
    return U.getAndBitwiseAndInt(this, STATUS, ~v);  
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

/**
 * Removes and transfers one or all waiters to sync queue.
 */
private void doSignal(ConditionNode first, boolean all) {
    while (first != null) {
        ConditionNode next = first.nextWaiter;
        if ((firstWaiter = next) == null)
            lastWaiter = null;
        if ((first.getAndUnsetStatus(COND) & COND) != 0) {
            enqueue(first);
            if (!all)
                break;
        }
        first = next;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

/**
 * Enqueues the node unless null. (Currently used only for
 * ConditionNodes; other cases are interleaved with acquires.)
 */
final void enqueue(ConditionNode node) {
    if (node != null) {
        boolean unpark = false;
        for (Node t;;) {
            if ((t = tail) == null && (t = tryInitializeHead()) == null) {
                unpark = true;             // wake up to spin on OOME
                break;
            }
            node.setPrevRelaxed(t);        // avoid unnecessary fence
            if (casTail(t, node)) {
                t.next = node;
                if (t.status < 0)          // wake up to clean link
                    unpark = true;
                break;
            }
        }
        if (unpark)
            LockSupport.unpark(node.waiter);
    }
}

1
2
3
4
5

public boolean add(E e) { 
    ensureCapacityInternal(size + 1); // Increments modCount!! 
    elementData[size++] = e; //这一句出现了数组越界 
    return true; 
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    int hash = spread(key.hashCode());    //计算键的hash值，用于确定在哈希表中的位置
    int binCount = 0;   //用来记录链表长度
    for (Node<K,V>[] tab = table;;) {    //CAS自旋锁
        Node<K,V> f; int n, i, fh;
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();    //如果数组（哈希表）为空肯定是要进行初始化的，然后再重新进下一轮循环
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {   //如果哈希表该位置为null，直接CAS插入结点作为头结即可（注意这里会将f设置当前哈希表位置上的头结点）
            if (casTabAt(tab, i, null,
                         new Node<K,V>(hash, key, value, null)))  
                break;                   // 如果CAS成功，直接break结束put方法，失败那就继续下一轮循环
        } else if ((fh = f.hash) == MOVED)   //头结点哈希值为-1，正在扩容
            tab = helpTransfer(tab, f);   //帮助进行迁移，完事之后再来下一次循环
        else {     //特殊情况都完了，这里就该是正常情况了，
            V oldVal = null;
            synchronized (f) {   //在前面的循环中f肯定是被设定为了哈希表某个位置上的头结点，这里直接把它作为锁加锁了，防止同一时间其他线程也在操作哈希表中这个位置上的链表或是红黑树
                if (tabAt(tab, i) == f) {
                    if (fh >= 0) {    //头结点的哈希值大于等于0说明是链表，下面就是针对链表的一些列操作
                        ...实现细节略
                    } else if (f instanceof TreeBin) {   //肯定不大于0，肯定也不是-1，还判断是不是TreeBin，所以不用猜了，肯定是红黑树，下面就是针对红黑树的情况进行操作
                          //在ConcurrentHashMap并不是直接存储的TreeNode，而是TreeBin
                        ...实现细节略
                    }
                }
            }
              //根据链表长度决定是否要进化为红黑树
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);   //注意这里只是可能会进化为红黑树，如果当前哈希表的长度小于64，它会优先考虑对哈希表进行扩容
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    addCount(1L, binCount);
    return null;
}

public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    int h = spread(key.hashCode());   //计算哈希值
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {
          // 如果头结点就是我们要找的，那直接返回值就行了
        if ((eh = e.hash) == h) {
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
          //要么是正在扩容，要么就是红黑树，负数只有这两种情况
        else if (eh < 0)
            return (p = e.find(h, key)) != null ? p.val : null;
          //确认无误，肯定在列表里，开找
        while ((e = e.next) != null) {
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
      //没找到只能null了
    return null;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

public interface BlockingQueue<E> extends Queue<E> {
    boolean add(E e);
    
    //入队，如果队列已满，返回false否则返回true（非阻塞）
    boolean offer(E e);
    
    //入队，如果队列已满，阻塞线程直到能入队为止
    void put(E e) throws InterruptedException;
    
    //入队，如果队列已满，阻塞线程直到能入队或超时、中断为止，入队成功返回true否则false
    boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException;
        
    //出队，如果队列为空，阻塞线程直到能出队为止
    E take() throws InterruptedException;
    
    //出队，如果队列为空，阻塞线程直到能出队超时、中断为止，出队成功正常返回，否则返回null
    E poll(long timeout, TimeUnit unit)
        throws InterruptedException;

    //返回此队列理想情况下（在没有内存或资源限制的情况下）可以不阻塞地入队的数量，如果没有限制，则返回 Integer.MAX_VALUE
    int remainingCapacity();

    boolean remove(Object o);

    public boolean contains(Object o);

    //一次性从BlockingQueue中获取所有可用的数据对象（还可以指定获取数据的个数）
    int drainTo(Collection<? super E> c);

    int drainTo(Collection<? super E> c, int maxElements);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

/** Main lock guarding all access */  
final ReentrantLock lock;  
  
/** Condition for waiting takes */  
@SuppressWarnings("serial")  // Classes implementing Condition may be serializable.  
private final Condition notEmpty;  
  
/** Condition for waiting puts */  
@SuppressWarnings("serial")  // Classes implementing Condition may be serializable.  
private final Condition notFull;

/**
 * Creates an {@code ArrayBlockingQueue} with the given (fixed)
 * capacity and the specified access policy.
 *
 * @param capacity the capacity of this queue
 * @param fair if {@code true} then queue accesses for threads blocked
 *        on insertion or removal, are processed in FIFO order;
 *        if {@code false} the access order is unspecified.
 * @throws IllegalArgumentException if {@code capacity < 1}
 */
public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
        throw new IllegalArgumentException();
    this.items = new Object[capacity];
    lock = new ReentrantLock(fair);
    notEmpty = lock.newCondition();
    notFull =  lock.newCondition();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

public boolean offer(E e) {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;    //使用了类里面的ReentrantLock进行加锁操作
    lock.lock();    //保证同一时间只有一个线程进入
    try {
        if (count == items.length)   //直接看看队列是否已满，如果没满则直接入队，如果已满则返回false
            return false;
        else {
            enqueue(e);
            return true;
        }
    } finally {
        lock.unlock();
    }
}

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;    //同样的，需要进行加锁操作
    lock.lockInterruptibly();    //注意这里是可以响应中断的
    try {
        while (count == items.length)
            notFull.await();    //当队列已满时会直接挂起当前线程，在其他线程出队操作时会被唤醒
        enqueue(e);   //直到队列有空位才将线程入队
    } finally {
        lock.unlock();
    }
}