在hadoop底层代码中,会看到有BlockingQueue的使用。
作为了解配置调优的作用,我也来了解一下BlockingQueue的使用。
BlockingQueue的核心方法:
放入数据:
offer(anObject):表示如果可能的话,将anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,
则返回true,否则返回false.(本方法不阻塞当前执行方法的线程)
offer(E o, long timeout, TimeUnit unit),可以设定等待的时间,如果在指定的时间内,还不能往队列中
加入BlockingQueue,则返回失败。
put(anObject):把anObject加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻断
直到BlockingQueue里面有空间再继续.
获取数据:
poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,
取不到时返回null;
poll(long timeout, TimeUnit unit):从BlockingQueue取出一个队首的对象,如果在指定时间内,
队列一旦有数据可取,则立即返回队列中的数据。否则知道时间超时还没有数据可取,返回失败。
take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到
BlockingQueue有新的数据被加入;
drainTo():一次性从BlockingQueue获取所有可用的数据对象(还可以指定获取数据的个数),
通过该方法,可以提升获取数据效率;不需要多次分批加锁或释放锁。
package com.test.concurrent; import java.util.Random; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; public class Producer implements Runnable { private volatile boolean isRunning = true; private BlockingQueue queue; private static AtomicInteger count = new AtomicInteger(); private static final int DEFAULT_RANGE_FOR_SLEEP = 1000; public Producer(BlockingQueue queue) { this.queue = queue; } @Override public void run() { String data = null; Random r = new Random(); System.out.println("启动生产者线程!"); try { while (isRunning) { System.out.println("正在生产数据..."); Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP)); data = "data:" + count.incrementAndGet(); System.out.println("将数据:" + data + "放入队列..."); if (!queue.offer(data, 2, TimeUnit.SECONDS)) { System.out.println("放入数据失败:" + data); } } } catch (InterruptedException e) { e.printStackTrace(); Thread.currentThread().interrupt(); } finally { System.out.println("退出生产者线程!"); } } public void stop() { isRunning = false; } }
package com.test.concurrent; import java.util.Random; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; public class Consumer implements Runnable{ private BlockingQueue<String> queue; private static final int DEFAULT_RANGE_FOR_SLEEP = 1000; public Consumer(BlockingQueue<String> queue) { this.queue = queue; } @Override public void run() { System.out.println("启动消费者线程!"); Random r = new Random(); boolean isRunning = true; try { while (isRunning) { System.out.println("正从队列获取数据..."); String data = queue.poll(2, TimeUnit.SECONDS); if (null != data) { System.out.println("拿到数据:" + data); System.out.println("正在消费数据:" + data); Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP)); } else { // 超过2s还没数据,认为所有生产线程都已经退出,自动退出消费线程。 isRunning = false; } } } catch (InterruptedException e) { e.printStackTrace(); Thread.currentThread().interrupt(); } finally { System.out.println("退出消费者线程!"); } } }
package com.test.concurrent; import java.util.concurrent.*; import java.util.Random; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; public class BlockingQueueTest { public static void main(String[] args) throws InterruptedException { // 声明一个容量为10的缓存队列 BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10); Producer producer1 = new Producer(queue); Producer producer2 = new Producer(queue); Producer producer3 = new Producer(queue); Consumer consumer = new Consumer(queue); // 借助Executors ExecutorService service = Executors.newCachedThreadPool(); // 启动线程 service.execute(producer1); service.execute(producer2); service.execute(producer3); service.execute(consumer); // 执行10s Thread.sleep(10 * 1000); producer1.stop(); producer2.stop(); producer3.stop(); Thread.sleep(2000); // 退出Executor service.shutdown(); } }
相关推荐
Java多线程工具篇BlockingQueue的详解 BlockingQueue是Java多线程工具篇中的一种高效安全的队列类,在多线程环境中扮演着关键角色。它能够高效解决多线程中数据传输的问题,使得数据由队列的一端输入,从另外一端...
Java多线程是Java编程中的重要概念,尤其在如今的多核处理器环境下,理解并熟练掌握多线程技术对于提高程序性能和响应速度至关重要。本资料详细讲解了Java多线程的原理,并提供了丰富的实战代码,非常适合Java初学者...
Java多线程是Java编程中的重要概念,它允许程序同时执行多个任务,极大地提升了程序的效率和性能。在Java中,实现多线程有两种主要方式:通过实现Runnable接口或者继承Thread类。本案例将深入探讨Java多线程中的关键...
这份“JAVA多线程编程技术PDF”是学习和掌握这一领域的经典资料,涵盖了多线程的全部知识点。 首先,多线程的核心概念包括线程的创建与启动。在Java中,可以通过实现Runnable接口或继承Thread类来创建线程。创建后...
在Java编程中,多线程并发是提升程序执行效率、充分利用多核处理器资源的重要手段。本文将基于"java 多线程并发实例"这个主题,深入探讨Java中的多线程并发概念及其应用。 首先,我们要了解Java中的线程。线程是...
《Java多线程编程实战指南》这本书深入浅出地讲解了Java多线程的核心概念和实战技巧,分为核心篇和设计模式篇,旨在帮助开发者掌握并应用多线程技术。 1. **线程基础** - **线程的创建**:Java提供了两种创建线程...
《Java多线程编程实战指南-核心篇》是一本深入探讨Java并发编程的书籍,旨在帮助读者掌握在Java环境中创建、管理和同步线程的核心技术。Java的多线程能力是其强大之处,使得开发者能够在同一时间执行多个任务,提高...
Java多线程技术是Java编程中的重要组成部分,它允许程序同时执行多个任务,极大地提高了程序的效率和响应性。在现代计算机系统中,多线程是实现并发处理的关键技术,尤其在服务器端应用和高性能计算中不可或缺。 ...
Java多线程下载器是一种利用Java编程语言实现的高效文件下载工具,它通过将大文件分割成多个部分并同时下载,显著提高了下载速度。在Java中实现多线程下载器涉及许多关键概念和技术,包括线程、并发控制、网络I/O...
Java多线程是Java编程中的核心概念,它允许程序同时执行多个任务,提高了系统的效率和响应性。在Java中,多线程的实现主要通过两种方式:继承Thread类和实现Runnable接口。理解并掌握多线程的使用对于任何Java开发者...
总之,Java多线程设计模式是每个Java开发者必备的技能之一。深入学习并熟练运用这些模式,将有助于你编写出更高效、稳定和易于扩展的多线程应用程序。这个PDF版教程和源代码集合是你学习多线程设计模式的理想资源,...
Java多线程是Java编程中的核心概念,尤其对于高级开发者来说,掌握多线程的深入理解和应用至关重要。这本书“java多线程进阶”显然旨在帮助读者深化这方面的理解,打通编程中的“任督二脉”,使开发者能够更加熟练地...
Java多线程是Java编程中的重要概念,它允许程序同时执行多个任务,从而提升系统效率和资源利用率。本文将深入探讨Java多线程机制,包括线程的创建、同步、通信以及常见设计模式。 首先,Java中创建线程主要有两种...
【Java多线程简单下载器】是一个初学者的编程作业,虽然代码可能较为混乱,但其核心功能已经实现,即通过多线程技术进行文件的下载。在Java中,多线程是并发处理的重要手段,它允许多个任务在同一时间执行,从而提高...
Java多线程是Java编程语言中的一个重要特性,它允许程序同时执行多个任务,极大地提高了程序的效率和响应性。在现代计算机系统中,多核处理器的普及使得多线程技术成为提升性能的关键手段。本篇将深入探讨Java多线程...
Java多线程是Java编程中的一个核心概念,它允许程序同时执行多个独立的任务,从而提高应用程序的效率和响应性。在Java中,多线程主要通过两种方式实现:继承Thread类和实现Runnable接口。这份"JAVA多线程的PPT和示例...
在`queueFactory`压缩包中,可能包含了示例代码或者类库,展示了如何使用队列工厂模式来创建和管理不同类型的队列,特别是针对Java多线程环境的优化。这些资源可以帮助开发者更好地理解和应用多线程和队列技术,提高...
Java多线程是Java编程中的核心概念,它允许程序同时执行多个任务,极大地提升了软件的效率和并发性。这份“Java多线程的经典资料.rar”压缩包包含了一份名为“Java线程.pdf”的文档,很可能是关于Java多线程的详细...
Java多线程设计模式是Java开发中的核心概念,它涉及到如何高效、安全地在多个执行线程之间共享资源和协调任务。设计模式是解决特定问题的成熟方案,它们是编程经验的结晶,可以帮助开发者在面临多线程挑战时快速找到...
在Java编程中,多线程是一种重要的技术,它允许程序同时执行多个任务,提升系统效率。在并发编程中,"生产者-消费者"模式是一种经典的解决问题的范式,用于协调两个或更多线程间的协作,其中一部分线程(生产者)...