Java中的LinkedTransferQueue tryTransfer()方法及示例

LinkedTransferQueue 简介

LinkedTransferQueue 是Java中的一个并发容器，继承至AbstractQueue和TransferQueue，它提供了一种线程之间进行数据传递的高效方式。

与其它Queue实现不同的是，LinkedTransferQueue 在处理元素时会自动阻塞线程，直到有一个另外的线程通过take或者poll获取这个元素或者被offer方法成功插入一个元素。

对于连续空间(Location)的take操作，LinkedTransferQueue提供了一个额外的方法tryTransfer，这篇文章就是要详细介绍它。

tryTransfer()方法

tryTransfer可以直接将元素加入到BlockingQueue, 但需要满足一定条件，如果条件不符合，则会直接返回一个boolean值，不会阻塞线程，具体如下:

如果当前有等待获取的线程，则直接返回true，若无等待线程则执行offer操作，若成功则返回true，否则返回false。
若当前有竞争者线程正在执行take操作，则直接交由竞争者线程获取数据，那么返回一个false。
若其他线程正在等待入队，则直接将元素入队，等待线程就可以成功获取元素，返回true。

示例：线程池中使用tryTransfer()方法

现在我们考虑一个案例，在一个线程池中使用tryTransfer()方法，每个线程都是准备取出一个任务进行处理，当线程池中有未完成的任务时则尝试使用tryTransfer()方法，添加任务进入线程池，如下所示：

public class LinkedTansferQueueDemo {

    private static LinkedTransferQueue<Integer> linkedTransferQueue = new LinkedTransferQueue<>();

    public static void main(String[] args) {
        int threadCount = 2;
        ExecutorService executorService = Executors.newFixedThreadPool(threadCount);

        for (int i = 0; i < threadCount; i++) {
            executorService.execute(new WaitTask());
            executorService.execute(new Task());
        }
    }

    static class WaitTask implements Runnable {
        @Override
        public void run() {
            try {
                Integer take = linkedTransferQueue.take();
                System.out.println(Thread.currentThread().getName()+" Take : " + take);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    static class Task implements Runnable {
        @Override
        public void run() {
            Random random = new Random();
            try {
                Thread.sleep(random.nextInt(3000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (linkedTransferQueue.hasWaitingConsumer()) {
                System.out.println(Thread.currentThread().getName() + " Transfer : 1");
                boolean b = linkedTransferQueue.tryTransfer(1);
                System.out.println(Thread.currentThread().getName() + " " + b);
            } else {
                try {
                    System.out.println(Thread.currentThread().getName() + " Offer : 2");
                    linkedTransferQueue.offer(2, 300, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

该示例中，我们创建一个指定线程数的线程池，每个线程分别由WaitTask和Task类来执行。WaitTask类会对linkedTransferQueue进行take操作，如果没有元素，则会阻塞一直等待，直到有元素加入。Task类则会把一个任务加入到线程池中，如果有等待take的线程则直接使用tryTransfer方法把任务加入队列，没有等待take操作的线程则使用offer方法将任务加入队列。

执行结果如下：

pool-1-thread-2 Transfer : 1
pool-1-thread-2 true
pool-1-thread-2 Transfer : 1
pool-1-thread-2 true
pool-1-thread-1 Take : 1
pool-1-thread-1 Take : 1
pool-1-thread-1 Transfer : 1
pool-1-thread-1 true
pool-1-thread-1 Take : 2

从结果中我们可以看出，在没有等待take操作的情况下，Task类使用的是offer方法将任务加入队列。而在有等待take操作的情况下，则直接使用tryTransfer方法将任务加入队列并返回true。