利用JPPF进行并行计算，计算任务运行在远端节点上，那么如何收集运行在远端的任务日志，用于跟踪和分析呢？

JPPF框架对此也有封装，主要的实现思路是，通过自定义实现一个log4j的appender，对外提供JMX服务。客户端（监控端）实现一个监听器，监听远端日志，这样即可把远端日志采集到本地进行统一的管理。这对于我们收集和管理并行计算实时日志是非常有用的。具体看一下：

在没个执行任务的Node节点，修改log4j的配置，启用jppf提供的JmxAppender：
log4j-node.properties

log4j.appender.JMX=org.jppf.logging.log4j.JmxAppender
log4j.appender.JMX.layout=org.apache.log4j.PatternLayout
log4j.appender.JMX.layout.ConversionPattern=%d [%-5p][%c.%M(%L)]: %m\n

### set log levels - for more verbose logging change 'info' to 'debug' ###
#log4j.rootLogger=DEBUG, JPPF
log4j.rootLogger=INFO, JPPF, JMX

客户端代码如下：

/**
* @author lihzh
* @alia OneCoder
* @blog http://www.coderli.com
* @date 2013年7月17日 上午10:34:44
*/
public class RemoteLog {
     public static void main(String args[]) throws Exception {
            // get a JMX connection to the node MBean server
           JMXNodeConnectionWrapper jmxNode = new JMXNodeConnectionWrapper("localhost" ,
                     12001);
           jmxNode.connectAndWait(500000L);
            // get a proxy to the MBean
           JmxLogger nodeProxy = jmxNode.getProxy(JmxLogger.DEFAULT_MBEAN_NAME ,
                     JmxLogger. class);
            // use a handback object so we know where the log messages come from
           String source = "node   " + jmxNode.getHost() + ":" + jmxNode.getPort();
            // subbscribe to all notifications from the MBean
           NotificationListener listener = new MyLoggingHandler();
           nodeProxy.addNotificationListener(listener, null, source);
           String source167 = "node   " + jmxNode167.getHost() + ":" + jmxNode167.getPort();
     }
}

public class MyLoggingHandler implements NotificationListener {
            // handle the logging notifications
            public void handleNotification(Notification notification,
                     Object handback) {
                String message = notification.getMessage();
                String toDisplay = handback.toString() + ": " + message;
                System. out.println(toDisplay);
           }
     }

简单介绍下，就是通过ip和端口与需要监听的node节点建立jmx链接，然后讲自己事先的监听处理类MyLoggingHandler实例，注册到nodeProxy上。这样如果node端有日志出现，就会通知到监听端，可以进行自己的分类处理。试验一下，启动JPPF环境和启动监听，运行之前介绍的JPPF任务，可以看到在本地成功打印出任务日志信息,这样，并行计算任务分节点的日志收集处理也就很容易实现了。

不算什么深入的研究，主要是了解下JPPF中类的加载和隔离机制。

JPPF中类的加载采用的是分布式类加载技术。这样既可在Node节点运行在node上并不存在的类。也就是类可以仅在用户的Client端存在。

如图，JPPF的class loader大致分三层。
System class loader是由JVM控制的加载器，用于启动node节点。在大多数JVM中是，该loader是java.net.URLClassloader的实例。
Server class loader是类AbstractJPPFClassloader的实现，提供了远程访问server端classpath中的类和其他资源的功能。该loader在node连接到server端的时候创建，断开连接的时候销毁。其父loader为System class loader。
Client class loader也是AbstractJPPFClassloader类的实现，提供了远程访问一个或多个客户端上的类和资源的功能。该loader在客户端提交任务的时候创建。一个node可能会持有很多client class loader。

JPPF的node节点仅持有一个与Server的连接，该连接被所有的loader共享，这样可以避免并发时潜在的的一致性，同步和冲突等问题。

Java的classloader是先从parent loader开始加载的饿，下图展示了node节点加载类的过程

node节点先从server节点请求class。如果server节点存在class，则直接加载返回。
如果class仅在client端存在，则会先访问server，server无，再通过server定位client，访问client端的class。

在JPPF框架里，每个client都有一个唯一的UUID，用于唯一标识该client。Node节点便可根据此UUID识别出当前classloader读取的是哪个client的class

每个Server也有唯一的UUID标识，这样就可以实现链式类加载：

需要注意的是，在这种情况下，node节点仍然只持有一个server classloader，仅连接跟它直接关联的server。

UUID与classloader
使用client端UUID，即可实现不同的client端在执行同一个任务的时候的classloader隔离，因为uuid不同，会使用不同的classloader加载类。不过，如果同一个UUID提交了两个不同版本的任务class，则有可能出现冲突和错误。    
你也可以手动指定client端的UUID，这样classloader即可重用。哪怕你重连client也可以重用同一个classloader。不过，此时随之而来的问题就是在客户端修改的版本可能不会在node端自动生效，不同的版本还可能造成冲突。这时，你可以重启node或者更换UUID使node重新加载class。

任务预部署
对于大任务，如果担心每次走网络效率低，根据上面介绍的classloader机制，自然想到可以预先将任务文件部署到server或者node节点。不过此时需要自己管理好部署的版本，所谓各有利弊。不过，如果任务版本稳定的话，建议采用此种方式。

classloader缓存池
为了避免加载过多的类造成内存溢出，JPPF在node端增加了classloader缓存池大小的设置，如果超过上限，则会销毁最老的classloader。

jppf.classloader.cache.size = n

本地缓存资源文件
当调用getResourceAsStream(),  getResource(),getResources() or getMultipleResources()方法的时候，class loader会缓存非类定义的文件至内存或者本地文件系统中。这样会避免再次请求该文件时候的网络访问，提高效率。你可以指定缓存在内存中还是文件系统中：

# either “file” (the default) or “memory“
jppf.resource.cache.storage = file

文件的保存路径当然也是可以配置的，默认是系统临时文件夹System.getProperty("java.io.tmpdir").：

jppf.resource.cache.dir = some_directory

JPPF server中的类缓存

JPPF的服务端会在内存中缓存被classloader加载进来的类和其他资。这样就可以避免与客户端的频繁的网络访问，从而提高执行速度。你也无需担心内存溢出的问题，因为这种缓存采用的是软引用，这样缓存的类就可以在必要的时候被JVM的垃圾回收机制回收。在一般情况，该缓存还是可以显著的提升效率的。

Class loader代理模型
JPPF默认采用的是”父优先“的类加载模型，也就是上面我们介绍的模型。上述模型虽然好用，但是类的加载顺序就相对固定，也就不能最大化的优化加载效率。因此，JPPF通过继承URLClassLoader，复写并开放addURL接口（jdk里是protected，JPPF为public），使得用户的自己开发任务可以直接调用该方法，形成对classloader的一种扩展。用户可调用addURL方法，指定自己想要加载类的地址。此时JPPF classloader的加载顺序会出现变化，从大的步骤看，会先查找指定的URL classpath：

* 从clientclassloader入手：代理到server的classloader
* server的classloader: 先仅查找用户指定的URL classpath
* 如果找到类，则结束查找
* 否则，回到client的classloader，也仅查找URL的classpath
* 如果找到，结束查找。

此书，如果仍未找到，则回到前面介绍的通过网络的查找顺序。
概括来说，就是如果URL优先的代理模型激活，则node会先从本地层级中查找URL指定的classpath，然后在网络上通过server查找。有三种指定代理模型的方式，最简单的就是直接配置在node节点的配置文件中：

  # possible values: parent | url, defaults to parent
  jppf.classloader.delegation = parent

下载文件

如我们之前的介绍的，我们可以通过addURL(URL)方法，动态指定加载类的URL地址，那么自然想到，我们可以通过先将文件下载到本地，然后指定URL地址，从而进行本地加载，以提到效率。在AbstractJPPFClassLoader类中，有方法：
public URL[] getMultipleResources(final String...names)
可以帮你同时下载多个文件到本地。

以上只是我了解到的JPPF中关于类加载部分的设计和实现。从已有掌握来看，JPPF考虑的还是非常全面周到的，隔离、效率、一致性都有考虑，应该说对并行计算来说，很有保障。

先说一个挺有意思的事情，就在OneCoder准备记录试用过程的时候，给大家截图下载页面的时候，发现最新版本变成3.3.4了。于是，我也只好重新下载了：）

关于JPPF的介绍，可访问其官网：http://www.jppf.org
下载页面：http://www.jppf.org/downloads.php

想要运行JPPF并行计算任务，需要至少一个Node节点(执行任务的节点)，一个driver节点(调度任务的节点)，和本地调用节点

依次分别运行driver包里的startDriver.bat和node包里的startNode.bat启动driver和node
 

然后，开发jppf应用，JPPF在application-template包中提供了任务模版，只需参照文档稍作修改即可：

 public static void main(String args[]) {
           TaskFlowOne taskOne = new TaskFlowOne();
            try {
                 // create the JPPFClient. This constructor call causes JPPF to read
                 // the configuration file
                 // and connect with one or multiple JPPF drivers.
                 jppfClient = new JPPFClient();
                 // create a JPPF job
                JPPFJob job = new JPPFJob();
                 // give this job a readable unique id that we can use to monitor and
                 // manage it.
                job.setName( "Template Job Id");
                 // add a task to the job.
                job.addTask(taskOne);
                job.setBlocking( true);

                 // Submit the job and wait until the results are returned.
                 // The results are returned as a list of JPPFTask instances,
                 // in the same order as the one in which the tasks where initially
                 // added the job.
                List<JPPFTask> results = jppfClient.submit(job);
                JPPFTask jTask = results.get(0);
                System. out.println(jTask.getResult());
                 // process the results
                TaskFlowTwo taskTwo = new TaskFlowTwo();
                job.addTask(taskTwo);
                job.setBlocking( true);
                List<JPPFTask> resultsTow = jppfClient.submit(job);
                JPPFTask jTaskTow = resultsTow.get(1);
                System. out.println(jTaskTow.getResult());
           } catch (Exception e) {
                e.printStackTrace();
           } finally {
                 if ( jppfClient != null)
                      jppfClient.close();
           }
     }

一个并行计算的任务，就是一个JPPFJob，每个job含有多个task，task之间是并行执行的，如果有多个nod节点，会按照其负载均衡策略分配到多个node上执行。通过client提交(submit)job后，会返回执行结果，为一个JPPFTask列表，列表里包含你add到job中每个task及其运行结果，作为后续计算处理之用。任务调用支持阻塞和非阻塞。

admin-ui中还提供了对节点状态和资源占用，任务信息等监控，总体来说上手很快，很好用：

最近用Hessian4.0.7做文件上传，先给出自己做试验的样例代码，写在tomcat7下，采用servlet3.0，配置代码如下：

**
 * 基于Servlet3.0的，相当于以前<b>web.xml</b>配置文件的配置类
 * 
 * @author OneCoder
 * @Blog http://www.coderli.com
 * @date 2012-9-30 下午1:16:59
 */
publicclass DefaultWebApplicationInitializer implements
WebApplicationInitializer {

@Override
publicvoid onStartup(ServletContext appContext) throws ServletException {
AnnotationConfigWebApplicationContext rootContext = new AnnotationConfigWebApplicationContext();
rootContext.register(DefaultAppConfig.class);

appContext.addListener(new ContextLoaderListener(rootContext));
ServletRegistration.Dynamic hessianServlet = appContext.addServlet("hessian", new HessianServlet());
hessianServlet.addMapping("/hessian");
hessianServlet.setInitParameter("service-class", HessianFileUploader.class.getName());
//ServletRegistration.Dynamic dispatcher = appContext.addServlet(
//"dispatcher", new DispatcherServlet(rootContext));
//dispatcher.setLoadOnStartup(1);
//dispatcher.addMapping("/");
//// Spring Security 过滤器配置
//FilterRegistration.Dynamic securityFilter = appContext.addFilter(
//"springSecurityFilterChain", DelegatingFilterProxy.class);
//securityFilter.addMappingForUrlPatterns(null, false, "/*");
}}

定义接口，并在服务端实现，Hessian4.0开始支持流作为参数传递，以前则采用byte[]方式传递：

public interface FileUploader {

void uplaodFile(String fileName，InputStream is);

}


public class HessianFileUploader implements FileUploader {

@Override
publicvoid uplaodFile(String fileName，InputStream is) {
try {
OutputStream out = new FileOutputStream("/Users/apple/Desktop/" + fileName);
int nLength = 0;
byte[] bData = newbyte[1024];
while (-1 != (nLength = is.read(bData))) {
out.write(bData, 0, nLength);
}
out.close();
} catch (FileNotFoundException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
} finally {
try {
is.close();
} catch (IOException e) {
e.printStackTrace();}}
}

}

客户端调用：

public class FileUploaderTest {
private FileUploader uploader;
private HessianProxyFactory factory = new HessianProxyFactory();
@Test publicvoid testFileUploader() throws FileNotFoundException, MalformedURLException {
uploader = (FileUploader) factory.create(FileUploader.class, "http://localhost:8080/onecoder-shurnim/hessian");
InputStream is = new FileInputStream("/Users/apple/git/onecoder-java/onecoder-shurnim/src/main/resources/logback.xml"); uploader.uplaodFile(is, "logback.xml"); }
 
}

很简单方便。这里曾遇到一个奇怪的问题，如果接口的参数顺序调换，即InputStream在前则会报错：

com.caucho.hessian.io.HessianProtocolException: uplaodFile: expected string at 0x3c (<)

Hessian协议没有深入研究，不知道是不是一个约定或是要求。开发时需要注意。

如果集成spring，只需将servlet交由spring的代理里即可，修改配置即可：

<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd"> 
<beans>         
<bean id="defaultHandlerMapping" class="org.springframework.web.servlet.handler.BeanNameUrlHandlerMapping"/>        
 <bean id="fileUploader" class="com.coderli.shurnim.file.HessianFileUploader"/>         
<bean name="/hello" class="org.springframework.remoting.caucho.HessianServiceExporter">                 
     <property name="service" ref="fileUploader"/> 
                <property name="serviceInterface" value="com.coderli.shurnim.file.FileUploader"/>        
 </bean> 
</beans>

亲测spring3.2.2+Hessian4.0.7可用。文件内容保存完整。

记录卸载过程和问题。现有环境Cloudera Manager + (1 + 2 )的CDH环境。

1、先在Manage管理端移除所有服务。
2、删除Manager Server
在Manager节点运行

$ sudo /usr/share/cmf/uninstall-cloudera-manager.sh

如果没有该脚本，则可以手动删除，先停止服务：

sudo service cloudera-scm-server stop
sudo service cloudera-scm-server-db stop

然后删除：

sudo yum remove cloudera-manager-server
sudo yum remove cloudera-manager-server-db

3、删除所有CDH节点上的CDH服务，先停止服务：

sudo service cloudera-scm-agent hard_stop

卸载安装的软件：

sudo yum remove 'cloudera-manager-*' hadoop hue-common 'bigtop-*'

4、删除残余数据：

sudo rm -Rf /usr/share/cmf /var/lib/cloudera* /var/cache/yum/cloudera*

5、kill掉所有Manager和Hadoop进程（选作，如果你正确停止Cloud Manager和所有服务则无须此步）

$ for u in hdfs mapred cloudera-scm hbase hue zookeeper oozie hive impala flume; do sudo kill $(ps -u $u -o pid=); done

6、删除Manager的lock文件
在Manager节点运行：

sudo rm /tmp/.scm_prepare_node.lock

至此，删除完成。
 

1、安装JDK。之前在部署Hadoop的时候已经安装完成。
2、下载解压HBase。
3、配置HBase数据存储路径，虽然单机模式可以使用本地文件系统，不过OneCoder还是配置HDFS文件系统。
修改hbase-site.xml

<configuration>
    <property>
        <name>hbase.rootdir</name>
        <value>hdfs://localhost:40000/hbase</value>
    </property>
    <property>
        <name>hbase.cluster.distributed</name>
        <value>true</value>
    </property>
</configuration>

这里hdfs路径跟之前hadoop环境中core-site.xml的配置保持一直。
4、修改conf/hbase-site.xml。配置JAVA_HOME
5、启动HBase

shell>bin/start-hbase.sh

6、验证安装

shell>bin/hbase shell
hbase>status

hbase(main):001:0> status

2013-04-07 22:58:24.203 java[3826:1703] Unable to load realm info from SCDynamicStore
1 servers, 0 dead, 2.0000 average load

部署成功。这里还是存在java的警告。同样配置了-Djava.security.krb5.realm=OX.AC.UK -Djava.security.krb5.kdc=kdc0.ox.ac.uk:kdc1.ox.ac.uk 仍然无法消除。暂且如此。

事务日志使得事务更加高效。每次数据改变的时候，存储引擎可以在内存中修改数据拷贝而不用每次都修改磁盘上数据。这速度很快。随后存储引擎会修改记录写入事务日志，该日志是在磁盘上的从而完成持久化。这也是相对高效的过程。因为，追加日志事件产生的是连续的小范围磁盘上的I/O操作，而不是大范围的随机I/O操作。然后，在随后的某时刻一个进程会更新磁盘上的表。因此，更多存储引擎都采用该技术(被称作：write-ahead loggin WAL) 避免将修改两次写入磁盘。

如果在写入日志之后，写入表数据之前发生了崩溃，存储引擎在重启后也可以进行恢复。恢复机制不通的存储引擎是不同的。

当多个事务同时持有和请求同一资源上的锁而产生循环依赖的时候就产生了死锁。死锁发生在事务试图以不同的顺序锁定资源。以StockPrice表上的两个事务为例：

事务1
START TRANSACTION;
UPDATE StockPrice SET close = 45.50 WHERE stock_id = 4 and date = '2002-05-01';
UPDATE StockPrice SET close = 19.80 WHERE stock_id = 3 and date = '2002-05-02';
COMMIT;
事务 #2
START TRANSACTION;
UPDATE StockPrice SET high = 20.12 WHERE stock_id = 3 and date = '2002-05-02';
UPDATE StockPrice SET;
COMMIT;

如果不走运的话，每个事务都可以执行完第一个语句，并在过程中锁住资源。然后每个事务都试图去执行第二行语句，当时却发现它被锁住了。两个事务将永远的等待对方完成，除非有其他原因打断死锁。

为了解决这个问题，数据库实现了各种死锁探查和超时机制。像InnoDB这样复杂的存储引擎会提示循环依赖并且立即返回错误。否则死锁将会导致查询非常缓慢。其他一些不好的做法是等待超时后放弃。当前InnoDB处理死锁的方式是回滚持有最少排他行级锁的事务。（几乎最简单的回滚的参考指标）

锁的行为是顺序是存储引擎决定的。因此，一些存储引擎可能会在特定的操作顺序下发生死锁，其他的可能没有。死锁有两种：一些是因为实际数据冲突而无法避免，一些是因为存储引擎的工作方式产生。

只有部分或者完全回滚其中的一个事务才可能打破死锁。死锁是事务系统中客观存在的事实，你的应该在设计上必须应该考虑处理死锁。一些业务系统可以从头重试事务。

OneCoder在自己的笔记本上部署Hadoop环境用于研究学习，记录部署过程和遇到的问题。

1、安装JDK。
2、下载Hadoop(1.0.4)，在Hadoop中配置JAVA_HOME环境变量。修改hadoop-env.sh文件。
export JAVA_HOME= /Library/Java/JavaVirtualMachines/jdk1.7.0_10.jdk/Contents/Home/
3、配置SSH
生成密钥

ssh-keygen -t dsa -P '' -f ~/.ssh/onecoder_dsa

将公钥追加到key中

cat ~/.ssh/onecoder_rsa.pub >> ~/.ssh/authorized_keys

打开Mac OS的远程访问选项。系统设置 -》 共享 -》 远程登录

4、配置namenode和datanode hdfs本地路径
在hdfs-site.xml中配置

<property>
 <name>dfs.name.dir</name>
 <value>/Users/apple/Documents/hadoop/name/</value>
</property>
<property>
   <name>dfs.data.dir</name>
  <value>/Users/apple/Documents/hadoop/data/</value>
</property>
 <property> 
    <name>dfs.replication</name> 
    <value>1</value> 
 </property>

由于是验证环境，所以数据备份设为1。

5、格式化namenode

bin/Hadoop NameNode -format

6、启动hadoop

可通过bin下的start-all.sh直接全部启动，也可以通过

hadoop namenode (datanode、jobtracker、tasktracker)

，按照上述顺序依次启动。
使用后者可以比较方便的查看启动日志，方便查错。并且也可以在控制台日志中看到监控页面的访问地址和端口。如：
13/04/04 15:52:18 INFO http.HttpServer: Jetty bound to port 50070
当然，这些地址你可以已经烂熟于胸了。通过浏览器进入web监控页面查看。可以看到一切就绪。Hadoop环境部署完成。

其他。
关于警告：Unable to load realm info from SCDynamicStore
网上给出的解决方案是：在hadoop-env.sh中设置

export HADOOP_OPTS="-Djava.security.krb5.realm=OX.AC.UK -Djava.security.krb5.kdc=kdc0.ox.ac.uk:kdc1.ox.ac.uk"

不过在OneCoder这里无效，但是不影响Hadoop环境使用。

HBase在扫描数据的时候，使用scanner表扫描器。HTable通过一个Scan实例，调用getScanner(scan)来获取扫描器。可以配置扫描起止位，以及其他的过滤条件。通过迭代器返回查询结果，使用起来虽然不是很方便，不过并不复杂。但是这里有一点可能被忽略的地方，就是返回的scanner迭代器，每次调用next的获取下一条记录的时候，默认配置下会访问一次RegionServer。这在网络不是很好的情况下，对性能的影响是很大的。测试中，未配置前，一个业务的消耗时间为：

Cost time: 159941

通过：

scan.setCaching(10000);

指定一次取出10000条记录后，该业务的消耗时间为：

Cost time: 64845

因为该 业务访问数据次数很多，所以效果很明显。

也有说可通过修改配置项hbase.client.scanner.caching的值，来使该配置生效。不过，OneCoder这里在hbase-site.xml中增加了该配置却没有生效。