RapidMiner Studio之Process源码分析

RapidMiner的核心处理类是Process，其是Operator的集合，而ProcessRenderer是负责Process上的事件。

com.rapidminer.Process是用观察者模式实现的，这里套用下《JAVA与模式》一书中对观察者模式的定义：观察者模式定义了一种一对多的依赖关系，让多个观察者对象同时监听某一个主题对象。这个主题对象在状态上发生变化时，会通知所有观察者对象，使它们能够自动更新自己。

下面来看下Process类的实现：

Observable接口：

package com.rapidminer.tools;

import java.util.NoSuchElementException;

/** 这个接口是Java观察者模式的重新实现，相比于jdk实现，该接口有如下改进:
* 接口与实现分离
* 更新方法上添加了泛型参数支持
* 最重要一点: 更新方法可以被Event Dispatch Thread (EDT)调用.
* 线程同步的
* 定义了异常处理方法
* */
public interface Observable<A> {

/** 添加一个observer，并且如果onEDT参数是true的话，会被通知到EDT，如果observer参数是null，则抛出NoSuchElementException异常
*  @throws a NPE if observer is null. */
public void addObserver(Observer<A> observer, boolean onEDT);

/** 删除一个observer，如果这个observer没有注册过，则抛出NoSuchElementException异常
*  @throws    NoSuchElementException if observer is not registered with this Observable. */
public void removeObserver(Observer<A> observer);

/** 与addObserver功能相同, 只是添加observer到列表最前端
*/
void addObserverAsFirst(Observer<A> observer, boolean onEDT);

}

Observer接口（观察者）：

package com.rapidminer.tools;

/**
* @see Observable
* @author Simon Fischer
*
*/
public interface Observer<A> {

public void update(Observable<A> observable, A arg);

}

AbstractObservable：是Observable接口的实现类

package com.rapidminer.tools;

import java.util.LinkedList;
import java.util.List;
import java.util.NoSuchElementException;

import javax.swing.SwingUtilities;
/**
* @author Simon Fischer
*/
public class AbstractObservable<A> implements Observable<A> {
//普通的观察者
private final LinkedList<Observer<A>> observersRegular = new LinkedList<Observer<A>>();
//启用EDT的观察者
private final LinkedList<Observer<A>> observersEDT  = new LinkedList<Observer<A>>();

//锁对象，同步时候在此对象上加锁
private final Object lock = new Object();

@Override
public void addObserverAsFirst(Observer<A> observer, boolean onEDT) {
if (observer == null) {
throw new NullPointerException("Observer is null.");
}
if (onEDT) {
synchronized (lock) {
observersEDT.addFirst(observer);
}
} else {
synchronized (lock) {
observersRegular.addFirst(observer);
}
}
}

@Override
public void addObserver(Observer<A> observer, boolean onEDT) {
if (observer == null) {
throw new NullPointerException("Observer is null.");
}
if (onEDT) {
synchronized (lock) {
observersEDT.add(observer);
}
} else {
synchronized (lock) {
observersRegular.add(observer);
}
}
}

@Override
public void removeObserver(Observer<A> observer) {
boolean success = false;
synchronized (lock) {
success |= observersRegular.remove(observer);
success |= observersEDT.remove(observer);
}
if (!success) {
throw new NoSuchElementException("No such observer: "+observer);
}
}

/** Updates the observers in the given list. */
private void fireUpdate(List<Observer<A>> observerList, A argument) {
for (Observer<A> observer : observerList) {
observer.update(this, argument);
}
}

/** Equivalent to <code>fireUpdate(null)</code>. */
protected void fireUpdate() {
fireUpdate(null);
}

/** Updates all observers with the given argument. */
protected void fireUpdate(final A argument) {
// lists are copied in order to avoid ConcurrentModification occurs if updating
// an observer triggers insertion of another
List<Observer<A>> copy;
synchronized (lock) {
copy = new LinkedList<Observer<A>>(observersRegular);
}
fireUpdate(copy, argument);
if (!observersEDT.isEmpty()) {
final List<Observer<A>> copyEDT;
synchronized (lock) {
copyEDT = new LinkedList<Observer<A>>(observersEDT);
}
SwingUtilities.invokeLater(new Runnable() {
@Override
public void run() {
fireUpdate(copyEDT, argument);
}
});
}
}
}

下面来看Process类：

成员变量：

MainFrame中的NEW_ACTION属性是负责响应创建新的process动作的，我们以此为入口，分析process类是如何工作的。

通过前面action的分析我们已经知道，Action中的actionPerformed方法是用来响应事件的，NewAction类的方法很简单，就是调用MainFrame中的newProcess方法来创建一个process窗口

public void newProcess() {
//在创建新的process之前，询问是否要停止正在运行（暂停）的process
if (getProcessState() == Process.PROCESS_STATE_RUNNING || getProcessState() == Process.PROCESS_STATE_PAUSED) {
if (SwingTools.showConfirmDialog("close_running_process",
ConfirmDialog.YES_NO_OPTION) == ConfirmDialog.NO_OPTION) {
return;
}
}

//关闭当前打开的process
if (close(false)) {
// undoindex标识置为0，undolist置空，undoaction和redoaction禁用
resetUndo();

//停止当前运行的process（因为是以线程方式运行的，只有正在运行的operator完成之后，process线程才会退出）
stopProcess();
changed = false;

//创建新的process，设置并注册监听
setProcess(new Process(), true);
addToUndoList();
if (!"false".equals(ParameterService.getParameterValue(PROPERTY_RAPIDMINER_GUI_SAVE_ON_PROCESS_CREATION))) {
SaveAction.save(getProcess());
}
SAVE_ACTION.setEnabled(false);
}
}