菜单

注册免费送38元体验金闲话并发(七)——Java中之死队列

2018年11月20日 - 注册免费送38元体验金

3. 死队列的兑现原理

闲谈并发(七)——Java中之封堵队列

注册免费送38元体验金 1

作者
方腾飞
发布于 2013年12月18日 |
ArchSummit全球架构师峰会(北京站)2016年12月02-03日设置,询问再多详情!
2 讨论


死队列(BlockingQueue)是一个支撑有限个叠加操作的队。这有限单叠加的操作是:在帮列为空时,获取元素的线程会等待队列化非空。当排满时,存储元素的线程会等待队列可用。阻塞队列常用于生产者与买主之状况,生产者是奔队列里添加元素的线程,消费者是从队列里以元素的线程。阻塞队列就是生产者存放元素的器皿,而顾客为只有打容器里将元素。

闭塞队列提供了季种植处理办法:

方法\处理方式 抛出异常 返回特殊值 一直阻塞 超时退出
插入方法 add(e) offer(e) put(e) offer(e,time,unit)
移除方法 remove() poll() take() poll(time,unit)
检查方法 element() peek() 不可用 不可用

系厂商内容

同技术大牛,侃侃容器那些事情!

有关红包、SSD云盘等核心技术集锦!

特别数目的新玩法

58届小技术架构快速规划和落地

UCloud UCan技术夜,与“老驾驶员”共侃云服务技能实施及风向

JDK7提供了7个闭塞队列。分别是

ArrayBlockingQueue是一个用数组实现的有界阻塞队列。此阵按照先进先出(FIFO)的法对素进行排序。默认情况下不保险访问者公平的看队列,所谓公平访问队列是恃阻塞的享有生产者线程或顾客线程,当行可用时,可以按照阻塞的先后顺序访问队列,即先阻塞的劳动者线程,可以预先向队列里安插入元素,先堵塞的买主线程,可以先行打队列里抱元素。通常情况下为确保公平性会跌吞吐量。我们可以用以下代码创建一个公正的堵截队列:

ArrayBlockingQueue fairQueue = new  ArrayBlockingQueue(1000,true);

访问者的公平性是采用可更可锁实现之,代码如下:

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();
}

LinkedBlockingQueue是一个之所以链表实现之有界阻塞队列。此行列的默认和最好特别尺寸也Integer.MAX_VALUE。此行列按照先进先出的规范对素进行排序。

PriorityBlockingQueue是一个支撑先行级的无界队列。默认情况下元素采用自然顺序排列,也得由此比较器comparator来指定元素的排序规则。元素以升序排列。

DelayQueue是一个支持延时获取元素的无界阻塞队列。队列使用PriorityQueue来实现。队列中的因素必须贯彻Delayed接口,在创建元素时可以指定多久才能够从队列中得当前元素。只有以缓满时才能够由队列中领取元素。我们得以用DelayQueue运用在以下应用场景:

行中之Delayed必须兑现compareTo来指定元素的顺序。比如吃延时时空最丰富之放在队列的最后。实现代码如下:

public int compareTo(Delayed other) {
           if (other == this) // compare zero ONLY if same object
                return 0;
            if (other instanceof ScheduledFutureTask) {
                ScheduledFutureTask x = (ScheduledFutureTask)other;
                long diff = time - x.time;
                if (diff < 0)
                    return -1;
                else if (diff > 0)
                    return 1;
       else if (sequenceNumber < x.sequenceNumber)
                    return -1;
                else
                    return 1;
            }
            long d = (getDelay(TimeUnit.NANOSECONDS) -
                      other.getDelay(TimeUnit.NANOSECONDS));
            return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
        }

如何兑现Delayed接口

俺们好参照ScheduledThreadPoolExecutor里ScheduledFutureTask类。这个看似实现了Delayed接口。首先:在靶创建的时光,使用time记录面前对象啊时候可以下,代码如下:

ScheduledFutureTask(Runnable r, V result, long ns, long period) {
            super(r, result);
            this.time = ns;
            this.period = period;
            this.sequenceNumber = sequencer.getAndIncrement();
}

下一场用getDelay可以查询时因素还待延时多久,代码如下:

public long getDelay(TimeUnit unit) {
            return unit.convert(time - now(), TimeUnit.NANOSECONDS);
        }

经构造函数可以望延迟时间参数ns的单位是纳秒,自己统筹的当儿最好应用纳秒,因为getDelay时可以指定任意单位,一旦以纳秒作为单位,而延时的时又准确不顶纳秒就麻烦了。使用时请注意当time小于当前时空常,getDelay会返回负数。

怎样实现延时行

延时队排的贯彻深简单,当消费者从队列里落元素时,如果元素没有直达延时时间,就不通时线程。

long delay = first.getDelay(TimeUnit.NANOSECONDS);
                    if (delay <= 0)
                        return q.poll();
                    else if (leader != null)
                        available.await();

SynchronousQueue是一个无存储元素的堵塞队列。每一个put操作必须等待一个take操作,否则不能够延续上加元素。SynchronousQueue可以当作是一个传球手,负责管劳动者线程处理的数据直接传送给顾客线程。队列本身并无存储任何因素,非常适合于传递性场景,比如以一个线程中使的数目,传递给另外一个线程使用,SynchronousQueue的吞吐量高于LinkedBlockingQueue
和 ArrayBlockingQueue。

LinkedTransferQueue是一个出于链表结构成的无界阻塞TransferQueue队列。相对于任何阻塞队列,LinkedTransferQueue多了tryTransfer和transfer方法。

transfer方法。如果手上发消费者在等待接受元素(消费者使用take()方法或者带时限定的poll()方法时),transfer方法可以把劳动者传入的要素即transfer(传输)给买主。如果没有顾客于等候接受元素,transfer方法会将元素存放于列的tail节点,并等交拖欠因素让消费者花了才回。transfer方法的严重性代码如下:

Node pred = tryAppend(s, haveData);
return awaitMatch(s, pred, e, (how == TIMED), nanos);

第一履行代码是计算把存放在时元素的s节点作为tail节点。第二实施代码是于CPU自旋等待买主花元素。因为自旋会消耗CPU,所以自旋一定的次数后下Thread.yield()方法来刹车当前正值实行的线程,并实行另外线程。

tryTransfer方法。则是因此来试探下生产者传入的要素是否会一直招为顾客。如果没顾客等接受元素,则回false。和transfer方法的分别是tryTransfer方法无论消费者是否收取,方法就回。而transfer方法是得顶及买主花了才回到。

于含时间范围的tryTransfer(E e, long timeout, TimeUnit
unit)方法,则是计算把劳动者传入的因素直接招为消费者,但是要没消费者花该元素则等待指定的流年再返,如果超时还没费元素,则赶回false,如果以过时间内花费了元素,则回true。

LinkedBlockingDeque是一个由链表结构成的双向阻塞队列。所谓双向队列指的公可以从队列的双面插入和移出元素。双端队列因为差不多了一个操作队列的入口,在差不多线程同时入队时,也就减少了大体上底竞争。相比另的堵塞队列,LinkedBlockingDeque多矣addFirst,addLast,offerFirst,offerLast,peekFirst,peekLast等措施,以First单词结尾的办法,表示插入,获取(peek)或移除双端队列的率先独要素。以Last单词结尾的章程,表示插入,获取或移除双端队列的结尾一个素。另外插入方法add等同于addLast,移除方法remove等效于removeFirst。但是take方法也一如既往于takeFirst,不知底凡是休是Jdk的bug,使用时还是为此饱含First和Last后缀的方式更明了。

于初始化LinkedBlockingDeque时可设置容量防止该联网膨胀。另外双向阻塞队列可以使用在“工作窃取”模式面临。

3. 封堵队列的贯彻原理

要是队列是空的,消费者见面直接守候,当生产者添加元素时候,消费者是哪理解当前行有素的也罢?如果为您来统筹阻塞队列你晤面如何统筹,让劳动者与消费者会高效率的进行报道为?让咱先行来看望JDK是怎么样贯彻的。

运通知模式实现。所谓通知模式,就是当生产者往满的序列里上加元素时会见卡住住生产者,当顾客花了一个行列中之要素后,会通报劳动者当前班可用。通过查阅JDK源码发现ArrayBlockingQueue使用了Condition来兑现,代码如下:

private final Condition notFull;
private final Condition notEmpty;

public ArrayBlockingQueue(int capacity, boolean fair) {
        //省略其他代码
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

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();
        }
}

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();
    }

当我们通往队列里安插入一个素时,如果队列不可用,阻塞生产者主要透过LockSupport.park(this);来贯彻

public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                LockSupport.park(this);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            if (interruptMode != 0)

reportInterruptAfterWait(interruptMode);
        }

接轨上源码,发现调用setBlocker先保存下就要阻塞的线程,然后调用unsafe.park阻塞时线程。

public static void park(Object blocker) {
        Thread t = Thread.currentThread();
        setBlocker(t, blocker);
        unsafe.park(false, 0L);
        setBlocker(t, null);
    }

unsafe.park是独native方法,代码如下:

public native void park(boolean isAbsolute, long time);

park这个方法会阻塞时线程,只有以下四种植情况中之均等种有时,该办法才见面回来。

咱延续羁押一下JVM是什么落实park方法的,park在不同的操作系统使用不同的方法贯彻,在linux下是动的凡系方法pthread_cond_wait实现。实现代码在JVM源码路径src/os/linux/vm/os_linux.cpp里之
os::PlatformEvent::park方法,代码如下:

void os::PlatformEvent::park() {      
             int v ;
         for (;;) {
        v = _Event ;
         if (Atomic::cmpxchg (v-1, &_Event, v) == v) break ;
         }
         guarantee (v >= 0, "invariant") ;
         if (v == 0) {
         // Do this the hard way by blocking ...
         int status = pthread_mutex_lock(_mutex);
         assert_status(status == 0, status, "mutex_lock");
         guarantee (_nParked == 0, "invariant") ;
         ++ _nParked ;
         while (_Event < 0) {
         status = pthread_cond_wait(_cond, _mutex);
         // for some reason, under 2.7 lwp_cond_wait() may return ETIME ...
         // Treat this the same as if the wait was interrupted
         if (status == ETIME) { status = EINTR; }
         assert_status(status == 0 || status == EINTR, status, "cond_wait");
         }
         -- _nParked ;

         // In theory we could move the ST of 0 into _Event past the unlock(),
         // but then we'd need a MEMBAR after the ST.
         _Event = 0 ;
         status = pthread_mutex_unlock(_mutex);
         assert_status(status == 0, status, "mutex_unlock");
         }
         guarantee (_Event >= 0, "invariant") ;
         }

     }

pthread_cond_wait是一个大抵线程的准绳变量函数,cond是condition的缩写,字面意思可以解啊线程在等候一个尺度发生,这个规则是一个全局变量。这个办法接收两独参数,一个共享变量_cond注册免费送38元体验金,一个互斥量_mutex。而unpark方法以linux下是以pthread_cond_signal实现的。park
在windows下则是下WaitForSingleObject实现的。

当排满时,生产者往死队列里安插入一个素,生产者线程会进WAITING
(parking)状态。我们得以使jstack dump阻塞的劳动者线程看到这点:

"main" prio=5 tid=0x00007fc83c000000 nid=0x10164e000 waiting on condition [0x000000010164d000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x0000000140559fe8> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2043)
        at java.util.concurrent.ArrayBlockingQueue.put(ArrayBlockingQueue.java:324)
        at blockingqueue.ArrayBlockingQueueTest.main(ArrayBlockingQueueTest.java:11)

4. 参考资料


方腾飞,花名清英,并发编程网站站长。目前当阿里巴巴微贷事业部工作。并发编程网:http://ifeve.com,个人微博:http://weibo.com/kirals,欢迎通过我的微博开展技术交流。

感谢张龙本着本文的校。

受InfoQ中文站投稿或参与内容翻译工作,请邮件及editors@cn.infoq.com。也接大家经过新浪微博(@InfoQ)或者腾讯微博(@InfoQ)关注我们,并同我们的编撰和另外读者朋友交流。

【ArchSummit北京2016】“双11”过后,各电商巨头对技术展现如何总结?机器上大热,业务与之组成会擦起哪些的灯火?视频直播和新闻资讯,新工作需什么的初技巧与的部署?业务相对,技术拔尖不同,在年终之技术擂台上,各家将显示什么的技巧杀手锏…ArchSummit开幕倒计时,当即查看详情>>

相关文章

发表评论

电子邮件地址不会被公开。 必填项已用*标注

网站地图xml地图