`
378629846
  • 浏览: 215699 次
  • 性别: Icon_minigender_1
  • 来自: 哈尔滨
社区版块
存档分类
最新评论

并行计算框架的Java实现--系列三

    博客分类:
  • java
 
阅读更多

接上篇并行计算框架的Java实现--系列

 

优化锁,之前的锁是采用一个static的Object实现的,这样会有一个问题,如果我创建了多个Executer,那么所有Job都会持有一把锁,既影响性能,也容易出现死锁的情况。所以,改成每个Executer持有一把锁。

Executer代码如下:

 public class Executer {

	//存储任务的执行结果
	private List<Future<Object>> futres = new ArrayList<Future<Object>>(); 
	//条件队列锁,以及线程计数器
	public final Lock lock = new Lock();
	//线程池
	private ExecutorService pool = null;
	public Executer() {
		this(1);
	}
	public Executer(int threadPoolSize) {
		pool = Executors.newFixedThreadPool(threadPoolSize);
	}
	/**
	 * 任务派发
	 * @param job
	 */
	public void fork(Job job){
		//设置同步锁
		job.setLock(lock);
		//将任务派发给线程池去执行
		futres.add(pool.submit(job));
		//增加线程数
		synchronized (lock) {
			lock.thread_count++;
		}
	}
	/**
	 * 统计任务结果
	 */
	public List<Object> join(){
		synchronized (lock) {
			while(lock.thread_count > 0){//检查线程数,如果为0,则表示所有任务处理完成
//				System.out.println("threadCount: "+THREAD_COUNT);
				try {
					lock.wait();//如果任务没有全部完成,则挂起。等待完成的任务给予通知
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
		List<Object> list = new ArrayList<Object>();
		//取出每个任务的处理结果,汇总后返回
		for (Future<Object> future : futres) {
			try {
				Object result = future.get();//因为任务都已经完成,这里直接get
				if(result != null){
					if(result instanceof List)
						list.addAll((List)result);
					else
						list.add(result);
				}
			} catch (Exception e) {
				e.printStackTrace();
			} 
		}
		return list;
	}
}
 

 Job类:

 

public abstract class Job implements Callable<Object> {

	//锁
	private Lock lock = null;

	void setLock(Lock lock) {
		this.lock = lock;
	}

	public Object call() throws Exception {
		Object result = null;
		try{
			result = this.execute();//执行子类具体任务
		}catch(Exception e){
			e.printStackTrace();
		}
		synchronized (lock) {
			//处理完业务后,任务结束,递减线程数,同时唤醒主线程
			lock.thread_count--;
			lock.notifyAll();
		}
		return result;
	}
	/**
	 * 业务处理函数
	 */
	public abstract Object execute();
	
}
class Lock {
	
	int thread_count;
	
}
 

 测试结果:

 

threadCount: 10
running thread id = 8
running thread id = 10
running thread id = 9
running thread id = 12
running thread id = 11
threadCount: 8
threadCount: 7
threadCount: 6
threadCount: 5
running thread id = 12
running thread id = 8
running thread id = 11
threadCount: 2
running thread id = 10
threadCount: 1
running thread id = 9
ResultSize: 10
time: 2001

 OK!

这样每个Executer就可以使用自己的lock,而相互不受同步的影响

 

2
2
分享到:
评论
2 楼 378629846 2013-05-14  
golly2009 写道
...如果我创建了多个Executer,那么所有Job都会持有一把锁,既影响性能,也容易出现死锁的情况。....

应该是所有job都会持有同一把锁吧,因为是静态成员,类在虚拟机中只有一份,所有Executer类的对象共享。
可以设计成单例不?系统中只有一个Executer实例,这样就不会出现死锁,另外你只有主线程在等待,可以把notifyAll换成notify不?

不能是单例的,因为一个executer是一个锁来控制的,如果是单例的,这个executer就只能执行一批job了,不能复用了
1 楼 golly2009 2013-05-08  
...如果我创建了多个Executer,那么所有Job都会持有一把锁,既影响性能,也容易出现死锁的情况。....

应该是所有job都会持有同一把锁吧,因为是静态成员,类在虚拟机中只有一份,所有Executer类的对象共享。
可以设计成单例不?系统中只有一个Executer实例,这样就不会出现死锁,另外你只有主线程在等待,可以把notifyAll换成notify不?

相关推荐

    并行计算框架的Java实现--系列二

    本系列的第二部分将深入探讨如何利用Java语言构建并行计算框架,以提高程序的运行效率。在本文中,我们将关注源码分析和相关工具的使用,以帮助开发者更好地理解和应用这些技术。 首先,Java为并行计算提供了丰富的...

    java8并行计算示例--可动态配置--简化框架结构--并行计算框架从五个类简化到两个类

    这个案例展示了如何将一个原本复杂的并行计算框架精简为两个关键类,从而实现更简洁、高效的编程模型。 首先,我们要理解并行计算的基本概念。并行计算是指同时使用多个处理器或计算机来处理任务,以减少总体完成...

    JAVAOpenMP并行计算框架

    3. **JavaOpenMP实现**:由于Java自身没有OpenMP规范,因此JavaOpenMP通常指的是第三方库或工具,它们试图模仿OpenMP的接口和行为,比如JOpenMP、JaOMP等。这些库通常通过JNI(Java Native Interface)调用底层的...

    JAVA并行计算的一些资料 论文

    综上所述,Java并行计算涵盖了从基础的多线程到高级的分布式计算框架,开发者可以通过这些工具和概念,设计和实现高效的并行程序,处理大规模的数据和计算任务。这些论文和资料应该会深入探讨这些主题,并可能包含...

    java8并行计算示例

    在压缩包文件`parallelTest`中,可能包含了实现上述并行计算的Java代码示例,可以作为学习和实践Java 8并行计算的参考。通过分析和理解这些代码,可以进一步深入理解并行计算的工作机制和优化技巧。总的来说,Java 8...

    基于Web的Java并行计算

    JET(Java Environment for Tasks)平台是一个旨在简化基于Web的并行计算过程的软件框架。它通过以下方式实现了高效的并行计算: - **Java Applet支持**:利用Java Applet作为客户端界面,用户可以通过简单的Web...

    一种对象化并行计算框架

    在大数据环境下,针对数据型统计分析系统性能劣化明显、不能满足用户使用需求的问题,本文提出了一种轻量级高性能对象化并行计算架构,研制了该架构的对象服务组件、对象管理服务组件和客户端代理组件,并将该架构和...

    OnJava8-Examples-3.0_soucecode_java_

    流可以进行过滤、映射、归约等一系列操作,支持串行和并行计算。例如,`list.stream().filter(x -&gt; x &gt; 10).collect(Collectors.toList())`将筛选出列表中大于10的元素。 4. **默认方法**:接口在Java 8中引入了...

    无锁并行框架Amino -- Concurrent Building Blocks

    在实际应用中,Amino通常与高级并行编程模型如C++的std::thread库、Java的Fork/Join框架或Actor模型结合使用,以实现更复杂的工作负载调度和任务分解。开发者可以通过阅读Amino的文档、示例代码和相关教程,学习如何...

    分布式与并行计算—Java实现并向算法.ZIP

    算法中的并行使用java的Fork / Join框架实现,他会将进程使用ForkJoinPool进行管理,并自动分配到空闲的CPU核心上来运算。由于个人PC的CPU核心数量较少,所以预期至多能产生常数倍的加速 效果。 本次实验使用的实验...

    java实现k-means算法

    在处理大规模数据时,还可以考虑使用分布式计算框架如 Apache Spark 来提升效率。 综上所述,通过 Java 和相关库,我们可以有效地在 MySQL 数据库中应用 K-Means 算法,进行数据聚类,并将结果存入新表,从而实现对...

    [并行计算——结构·算法·编程]

    - 大数据分析:并行计算在Hadoop、Spark等框架中的应用,加速数据处理。 - 科学计算:如气候模拟、生物信息学分析、物理模拟等领域的高性能计算。 - 机器学习与人工智能:并行计算提升深度学习模型的训练速度。 ...

    JAVA并发编程深度学习-无锁并行计算框架1

    在Java并发编程中,无锁并行计算框架如Disruptor提供了一种高效且低延迟的方式来处理高并发场景。无锁技术避免了线程之间的竞争条件,从而提升了多线程环境下的性能。本节我们将深入探讨Disruptor框架以及与...

    MPJ并行编程框架的实现及安装配置.pdf

    MPJ作为一种将MPI引入Java的并行编程框架,极大地丰富了Java在并行计算领域的应用。通过上述介绍,我们可以看到MPJ不仅在设计上充分考虑了并行计算的特点,还在实现机制和技术特征上做出了诸多创新。对于希望从事...

    并行计算的介绍

    #### 三、并行计算的应用领域 1. **科学计算**:如气候模拟、分子动力学等。 2. **大数据处理**:数据挖掘、机器学习算法等。 3. **图形渲染**:实时渲染、电影特效制作等。 4. **金融建模**:风险评估、期权定价等...

    pi.rar_PI_java 并行计算_并行计算 pi

    在实际应用中,除了Java,还有其他并行计算框架,如Hadoop和Spark,它们适用于大数据处理和分布式计算。对于更复杂的并行算法,如MapReduce,可以处理更大规模的数据并行计算。 总的来说,这个实验为学习者提供了一...

    并行计算--并发构造纵览

    7. **MapReduce**:Google提出的分布式计算框架,用于大规模数据集的并行处理,由Hadoop等开源项目实现。 8. **Actor模型**:每个Actor是独立的计算实体,通过异步消息传递进行通信,确保并发安全,如Erlang和Akka...

    K-Means算法java实现

    对于大数据集,可以考虑使用多线程或分布式计算框架(如Apache Spark)来并行化K-Means算法,以提升性能。 8. **错误处理和调试**: Java实现应包含适当的异常处理,以防止数据读取错误、无效输入等情况导致程序...

    java72-java-advance.zip

    8. **并发改进**:Java 7对并发API进行了一些优化,如Fork/Join框架,用于实现高效的并行计算。此外,`ConcurrentHashMap`的性能也有所提升。 9. **改进的数组初始化**:Java 7允许在数组初始化时使用紧凑的语法,...

    java-1.7.0-openjdk

    - **多线程并发工具的增强**:包括Fork/Join框架,用于并行计算,以及新的并发集合类。 - **类型推断**:在泛型中引入类型推断,简化代码编写。 - **字符串内联**:对字符串操作进行优化,提高性能。 - **动态...

Global site tag (gtag.js) - Google Analytics