当前位置:首页 > 开发 > 编程语言 > 编程 > 正文

linkedBlokingQueue源码学习

发表于: 2015-04-29   作者:cfyme   来源:转载   浏览次数:
摘要: ReentrantLock是一个较为常用的锁对象 可重入的互斥锁定 与synchronized 相同的一些基本行为和语义,但功能更强大   Condition类与lock绑定,用newCondition()方法创建,提供了线程之间通信的方式(类似信号量)。 其使用基本与object类的wait,notify,notifyAll相同。   1,用condition.aw

ReentrantLock是一个较为常用的锁对象 可重入的互斥锁定 与synchronized 相同的一些基本行为和语义,但功能更强大

 

Condition类与lock绑定,用newCondition()方法创建,提供了线程之间通信的方式(类似信号量)。

其使用基本与object类的wait,notify,notifyAll相同。

 

1,用condition.await()替换Object,wait(),调用时该线程阻塞,释放该线程的锁。

2,用condition.signal()替换Object.notify(),用condition.signalAll()替换Object.notifyAll(),唤醒该condition await方法所阻塞的线程

 

public class LinkedBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable { 
      //链表节点node类结构   
      static class Node<E> {  
          volatile E item;//volatile,保证了数据的可见性   
          Node<E> next;  
          Node(E x) { item = x; }  
      }  
      //容量
      private final int capacity;  
      //用原子变量,当前元素个数  
      private final AtomicInteger count = new AtomicInteger(0);  
      //头节点
      private transient Node<E> head;  
      //表尾节点 
      private transient Node<E> last;  
      //获取元素或删除元素时,要加的takeLock锁  
      private final ReentrantLock takeLock = new ReentrantLock();  
      //获取元素时若队列为空,线程阻塞,直至notEmpty条件满足(被通知) 
      private final Condition notEmpty = takeLock.newCondition();  
      //插入元素时 要加putLock锁  
      private final ReentrantLock putLock = new ReentrantLock();  
      //插入时,若队列已满,线程阻塞,直至notFull条件满足(被通知)
      private final Condition notFull = putLock.newCondition();  
      // 唤醒等待的take操作,插入数据时若插入前链表中无数据,则调用,表示链表不再为空
      private void signalNotEmpty() {  
          final ReentrantLock takeLock = this.takeLock;  
          takeLock.lock();  
          try {  
              notEmpty.signal();  
          } finally {  
              takeLock.unlock();  
          }  
      }  
      //唤醒等待插入操作,移除数据时若链表原先已满则调用,表示链表不再满 
      private void signalNotFull() {  
          final ReentrantLock putLock = this.putLock;  
          putLock.lock();  
          try {  
              notFull.signal();  
          } finally {  
              putLock.unlock();  
          }  
      }  
      // 插入到链表尾部 
      private void insert(E x) {  
          last = last.next = new Node<E>(x);  
      }  
      //获取并移除头元素 
      private E extract() {  
          Node<E> first = head.next;  
          head = first;  
          E x = first.item;  
          first.item = null;  
          return x;  
      }  
      //锁住两把锁,在remove,clear等方法中调用   
      private void fullyLock() {  
          putLock.lock();  
          takeLock.lock();  
      }  
      //和fullyLock成对使用 
      private void fullyUnlock() {  
          takeLock.unlock();  
          putLock.unlock();  
      }  
      //默认构造,容量为 Integer.MAX_VALUE  
 
      public LinkedBlockingQueue() {  
          this(Integer.MAX_VALUE);  
      }  
      //指定容量的构造   
      public LinkedBlockingQueue(int capacity) {  
          if (capacity <= 0) throw new IllegalArgumentException();  
          this.capacity = capacity;  
          last = head = new Node<E>(null);  
      }  
      //指定初始化集合的构造   
      public LinkedBlockingQueue(Collection<? extends E> c) {  
          this(Integer.MAX_VALUE);  
          for (E e : c)  
              add(e);  
      }  
      //获得大小 
        
      public int size() {  
          return count.get();  
      }  
      //剩余容量  
      public int remainingCapacity() {  
          return capacity - count.get();  
      }  
      // 将指定元素插入到此队列的尾部,如已满,阻塞至队列中有元素被移除 
      public void put(E e) throws InterruptedException {  
          if (e == null) throw new NullPointerException();  
          int c = -1;  
          final ReentrantLock putLock = this.putLock;  
          final AtomicInteger count = this.count;
   //加put锁,多个线程不能同时进入  
          putLock.lockInterruptibly();  
          try {  
              try {  
    //容量已满,则一直阻塞
                  while (count.get() == capacity)  
                      notFull.await();  
              } catch (InterruptedException ie) {  
                  notFull.signal(); // propagate to a non-interrupted thread  
                  throw ie;  
              }  
//插入
              insert(e);  
              c = count.getAndIncrement();
//通知链表未满  
              if (c + 1 < capacity)  
                  notFull.signal();  
          } finally {  
//解锁,注意必须在finally里调用,反正各种异常导致没有unlock使线程死锁
              putLock.unlock();  
          }  
    //通知链表非空
          if (c == 0)  
              signalNotEmpty();  
      }  
      // 将指定元素插入到此队列的尾部,如有必要,则等待一定时间以使空间变得可用。 
       
      public boolean offer(E e, long timeout, TimeUnit unit)  
          throws InterruptedException {  
          if (e == null) throw new NullPointerException();  
          long nanos = unit.toNanos(timeout);  
          int c = -1;  
          final ReentrantLock putLock = this.putLock;  
          final AtomicInteger count = this.count;  
   //加锁
          putLock.lockInterruptibly();  
          try {  
              for (;;) {
                  //未满可插入  
                  if (count.get() < capacity) {  
                      insert(e);  
                      c = count.getAndIncrement();
    //通知未满  
                      if (c + 1 < capacity)  
                          notFull.signal();
    //跳出循环  
                      break;  
                  }  
   //队列已满,未能插入,等待时间是负的,直接返回
                  if (nanos <= 0)  
                      return false;  
                  try {  
    //等待一定时间后再次尝试
                      nanos = notFull.awaitNanos(nanos);  
                  } catch (InterruptedException ie) {  
                      notFull.signal(); // propagate to a non-interrupted thread  
                      throw ie;  
                  }  
              }  
          } finally {  
//解锁
              putLock.unlock();  
          }  
//通知已插入数据,链表非空
          if (c == 0)  
              signalNotEmpty();  
          return true;  
      }  
      //将指定元素插入到此队列的尾部(如果立即可行且不会超出此队列的容量), 
       在成功时返回 true,如果此队列已满,则返回 false。 
        
      public boolean offer(E e) {  
          if (e == null) throw new NullPointerException();  
          final AtomicInteger count = this.count;  
          if (count.get() == capacity)  
              return false;  
          int c = -1;  
          final ReentrantLock putLock = this.putLock;  
          putLock.lock();  
          try {  
//由于可能在lock被阻塞时其他线程进行了插入操作,需再次判断count
              if (count.get() < capacity) {  
                  insert(e);  
                  c = count.getAndIncrement();
    //通知未满  
                  if (c + 1 < capacity)  
                      notFull.signal();  
              }  
          } finally {  
              putLock.unlock();  
          }  
  //通知非空
          if (c == 0)  
              signalNotEmpty();
          // >0表示已成功插入  
          return c >= 0;  
      }  
      //获取并移除此队列的头部,若队列为空,则阻塞。  
      public E take() throws InterruptedException {  
          E x;  
          int c = -1;  
          final AtomicInteger count = this.count;  
          final ReentrantLock takeLock = this.takeLock;
   //加锁 
          takeLock.lockInterruptibly();  
          try {  
              try {
   //队列为空时阻塞 
                  while (count.get() == 0)  
                      notEmpty.await();  
              } catch (InterruptedException ie) {  
                  notEmpty.signal(); // propagate to a non-interrupted thread  
                  throw ie;  
              }  
//获取数据
              x = extract();  
              c = count.getAndDecrement();
//通知非空  
              if (c > 1)  
                  notEmpty.signal();  
          } finally {  
              takeLock.unlock();  
          }  
    //通知未满
          if (c == capacity)  
              signalNotFull();  
          return x;  
      }  
        
      //与offer方法结构基本一致,若队列为空,则阻塞一段时间,一段时间后仍为空,则返回null
      public E poll(long timeout, TimeUnit unit) throws InterruptedException {  
          E x = null;  
          int c = -1;  
          long nanos = unit.toNanos(timeout);  
          final AtomicInteger count = this.count;  
          final ReentrantLock takeLock = this.takeLock;  
          takeLock.lockInterruptibly();  
          try {  
              for (;;) {  
                  if (count.get() > 0) {  
                      x = extract();  
                      c = count.getAndDecrement();  
                      if (c > 1)  
                          notEmpty.signal();  
                      break;  
                  }  
                  if (nanos <= 0)  
                      return null;  
                  try {  
                      nanos = notEmpty.awaitNanos(nanos);  
                  } catch (InterruptedException ie) {  
                      notEmpty.signal(); // propagate to a non-interrupted thread  
                      throw ie;  
                  }  
              }  
          } finally {  
              takeLock.unlock();  
          }  
          if (c == capacity)  
              signalNotFull();  
          return x;  
      }  
        
      ////与offer方法结构基本一致 队列为空,不阻塞,直接返回null
      public E poll() {  
          final AtomicInteger count = this.count;  
          if (count.get() == 0)  
              return null;  
          E x = null;  
          int c = -1;  
          final ReentrantLock takeLock = this.takeLock;  
          takeLock.lock();  
          try {  
              if (count.get() > 0) {  
                  x = extract();  
                  c = count.getAndDecrement();  
                  if (c > 1)  
                      notEmpty.signal();  
              }  
          } finally {  
              takeLock.unlock();  
          }  
          if (c == capacity)  
              signalNotFull();  
          return x;  
      }  
      //获取但不移除此队列的头;如果此队列为空,则返回 null。  
      public E peek() {  
          if (count.get() == 0)  
              return null;  
          final ReentrantLock takeLock = this.takeLock;  
          takeLock.lock();  
          try {  
              Node<E> first = head.next;  
              if (first == null)  
                  return null;  
              else  
                  return first.item;  
          } finally {  
              takeLock.unlock();  
          }  
      }  
      /** 
       * 从此队列移除指定元素的单个实例(如果存在)。 
       */  
      public boolean remove(Object o) {  
          if (o == null) return false;  
          boolean removed = false;
   //同时加锁,此时其他线程不能插入,不能移除
          fullyLock();  
          try {  
              Node<E> trail = head;  
              Node<E> p = head.next;
//遍历,获取到该元素  
              while (p != null) {  
                  if (o.equals(p.item)) {  
                      removed = true;  
                      break;  
                  }  
                  trail = p;  
                  p = p.next;  
              }  
//删除该元素
              if (removed) {  
                  p.item = null;  
                  trail.next = p.next;  
                  if (last == p)  
                      last = trail;  
                  if (count.getAndDecrement() == capacity)  
                      notFull.signalAll();  
              }  
          } finally {  
              fullyUnlock();  
          }  
          return removed;  
      }  
      ……  
  }

 

linkedBlokingQueue源码学习

  • 0

    开心

    开心

  • 0

    板砖

    板砖

  • 0

    感动

    感动

  • 0

    有用

    有用

  • 0

    疑问

    疑问

  • 0

    难过

    难过

  • 0

    无聊

    无聊

  • 0

    震惊

    震惊

编辑推荐
学习vaniglia的状态机特性源代码,这个特性在企业化开发中也是适用的,类似于struts2里配置action。
最近在做日志统计程序,发现对方的程序是在Jquery基础上进行开发的,而公司的网站的框架是prototype
读《Google Chrome源码剖析》笔记,学习Chrome中优秀的设计思想。 原文地址: http://www.ha97.com/
相关性能测试和使用例子,可以参见另一篇文章: cglib相关性能测试对比 背景 前段时间在工作中,包
背景 最忌工作中接触到Netty相关应用场景,之前看过mima的部分源码,所以最近看了Netty的部分源码和
相关性能测试和使用例子,可以参见另一篇文章: cglib相关性能测试对比 背景 前段时间在工作中,包
a) Spring 初始化的时候首先要运行的类为: org.springframework.web.context. ContextLoaderListen
本文从设计框架角度介绍mochiweb项目源码,用于总结并备忘最近对mochiweb源码的研究学习,下面直接
读《Google Chrome源码剖析》笔记,学习Chrome中优秀的设计思想。 原文地址: http://www.ha97.com/
众所周知,Spring以其强大而又灵活的IoC管理功能著称。IoC本质上是通过依赖注入DI的方式实现组件之间
版权所有 IT知识库 CopyRight © 2009-2015 IT知识库 IT610.com , All Rights Reserved. 京ICP备09083238号