溪石的博客

[置頂]移植jQuery deferred到java，基于java的promise編程模型

                                  移植jQuery deferred到java，基于java的promise編程模型

 

很多語言都支持promise編程模型，像是scala中promise類和jquery（javascript）中的deferred對象等，在java中好像缺少相關實現(xiàn)。筆者不得以，只能自己動手弄了一個。最后選擇將jquery中的deferred對象移植到java中來的方案。目前已經(jīng)應用在企業(yè)級項目的高性能服務器和android客戶端等項目中。

Promise編程模型的概念這里也不再贅述，大家自己上網(wǎng)查找即可。這種編程模型主要解決的問題就是“同步調(diào)用變異步的問題”，通常解決異步調(diào)用的方式是使用“回調(diào)”。但普通回調(diào)的使用在代碼書寫，返回值傳遞和“異步方法編排上”非常的不方便。所以才會有Promise模型的誕生。

這次會介紹java版的deferred對象的使用方法，以及用jquery版之間的變化和改進。目前開放的版本是基于線程池的版本，正在開發(fā)基于akka的版本。在jquery的實現(xiàn)中，因為javascript是單線程的，所以不用考慮線程同步的問題。在java線程池的版的deferred里，基于多線程環(huán)境做了很多測試，保證了線程安全及可靠性。

 

一．            基本調(diào)用形式

final Deferred  def = new Deferred (App. executor);

執(zhí)行某個異步調(diào)用，比如某個基于網(wǎng)絡的異步服務

callService(new Response(){

            public void onMessage(Object message){

            def.resolve(message);

}

Public void onFail(Exception e){

            def.reject(e);

}

});

你可以在構造Deferred 對象后的任意時候，使用def的then方法。比如

                        def.then(new Reply(){

             public Object done(Object d) {

                  System.out.println("response:"+d);

                  return d;

             }

             public void fail(Object f) {

                  System.out.println("error:"+f);

             }

});

一個經(jīng)常遇到的場景是callService后將def作為參數(shù)傳遞到其他方法，在其他方法內(nèi)部再決定def要綁定什么樣的后續(xù)動作，也就是綁定什么樣的then。

                      注意then方法的定義public Object done(Object d)，在實際使用中done通常是以“處理鏈”的方式來使用的，即你會看到def.then().then().then()…這樣的方式，每一個then的done方法接收的參數(shù)都是其上一個then的done方法的返回值。通常作為參數(shù)傳遞給某個方法的Deferred上面已經(jīng)綁定了一些默認的then對象，來處理一些必要的步驟。比如對接收報文的初步解碼。

             注意同在Reply接口中fail方法是沒有返回值的，一旦異步處理鏈上的某個Deferred被reject，其本身及后面所有的Deferred綁定的then都會被觸發(fā)fail方法。這保證了整個業(yè)務編排上或是你精心設計的算法編排上任意一個環(huán)節(jié)，無論如何都會得到響應，這也是Promise模型關于異常的最重要的處理方式。

             Promise編程模型本身是強健的，但異步服務卻不是總能得到響應。在實際應用中，每一個作為計算或業(yè)務環(huán)節(jié)的Deferred都應該被定時輪詢，以保證在異步服務徹底得不到響應的時候（比如你執(zhí)行了一個數(shù)據(jù)庫查詢，但過了很長很長時間仍沒有得到回應），可以給Deferred對象reject一個超時錯誤。

              響應處理對象then中方法done和fail都是不允許拋出任何異常的，特別是done方法，如果你的算法依賴異常，請在done中加上try…catch，并將異常傳換成下一個then可以理解的信息，以便這個Deferred處理鏈中可以正常執(zhí)行下去。

 

二． pipe到另外一個異步處理流程上去

假如你有如下的業(yè)務場景，你需要順序調(diào)用三個異步的webservice服務來得到最終的返回結果，其中沒個webservice的入?yún)⒍己蜕弦粋€的異步返回結果相關。（注意，異步的webservice是調(diào)用之后，服務端立刻返回，服務端處理完成后再主動訪問剛才的請求方返回結果的方式）如果將這種webservice調(diào)用封裝成同步方法無疑在編程上是非常方便的，可以使用我們平常寫程序時順序的書寫方式，比如

reval1 = callwebservice1(param0)

reval2 = callwebservice2(reval1)

reval3 = callwebservice3(reval2)

    方便的同時卻犧牲了性能。調(diào)用線程要在callwebservice方法內(nèi)阻塞，以等待異步返回。這樣的編程方法無法滿足高性能及高并發(fā)的需要。那么有沒有既能類似于平常寫程序時順序的書寫方式又能滿足異步無阻塞的需要呢，這就是Promise編程模型本身要解決的最大問題。

通常解決這種問題的方式是使用pipe，pipe這個方法名稱的由來應該是來自于linux shell的管道符，即“|”

使用Deferred對象的解決方案類似于如下：

Deferred.resolvedDeferred(App.executor,param0).pipe(new AsyncRequest2(){

    public void apply(Object param0,final Deferred newDefered) throws Exception{

asyncCallwebservice1(param0).onResponse(new Response(){

    public void onMessage(String message){

         newDefered.resolve(message);

}

});

}

}).pipe(new AsyncRequest2(){

    public void apply(Object reval1,final Deferred newDefered) throws Exception{

asyncCallwebservice2(reval1).onResponse(new Response(){

    public void onMessage(String message){

         newDefered.resolve(message);

}

});

}

}).pipe(new AsyncRequest2(){

    public void apply(Object reval2,final Deferred newDefered) throws Exception{

asyncCallwebservice3(reval3).onResponse(new Response(){

    public void onMessage(String message){

         newDefered.resolve(message);

}

});

}

}).then(new new Reply(){

                 public Object done(Object d) {

                      //在這里消費最終結果

return d;

                 }

                 public void fail(Object f) {

                             

}

});

使用Deferred對象提供的方案好處就是，所有的調(diào)用都是異步的，上面這一連串代碼立刻就會返回。所有的業(yè)務編排會按照書寫順序在線程池中的線程里被調(diào)用，你也不必擔心返回值結果和參數(shù)傳遞過程中的線程安全問題，框架在關鍵位置都做了同步，也做了相當多的測試用于驗證。

可以看出，對于異步方法調(diào)用而言，比較難以解決的問題是異步算法的編排問題。Deferred對象為異步算法提供了很好的解決方案。

相較于AsyncRequest2類還有一個AsyncRequest1類，接口如下：

public interface AsyncRequest1<R> {

                 public Deferred apply(R result) throws Exception;

}

這個類要求在在apply方法中要自己創(chuàng)建Deferred對象。

 

三. 一些小改進

    相較于傳統(tǒng)promise編程模型，在java多線程環(huán)境下做了一些小升級。這里主要介紹synchronize方法

Synchronize方法簽名如下:

Deferred synchronize(ExecutorService executor,Deferred... deferreds)

實際上，synchronize方法將眾多的Deferred對象的完成狀態(tài)同歸集到一個唯一的Deferred對象上去，即如果所有的Deferred對象參數(shù)都resolved了，作為最終結果的Deferred也resolve，如果眾多的Deferred對象參數(shù)有一個reject了，最終的那個Deferred也會立即reject(其他參數(shù)的狀態(tài)都舍棄)。

這個方法一般用于多個并行流程最終狀態(tài)的“歸并”中。

除了synchronize，框架還提供一些傳統(tǒng)promise編程模型沒有的改進，比如pipe4fail和source等。

 

四．在android項目中的應用

    （略）

https://github.com/jonenine/javaDeferred

posted @ 2017-07-23 11:59 溪石 閱讀(839) | 評論 (0) | 編輯 收藏

[置頂]HDAOOP SIMPLIZE TOOLKIT hadoop mapreduce簡化開發(fā)包

https://github.com/jonenine/HST

 

雖然大數(shù)據(jù)的發(fā)展已經(jīng)將近10個年頭了，hadoop技術仍然沒有過時，特別是一些低成本，入門級的小項目，使用hadoop還是蠻不錯的。而且，也不是每一個公司都有能力招聘和培養(yǎng)自己的spark人才。

         我本人對于hadoop mapreduce是有一些意見的。hadoop mapreduce技術對于開發(fā)人員的友好度不高，程序難寫，調(diào)試困難，對于復雜的業(yè)務邏輯遠沒有spark得心應手。

2016年的春節(jié)前接到一個任務，要在一個沒有spark的平臺實現(xiàn)電力系統(tǒng)的一些統(tǒng)計分析算法，可選的技術只有hadoop mapreduce。受了這個刺激之后產(chǎn)生了一些奇思妙想，然后做了一些試驗，并最終形成HST---hadoop simplize toolkit，還真是無心載柳柳成蔭啊。

 

HST基本優(yōu)點如下：

屏蔽了hadoop數(shù)據(jù)類型，取消了driver，將mapper和reducer轉(zhuǎn)化為transformer和joiner，業(yè)務邏輯更接近sql。相當程度的減少了代碼量，極大的降低了大數(shù)據(jù)編程的門檻，讓基層程序員通過簡單的學習即可掌握大數(shù)據(jù)的開發(fā)。

克服了hadoop mapreduce數(shù)據(jù)源單一的情況，比如在一個job內(nèi)，input可以同時讀文件和來自不同集群的hbase。

遠程日志系統(tǒng)，讓mapper和reducer的日志集中到driver的控制臺，極大減輕了并行多進程程序的調(diào)試難度。

克服了hadoop mapreduce編寫業(yè)務邏輯時，不容易區(qū)分數(shù)據(jù)來自哪個數(shù)據(jù)源的困難。接近了spark（或者sql）的水平。

天生的多線程執(zhí)行，即在mapper和reducer端都默認使用多線程來執(zhí)行業(yè)務邏輯。

對于多次迭代的任務，相連的兩個任務可以建立關聯(lián)，下一個任務直接引用上一個任務的結果，使多次迭代任務的代碼結構變得清晰優(yōu)美。

 

以下會逐條說明

 

基本概念的小變化：

   Source類代替了hadoop Input體系(format，split和reader)

   Transformer代替了mapper

   Joiner代替了Reducer

   去掉了飽受詬病的Driver，改為內(nèi)置的實現(xiàn)，現(xiàn)在完全不用操心了。

                                                                          

1.      基本上，屏蔽了hadoop的數(shù)據(jù)類型，使用純java類型

在原生的hadoop mapreduce開發(fā)中，使用org.apache.hadoop.io包下的各種hadoop數(shù)據(jù)類型，比如hadoop的Text類型，算法的編寫中一些轉(zhuǎn)換非常不方便。而在HST中一律使用java基本類型，完全屏蔽了hadoop類型體系。

比如在hbase作為source（Input）的時候，再也不用直接使用ImmutableBytesWritable和Result了，HST為你做了自動的轉(zhuǎn)換。

現(xiàn)在的mapper（改名叫Transformer了）風格是這樣的

public static class TransformerForHBase0 extends HBaseTransformer<Long>

…

現(xiàn)在map方法叫flatmap，看到?jīng)]，已經(jīng)幫你自動轉(zhuǎn)成了string和map

public void flatMap(String key, Map<String, String> row,

           Collector<Long> collector)

 

可閱讀xs.hadoop.iterated.IteratedUtil類中關于類型自動轉(zhuǎn)換的部分

2.      克服了hadoop mapreduce數(shù)據(jù)源單一的情況。比如在一個job內(nèi)，數(shù)據(jù)源同時讀文件和hbase，這在原生的hadoop mapreduce是不可能做到的

 

以前訪問hbase，需要使用org.apache.hadoop.hbase.client.Scan和TableMapReduceUtil，現(xiàn)在完全改為與spark相似的方式。

現(xiàn)在的風格是這樣的：

Configuration conf0 = HBaseConfiguration.create();

        conf0.set("hbase.zookeeper.property.clientPort", "2181");

        conf0.set("hbase.zookeeper.quorum", "172.16.144.132,172.16.144.134,172.16.144.136");

        conf0.set(TableInputFormat.INPUT_TABLE,"APPLICATION_JOBS");

        conf0.set(TableInputFormat.SCAN_COLUMN_FAMILY,"cf");

        conf0.set(TableInputFormat.SCAN_CACHEBLOCKS,"false");

conf0.set(TableInputFormat.SCAN_BATCHSIZE,"20000");

...其他hbase的Configuration，可以來自不同集群。

 

IteratedJob<Long> iJob = scheduler.createJob("testJob")

                .from(Source.hBase(conf0), TransformerForHBase0.class)

              .from(Source.hBase(conf1), TransformerForHBase1.class)

               .from(Source.textFile("file:///home/cdh/0.txt"),Transformer0.class)

        .join(JoinerHBase.class)

 

Hadoop中的input,現(xiàn)在完全由source類來代替。通過內(nèi)置的機制轉(zhuǎn)化為inputformat，inputsplit和reader。在HST的框架下，其實可以很容易的寫出諸如Source.dbms(),Source.kafka()以及Source.redis()方法。想想吧，在一個hadoop job中，你終于可以將任意數(shù)據(jù)源，例如來自不同集群的HBASE和來自數(shù)據(jù)庫的source進行join了，這是多么happy的事情啊！

 

3.      遠程日志系統(tǒng)。讓mapper和reducer的日志集中在driver進行顯示，極大減輕了了并行多進程程序的調(diào)試難度

 

各位都體驗過，job fail后到控制臺頁面，甚至ssh到計算節(jié)點去查看日志的痛苦了吧。對，hadoop原生的開發(fā)，調(diào)試很痛苦的呢！

現(xiàn)在好了，有遠程日志系統(tǒng)，可以在調(diào)試時將mapper和reducer的日志集中在driver上，錯誤和各種counter也會自動發(fā)送到driver上，并實時顯示在你的控制臺上。如果在eclipse中調(diào)試程序，就可以實現(xiàn)點擊console中的錯誤，直接跳到錯誤代碼行的功能嘍！

Ps：有人可能會問，如何在集群外使用eclipse調(diào)試一個job，卻可以以集群方式運行呢？這里不再贅述了，網(wǎng)上有很多答案的哦

 

4.      克服了hadoop mapreduce在join上，區(qū)分數(shù)據(jù)來自哪個數(shù)據(jù)源的困難，接近spark（或者sql）的水平

在上面給出示例中，大家都看到了，現(xiàn)在的mapper可以綁定input嘍！，也就是每個input都有自己獨立的mapper。正因為此，現(xiàn)在的input和mapper改名叫Source和Transformer。

那么，大家又要問了，在mapper中，我已經(jīng)可以輕松根據(jù)不同的數(shù)據(jù)輸入寫出不同的mapper了，那reducer中怎么辦，spark和sql都是很容易實現(xiàn)的哦？比如看人家sql

Select a.id,b.name from A  a,B  b where a.id = b.id

多么輕松愉悅啊!

在原生hadoop mapreduce中，在reducer中找出哪個數(shù)據(jù)對應來自哪個input可是一個令人抓狂的問題呢！

現(xiàn)在這個問題已經(jīng)被輕松解決嘍！看下面這個joiner，對應原生的reducer

public static class Joiner0 extends Joiner<Long, String, String>

…

Reduce方法改名叫join方法，是不是更貼近sql的概念呢？

        public void join(Long key,RowHandler handler,Collector collector) throws Exception{

                List<Object> row  = handler.getSingleFieldRows(0);//對應索引為0的source

         List<Object> row2 = handler.getSingleFieldRows(1);//對應第二個定義的source

注意上面兩句，可以按照數(shù)據(jù)源定義的索引來取出來自不同數(shù)據(jù)源join后的數(shù)據(jù)了，以后有時間可能會改成按照別名來取出，大家看源碼的時候，會發(fā)現(xiàn)別名這個部分的接口都寫好了，要不你來幫助實現(xiàn)了吧。

 

5.      天生的多線程執(zhí)行，即在mapper和reducer端都默認使用多線程來執(zhí)行業(yè)務邏輯。

    看看源碼吧，HST框架是并發(fā)調(diào)用flatMap和join方法的，同時又不能改變系統(tǒng)調(diào)用reduce方法的順序(否則hadoop的辛苦排序可就白瞎了)，這可不是一件容易的事呢!

看到這里，有的同學說了。你這個HST好是好，但你搞的自動轉(zhuǎn)換類型這個機制可能會把性能拉下來的。這個嗎，不得不承認，可能是會有一點影響。但在生產(chǎn)環(huán)境做的比對可以證明，影響太小了，基本忽略不計。

筆者在生產(chǎn)環(huán)境做了做了多次試驗，mapper改成多線程后性能并未有提高，特別是對一些業(yè)務簡單的job，增加Transformer中的并發(fā)級別效率可能還會下降。

很多同學喜歡在mapper中做所謂“mapper端的join”。這種方式，相信在HST中通過提高mapper的并發(fā)級別后會有更好的表現(xiàn)。

Reducer中的性能相對原生提升的空間還是蠻大的。大部分的mapreduce項目，都是mapper簡單而reducer復雜，HST采用并發(fā)執(zhí)行join的方式對提升reducer性能是超好的。

 

6.      對于多次迭代的任務，相連的兩個任務可以建立關聯(lián)，在流程上的下一個job直接引用上一個job的結果，使多次迭代任務的代碼結構變得清晰優(yōu)美

雖然在最后才提到這一點，但這卻是我一開始想要寫HST原因。多次迭代的任務太麻煩了，上一個任務要寫在hdfs做存儲，下一個任務再取出使用，麻煩不麻煩。如果都由程序自動完成，豈不美哉！

在上一個任務里format一下

IteratedJob<Long> iJob = scheduler.createJob("testJob")

...//各種source定義

.format("f1","f2")

在第二個任務中，直接引用

IteratedJob<Long> stage2Job = scheduler.createJob("stage2Job")

.fromPrevious(iJob, Transformer2_0.class);

                  //Transformer2_0.class

 

       public static class Transformer2_0 extends PreviousResultTransformer<Long>

       ...

           public void flatMap(Long inputKey, String[] inputValues,Collector<Long> collector) {

           String f1 = getFiledValue(inputValues, "f1");

           String f2 = getFiledValue(inputValues, "f2");

    看到?jīng)]，就是這么簡單。

在最開始的計劃中，我還設計了使用redis隊列來緩沖前面job的結果，供后面的job作為輸入。這樣本來必須嚴格串行的job可以在一定程度上并發(fā)。另外還設計了子任務的并發(fā)調(diào)度，這都留給以后去實現(xiàn)吧。

 

7. 便捷的自定義參數(shù)傳遞。

有時候，在業(yè)務中需要作一些“開關變量”，在運行時動態(tài)傳入不同的值以實現(xiàn)不同的業(yè)務邏輯。這個問題HST框架其實也為你考慮到了。

Driver中的自定義參數(shù)，source中的自定義參數(shù)都會以內(nèi)置的方式傳到transformer或joiner中去，方便程序員書寫業(yè)務。

   查看transformer或joiner的源碼就會發(fā)現(xiàn)：

getSourceParam(name)和getDriverParam(pIndex)方法，在計算節(jié)點輕松的得到在driver和source中設置的各層次級別的自定義參數(shù),爽吧!

 

8. 其他工具

HST提供的方便還不止以上這些，比如在工具類中還提供了兩行數(shù)據(jù)（map類型）直接join的方法。這些都留給你自己去發(fā)現(xiàn)并實踐吧!
https://github.com/jonenine/HST

posted @ 2017-01-29 10:10 溪石 閱讀(1751) | 評論 (0) | 編輯 收藏

[置頂]在web開發(fā)中更加富有效率的使用JSON

     摘要: /Files/jonenine/blog.rar本文一共分3個部分，介紹了在web中如何靈活而高效使用json，并同大家分享了實際的代碼，希望大家喜歡。（一）   在jvm中使用jsonjson的使用在當下已經(jīng)普及起來了，作者使用json也已經(jīng)有很長的時間了，早在它普及之前就嘗試著在項目中應用它。今天給大家介紹的代碼都是實際項目中真實的代碼，雖然不盡完美，但卻禁得住實...  閱讀全文

posted @ 2014-12-06 14:08 溪石 閱讀(11003) | 評論 (1) | 編輯 收藏

移植jQuery deferred到java，基于java的promise編程模型

                                  移植jQuery deferred到java，基于java的promise編程模型

 

很多語言都支持promise編程模型，像是scala中promise類和jquery（javascript）中的deferred對象等，在java中好像缺少相關實現(xiàn)。筆者不得以，只能自己動手弄了一個。最后選擇將jquery中的deferred對象移植到java中來的方案。目前已經(jīng)應用在企業(yè)級項目的高性能服務器和android客戶端等項目中。

Promise編程模型的概念這里也不再贅述，大家自己上網(wǎng)查找即可。這種編程模型主要解決的問題就是“同步調(diào)用變異步的問題”，通常解決異步調(diào)用的方式是使用“回調(diào)”。但普通回調(diào)的使用在代碼書寫，返回值傳遞和“異步方法編排上”非常的不方便。所以才會有Promise模型的誕生。

這次會介紹java版的deferred對象的使用方法，以及用jquery版之間的變化和改進。目前開放的版本是基于線程池的版本，正在開發(fā)基于akka的版本。在jquery的實現(xiàn)中，因為javascript是單線程的，所以不用考慮線程同步的問題。在java線程池的版的deferred里，基于多線程環(huán)境做了很多測試，保證了線程安全及可靠性。

 

一．            基本調(diào)用形式

final Deferred  def = new Deferred (App. executor);

執(zhí)行某個異步調(diào)用，比如某個基于網(wǎng)絡的異步服務

callService(new Response(){

            public void onMessage(Object message){

            def.resolve(message);

}

Public void onFail(Exception e){

            def.reject(e);

}

});

你可以在構造Deferred 對象后的任意時候，使用def的then方法。比如

                        def.then(new Reply(){

             public Object done(Object d) {

                  System.out.println("response:"+d);

                  return d;

             }

             public void fail(Object f) {

                  System.out.println("error:"+f);

             }

});

一個經(jīng)常遇到的場景是callService后將def作為參數(shù)傳遞到其他方法，在其他方法內(nèi)部再決定def要綁定什么樣的后續(xù)動作，也就是綁定什么樣的then。

                      注意then方法的定義public Object done(Object d)，在實際使用中done通常是以“處理鏈”的方式來使用的，即你會看到def.then().then().then()…這樣的方式，每一個then的done方法接收的參數(shù)都是其上一個then的done方法的返回值。通常作為參數(shù)傳遞給某個方法的Deferred上面已經(jīng)綁定了一些默認的then對象，來處理一些必要的步驟。比如對接收報文的初步解碼。

             注意同在Reply接口中fail方法是沒有返回值的，一旦異步處理鏈上的某個Deferred被reject，其本身及后面所有的Deferred綁定的then都會被觸發(fā)fail方法。這保證了整個業(yè)務編排上或是你精心設計的算法編排上任意一個環(huán)節(jié)，無論如何都會得到響應，這也是Promise模型關于異常的最重要的處理方式。

             Promise編程模型本身是強健的，但異步服務卻不是總能得到響應。在實際應用中，每一個作為計算或業(yè)務環(huán)節(jié)的Deferred都應該被定時輪詢，以保證在異步服務徹底得不到響應的時候（比如你執(zhí)行了一個數(shù)據(jù)庫查詢，但過了很長很長時間仍沒有得到回應），可以給Deferred對象reject一個超時錯誤。

              響應處理對象then中方法done和fail都是不允許拋出任何異常的，特別是done方法，如果你的算法依賴異常，請在done中加上try…catch，并將異常傳換成下一個then可以理解的信息，以便這個Deferred處理鏈中可以正常執(zhí)行下去。

 

二． pipe到另外一個異步處理流程上去

假如你有如下的業(yè)務場景，你需要順序調(diào)用三個異步的webservice服務來得到最終的返回結果，其中沒個webservice的入?yún)⒍己蜕弦粋€的異步返回結果相關。（注意，異步的webservice是調(diào)用之后，服務端立刻返回，服務端處理完成后再主動訪問剛才的請求方返回結果的方式）如果將這種webservice調(diào)用封裝成同步方法無疑在編程上是非常方便的，可以使用我們平常寫程序時順序的書寫方式，比如

reval1 = callwebservice1(param0)

reval2 = callwebservice2(reval1)

reval3 = callwebservice3(reval2)

    方便的同時卻犧牲了性能。調(diào)用線程要在callwebservice方法內(nèi)阻塞，以等待異步返回。這樣的編程方法無法滿足高性能及高并發(fā)的需要。那么有沒有既能類似于平常寫程序時順序的書寫方式又能滿足異步無阻塞的需要呢，這就是Promise編程模型本身要解決的最大問題。

通常解決這種問題的方式是使用pipe，pipe這個方法名稱的由來應該是來自于linux shell的管道符，即“|”

使用Deferred對象的解決方案類似于如下：

Deferred.resolvedDeferred(App.executor,param0).pipe(new AsyncRequest2(){

    public void apply(Object param0,final Deferred newDefered) throws Exception{

asyncCallwebservice1(param0).onResponse(new Response(){

    public void onMessage(String message){

         newDefered.resolve(message);

}

});

}

}).pipe(new AsyncRequest2(){

    public void apply(Object reval1,final Deferred newDefered) throws Exception{

asyncCallwebservice2(reval1).onResponse(new Response(){

    public void onMessage(String message){

         newDefered.resolve(message);

}

});

}

}).pipe(new AsyncRequest2(){

    public void apply(Object reval2,final Deferred newDefered) throws Exception{

asyncCallwebservice3(reval3).onResponse(new Response(){

    public void onMessage(String message){

         newDefered.resolve(message);

}

});

}

}).then(new new Reply(){

                 public Object done(Object d) {

                      //在這里消費最終結果

return d;

                 }

                 public void fail(Object f) {

                             

}

});

使用Deferred對象提供的方案好處就是，所有的調(diào)用都是異步的，上面這一連串代碼立刻就會返回。所有的業(yè)務編排會按照書寫順序在線程池中的線程里被調(diào)用，你也不必擔心返回值結果和參數(shù)傳遞過程中的線程安全問題，框架在關鍵位置都做了同步，也做了相當多的測試用于驗證。

可以看出，對于異步方法調(diào)用而言，比較難以解決的問題是異步算法的編排問題。Deferred對象為異步算法提供了很好的解決方案。

相較于AsyncRequest2類還有一個AsyncRequest1類，接口如下：

public interface AsyncRequest1<R> {

                 public Deferred apply(R result) throws Exception;

}

這個類要求在在apply方法中要自己創(chuàng)建Deferred對象。

 

三. 一些小改進

    相較于傳統(tǒng)promise編程模型，在java多線程環(huán)境下做了一些小升級。這里主要介紹synchronize方法

Synchronize方法簽名如下:

Deferred synchronize(ExecutorService executor,Deferred... deferreds)

實際上，synchronize方法將眾多的Deferred對象的完成狀態(tài)同歸集到一個唯一的Deferred對象上去，即如果所有的Deferred對象參數(shù)都resolved了，作為最終結果的Deferred也resolve，如果眾多的Deferred對象參數(shù)有一個reject了，最終的那個Deferred也會立即reject(其他參數(shù)的狀態(tài)都舍棄)。

這個方法一般用于多個并行流程最終狀態(tài)的“歸并”中。

除了synchronize，框架還提供一些傳統(tǒng)promise編程模型沒有的改進，比如pipe4fail和source等。

 

四．在android項目中的應用

    （略）

https://github.com/jonenine/javaDeferred

posted @ 2017-07-23 11:59 溪石 閱讀(839) | 評論 (0) | 編輯 收藏

HDAOOP SIMPLIZE TOOLKIT hadoop mapreduce簡化開發(fā)包

https://github.com/jonenine/HST

 

雖然大數(shù)據(jù)的發(fā)展已經(jīng)將近10個年頭了，hadoop技術仍然沒有過時，特別是一些低成本，入門級的小項目，使用hadoop還是蠻不錯的。而且，也不是每一個公司都有能力招聘和培養(yǎng)自己的spark人才。

         我本人對于hadoop mapreduce是有一些意見的。hadoop mapreduce技術對于開發(fā)人員的友好度不高，程序難寫，調(diào)試困難，對于復雜的業(yè)務邏輯遠沒有spark得心應手。

2016年的春節(jié)前接到一個任務，要在一個沒有spark的平臺實現(xiàn)電力系統(tǒng)的一些統(tǒng)計分析算法，可選的技術只有hadoop mapreduce。受了這個刺激之后產(chǎn)生了一些奇思妙想，然后做了一些試驗，并最終形成HST---hadoop simplize toolkit，還真是無心載柳柳成蔭啊。

 

HST基本優(yōu)點如下：

屏蔽了hadoop數(shù)據(jù)類型，取消了driver，將mapper和reducer轉(zhuǎn)化為transformer和joiner，業(yè)務邏輯更接近sql。相當程度的減少了代碼量，極大的降低了大數(shù)據(jù)編程的門檻，讓基層程序員通過簡單的學習即可掌握大數(shù)據(jù)的開發(fā)。

克服了hadoop mapreduce數(shù)據(jù)源單一的情況，比如在一個job內(nèi)，input可以同時讀文件和來自不同集群的hbase。

遠程日志系統(tǒng)，讓mapper和reducer的日志集中到driver的控制臺，極大減輕了并行多進程程序的調(diào)試難度。

克服了hadoop mapreduce編寫業(yè)務邏輯時，不容易區(qū)分數(shù)據(jù)來自哪個數(shù)據(jù)源的困難。接近了spark（或者sql）的水平。

天生的多線程執(zhí)行，即在mapper和reducer端都默認使用多線程來執(zhí)行業(yè)務邏輯。

對于多次迭代的任務，相連的兩個任務可以建立關聯(lián)，下一個任務直接引用上一個任務的結果，使多次迭代任務的代碼結構變得清晰優(yōu)美。

 

以下會逐條說明

 

基本概念的小變化：

   Source類代替了hadoop Input體系(format，split和reader)

   Transformer代替了mapper

   Joiner代替了Reducer

   去掉了飽受詬病的Driver，改為內(nèi)置的實現(xiàn)，現(xiàn)在完全不用操心了。

                                                                          

1.      基本上，屏蔽了hadoop的數(shù)據(jù)類型，使用純java類型

在原生的hadoop mapreduce開發(fā)中，使用org.apache.hadoop.io包下的各種hadoop數(shù)據(jù)類型，比如hadoop的Text類型，算法的編寫中一些轉(zhuǎn)換非常不方便。而在HST中一律使用java基本類型，完全屏蔽了hadoop類型體系。

比如在hbase作為source（Input）的時候，再也不用直接使用ImmutableBytesWritable和Result了，HST為你做了自動的轉(zhuǎn)換。

現(xiàn)在的mapper（改名叫Transformer了）風格是這樣的

public static class TransformerForHBase0 extends HBaseTransformer<Long>

…

現(xiàn)在map方法叫flatmap，看到?jīng)]，已經(jīng)幫你自動轉(zhuǎn)成了string和map

public void flatMap(String key, Map<String, String> row,

           Collector<Long> collector)

 

可閱讀xs.hadoop.iterated.IteratedUtil類中關于類型自動轉(zhuǎn)換的部分

2.      克服了hadoop mapreduce數(shù)據(jù)源單一的情況。比如在一個job內(nèi)，數(shù)據(jù)源同時讀文件和hbase，這在原生的hadoop mapreduce是不可能做到的

 

以前訪問hbase，需要使用org.apache.hadoop.hbase.client.Scan和TableMapReduceUtil，現(xiàn)在完全改為與spark相似的方式。

現(xiàn)在的風格是這樣的：

Configuration conf0 = HBaseConfiguration.create();

        conf0.set("hbase.zookeeper.property.clientPort", "2181");

        conf0.set("hbase.zookeeper.quorum", "172.16.144.132,172.16.144.134,172.16.144.136");

        conf0.set(TableInputFormat.INPUT_TABLE,"APPLICATION_JOBS");

        conf0.set(TableInputFormat.SCAN_COLUMN_FAMILY,"cf");

        conf0.set(TableInputFormat.SCAN_CACHEBLOCKS,"false");

conf0.set(TableInputFormat.SCAN_BATCHSIZE,"20000");

...其他hbase的Configuration，可以來自不同集群。

 

IteratedJob<Long> iJob = scheduler.createJob("testJob")

                .from(Source.hBase(conf0), TransformerForHBase0.class)

              .from(Source.hBase(conf1), TransformerForHBase1.class)

               .from(Source.textFile("file:///home/cdh/0.txt"),Transformer0.class)

        .join(JoinerHBase.class)

 

Hadoop中的input,現(xiàn)在完全由source類來代替。通過內(nèi)置的機制轉(zhuǎn)化為inputformat，inputsplit和reader。在HST的框架下，其實可以很容易的寫出諸如Source.dbms(),Source.kafka()以及Source.redis()方法。想想吧，在一個hadoop job中，你終于可以將任意數(shù)據(jù)源，例如來自不同集群的HBASE和來自數(shù)據(jù)庫的source進行join了，這是多么happy的事情啊！

 

3.      遠程日志系統(tǒng)。讓mapper和reducer的日志集中在driver進行顯示，極大減輕了了并行多進程程序的調(diào)試難度

 

各位都體驗過，job fail后到控制臺頁面，甚至ssh到計算節(jié)點去查看日志的痛苦了吧。對，hadoop原生的開發(fā)，調(diào)試很痛苦的呢！

現(xiàn)在好了，有遠程日志系統(tǒng)，可以在調(diào)試時將mapper和reducer的日志集中在driver上，錯誤和各種counter也會自動發(fā)送到driver上，并實時顯示在你的控制臺上。如果在eclipse中調(diào)試程序，就可以實現(xiàn)點擊console中的錯誤，直接跳到錯誤代碼行的功能嘍！

Ps：有人可能會問，如何在集群外使用eclipse調(diào)試一個job，卻可以以集群方式運行呢？這里不再贅述了，網(wǎng)上有很多答案的哦

 

4.      克服了hadoop mapreduce在join上，區(qū)分數(shù)據(jù)來自哪個數(shù)據(jù)源的困難，接近spark（或者sql）的水平

在上面給出示例中，大家都看到了，現(xiàn)在的mapper可以綁定input嘍！，也就是每個input都有自己獨立的mapper。正因為此，現(xiàn)在的input和mapper改名叫Source和Transformer。

那么，大家又要問了，在mapper中，我已經(jīng)可以輕松根據(jù)不同的數(shù)據(jù)輸入寫出不同的mapper了，那reducer中怎么辦，spark和sql都是很容易實現(xiàn)的哦？比如看人家sql

Select a.id,b.name from A  a,B  b where a.id = b.id

多么輕松愉悅啊!

在原生hadoop mapreduce中，在reducer中找出哪個數(shù)據(jù)對應來自哪個input可是一個令人抓狂的問題呢！

現(xiàn)在這個問題已經(jīng)被輕松解決嘍！看下面這個joiner，對應原生的reducer

public static class Joiner0 extends Joiner<Long, String, String>

…

Reduce方法改名叫join方法，是不是更貼近sql的概念呢？

        public void join(Long key,RowHandler handler,Collector collector) throws Exception{

                List<Object> row  = handler.getSingleFieldRows(0);//對應索引為0的source

         List<Object> row2 = handler.getSingleFieldRows(1);//對應第二個定義的source

注意上面兩句，可以按照數(shù)據(jù)源定義的索引來取出來自不同數(shù)據(jù)源join后的數(shù)據(jù)了，以后有時間可能會改成按照別名來取出，大家看源碼的時候，會發(fā)現(xiàn)別名這個部分的接口都寫好了，要不你來幫助實現(xiàn)了吧。

 

5.      天生的多線程執(zhí)行，即在mapper和reducer端都默認使用多線程來執(zhí)行業(yè)務邏輯。

    看看源碼吧，HST框架是并發(fā)調(diào)用flatMap和join方法的，同時又不能改變系統(tǒng)調(diào)用reduce方法的順序(否則hadoop的辛苦排序可就白瞎了)，這可不是一件容易的事呢!

看到這里，有的同學說了。你這個HST好是好，但你搞的自動轉(zhuǎn)換類型這個機制可能會把性能拉下來的。這個嗎，不得不承認，可能是會有一點影響。但在生產(chǎn)環(huán)境做的比對可以證明，影響太小了，基本忽略不計。

筆者在生產(chǎn)環(huán)境做了做了多次試驗，mapper改成多線程后性能并未有提高，特別是對一些業(yè)務簡單的job，增加Transformer中的并發(fā)級別效率可能還會下降。

很多同學喜歡在mapper中做所謂“mapper端的join”。這種方式，相信在HST中通過提高mapper的并發(fā)級別后會有更好的表現(xiàn)。

Reducer中的性能相對原生提升的空間還是蠻大的。大部分的mapreduce項目，都是mapper簡單而reducer復雜，HST采用并發(fā)執(zhí)行join的方式對提升reducer性能是超好的。

 

6.      對于多次迭代的任務，相連的兩個任務可以建立關聯(lián)，在流程上的下一個job直接引用上一個job的結果，使多次迭代任務的代碼結構變得清晰優(yōu)美

雖然在最后才提到這一點，但這卻是我一開始想要寫HST原因。多次迭代的任務太麻煩了，上一個任務要寫在hdfs做存儲，下一個任務再取出使用，麻煩不麻煩。如果都由程序自動完成，豈不美哉！

在上一個任務里format一下

IteratedJob<Long> iJob = scheduler.createJob("testJob")

...//各種source定義

.format("f1","f2")

在第二個任務中，直接引用

IteratedJob<Long> stage2Job = scheduler.createJob("stage2Job")

.fromPrevious(iJob, Transformer2_0.class);

                  //Transformer2_0.class

 

       public static class Transformer2_0 extends PreviousResultTransformer<Long>

       ...

           public void flatMap(Long inputKey, String[] inputValues,Collector<Long> collector) {

           String f1 = getFiledValue(inputValues, "f1");

           String f2 = getFiledValue(inputValues, "f2");

    看到?jīng)]，就是這么簡單。

在最開始的計劃中，我還設計了使用redis隊列來緩沖前面job的結果，供后面的job作為輸入。這樣本來必須嚴格串行的job可以在一定程度上并發(fā)。另外還設計了子任務的并發(fā)調(diào)度，這都留給以后去實現(xiàn)吧。

 

7. 便捷的自定義參數(shù)傳遞。

有時候，在業(yè)務中需要作一些“開關變量”，在運行時動態(tài)傳入不同的值以實現(xiàn)不同的業(yè)務邏輯。這個問題HST框架其實也為你考慮到了。

Driver中的自定義參數(shù)，source中的自定義參數(shù)都會以內(nèi)置的方式傳到transformer或joiner中去，方便程序員書寫業(yè)務。

   查看transformer或joiner的源碼就會發(fā)現(xiàn)：

getSourceParam(name)和getDriverParam(pIndex)方法，在計算節(jié)點輕松的得到在driver和source中設置的各層次級別的自定義參數(shù),爽吧!

 

8. 其他工具

HST提供的方便還不止以上這些，比如在工具類中還提供了兩行數(shù)據(jù)（map類型）直接join的方法。這些都留給你自己去發(fā)現(xiàn)并實踐吧!
https://github.com/jonenine/HST

posted @ 2017-01-29 10:10 溪石 閱讀(1751) | 評論 (0) | 編輯 收藏

使用jboss netty 創(chuàng)建高性能webservice客戶端及服務端

     摘要:                 使用jboss netty 創(chuàng)建高性能webservice客戶端及服務端      通過本文，讀者將了解以下內(nèi)容 （1）利用jboss netty創(chuàng)建一個高性能的web服務客戶端 （2）不使用任何第三方框架，手工在web容器內(nèi)創(chuàng)建webse...  閱讀全文

posted @ 2014-12-28 14:35 溪石 閱讀(5482) | 評論 (0) | 編輯 收藏

在web開發(fā)中更加富有效率的使用JSON

     摘要: /Files/jonenine/blog.rar本文一共分3個部分，介紹了在web中如何靈活而高效使用json，并同大家分享了實際的代碼，希望大家喜歡。（一）   在jvm中使用jsonjson的使用在當下已經(jīng)普及起來了，作者使用json也已經(jīng)有很長的時間了，早在它普及之前就嘗試著在項目中應用它。今天給大家介紹的代碼都是實際項目中真實的代碼，雖然不盡完美，但卻禁得住實...  閱讀全文

posted @ 2014-12-06 14:08 溪石 閱讀(11003) | 評論 (1) | 編輯 收藏

利用動態(tài)類加載技術調(diào)式ECLIPSE插件

 

                     利用動態(tài)類加載技術調(diào)式ECLIPSE插件

 

   eclipse平臺提供runtime方式調(diào)試插件和RCP項目，但隨著插件項目越寫約復雜，啟動時間也越來越長，特別是集成了諸如 Hibernate和Spring之類的容器級框架的時候。僅僅為了調(diào)試代碼中一些瑣碎的片段而頻繁的重啟項目實在是一件異常煩人的工作。

        即使重啟了項目也許還沒完。為了使項目處于某個特定的狀態(tài)下以方便測試，每次都要重新操作一遍前面業(yè)務流程，這同樣是十分令人厭倦的。

        eclispe使用了OGSI作為微內(nèi)核，引入了一些動態(tài)特性。但是OSGI的動態(tài)特性是在保持平臺運行的情況下動態(tài)更新Bundle，也就是說需要重啟插件才能完成動態(tài)加載的過程。有沒有一些更加細粒度的動態(tài)載入方案呢？

        在Tomcat下開發(fā)過web項目的人都知道，使用調(diào)試模式來部署項目即”熱部署”可以實現(xiàn)動態(tài)載入class文件,讓程序員得以動態(tài)調(diào)試項目。今天向大家提供的這個jar包使得這種效果可以在eclipse runtime上實現(xiàn)。

這是我在自己的插件平臺項目——SCOOP框架中使用的幾個包。

它可以非常好的解決動態(tài)類載入的問題,包括內(nèi)部類的動態(tài)載入都可以很好解決。

其他的幾個包還進行了以下嘗試

1.      使用元數(shù)據(jù)標注的辦法解決SWT UI 線程的種種問題

2.      還提供了eclipse流程框架的簡單實現(xiàn)，以規(guī)范插件開發(fā)。特別是提出了一個面向業(yè)務而不是面向技術的工作流概念，使得編碼粒度變大，并得以提高效率。另外這個簡單的流程框架還將前面的兩種機制很好的結合起來，并且可以和eclipse平臺的一些復雜機制解耦，為復雜流程的開發(fā)測試提供了方便。為將來實現(xiàn)自定義腳本語言(比如某個類似于BPEL的工作流語言)開發(fā)eclise插件項目甚至使用圖形化的開發(fā)奠定了一定的基礎。

        我給出了一個完整的示例——JAXB插件。很多框架同jaxb一樣提供了code generation工具，可以在這個例子的基礎上經(jīng)過簡單修改為這些框架提供插件，比如CXF插件、AXIS2插件等等。

 

動態(tài)類載入編碼原則

         使用動態(tài)類載入機制來進行調(diào)試在編碼上有一定限制。

         首先是要進行動態(tài)載入的實例不要在非動態(tài)域進行引用。只有這樣，當一個流程結束時此實例才會在jvm中得以釋放。當然，非要在其他地方進行引用從而長久的在運行時保持這個實例也是有解決方案的(可以使用代理類技術來實現(xiàn)，具體解決辦法不在本文之內(nèi))。

        其次是進行接口同實現(xiàn)類的分離或父類同子類的分離，以隔離不同的class load scope。接口由父類載入器載入，不同的實現(xiàn)(比如修改后的實現(xiàn))由不同的子類載入器載入。使得最終同一個類型由同一個類載入器載入，這樣才能符合jvm的類載入規(guī)范。在父類載入域的父類型的方法的參數(shù)類型及返回值類型也不能在動態(tài)域中。

最后,那些注冊在擴展點上的類如ActionDelegate和WorkbenchPart等是不能夠動態(tài)載入的，他們必須由eclipse提供的類載入器載入（平臺會自動載入并管理其生命周期）。如果需要讓這些類也動態(tài)載入，就需要在平臺提供的動態(tài)注冊機制基礎上使用代理或像EJB2.0一樣使用侵入式編譯來實現(xiàn)代理機制，這個話題同樣不在本文之內(nèi)。

 

下面就舉例說明使用方法

因為時間有限我無法詳盡的完成此文，請感興趣的朋友自己閱讀示例源碼。

1.      幾種動態(tài)類載入的辦法

(1)    使用手動編碼進行類載入

因為會使二次開發(fā)人員產(chǎn)生迷惑，故未舉例

(2)    在調(diào)試時使用spring文件配置動態(tài)類載入域

Spring配置文件同樣也是動態(tài)的，而且會使調(diào)試開發(fā)工作變得更加清晰，推薦使用

(3)    使用Flow框架來進行動態(tài)類載入

 

2.      使用元數(shù)據(jù)標注解決UI線程訪問的問題

使用三個元數(shù)據(jù)及動態(tài)代理類解決SWT及Eclipse的線程問題。

 

3.      同eclipse內(nèi)部機制解耦以方便開發(fā)和測試

    開發(fā)插件項目很多時候需要實現(xiàn)Eclipse內(nèi)部的一些回調(diào)接口來實現(xiàn)功能，這對程序員的水平是一種考驗，也使得插件開發(fā)工作更加復雜化。比如在實現(xiàn)progressMonitor的時候，就需要實現(xiàn)它的回調(diào)接口，將業(yè)務邏輯放置在其指定的回調(diào)接口——runable接口來實現(xiàn)，這是非常不方便的。我們需要是一種可以提供功能的工具類，像調(diào)用一個普通javaBean一樣來調(diào)用它，而不是將我們的業(yè)務代碼變形撕碎去融入到Eclipse的種種機制中去。

    這樣做的另外一個壞處就是很難進行單元測試，比如脫離eclipse平臺，使用一些mock類來進行簡單有效的單元測試。

    我在這方面也進行了一些嘗試——”反客為主”，將必須遵守eclipse的回調(diào)要求變?yōu)楸仨氉袷貥I(yè)務開發(fā)簡單快捷的要求。同樣是在progressMonitor上面，使用工廠類來創(chuàng)建delegate，然后可以在回調(diào)機制的外部向調(diào)用javaBean一樣來使用平臺給我們提供的這個功能。

這種嘗試是有一定難度的，要使用到不同的設計模式，處理各種線程問題。更重要的一點是eclipse平臺本身有這樣的潛力，它也在等待著我們這樣做。

 

4.      使用Flow規(guī)范插件項目開發(fā)

 

 

我的scoop項目最終擱淺，到最后我發(fā)現(xiàn)實現(xiàn)它已經(jīng)超出了我當時的能力。我原本是想開發(fā)一個統(tǒng)一的插件開發(fā)及部署平臺。這樣很多中小軟件企業(yè)，特別是像我原來工作的那家公司就可以擁有自己的eclipse插件集，以適應自己特殊的要求。我還想提供一套面向插件開發(fā)業(yè)務的接口，而不再面向技術也屏蔽各種技術細節(jié)，使得可以非常方便的擴展、修改甚至移植插件。我只是心里有想念就去實現(xiàn)而已，當最終認識到它有多么困難的時候不得不放棄了。最后雖然失敗了，但并不覺得氣餒。因為知道了要努力的方向，同時也積累了豐富的經(jīng)驗。最后就將這個jar包命名為SCOOP已示紀念。

 演示文檔及框架JAR包如下

/Files/jonenine/Eclipse_Dynamic_Classload.rar

posted @ 2009-05-04 14:06 溪石 閱讀(1672) | 評論 (3) | 編輯 收藏

在web開發(fā)中的樹狀視圖技術

     摘要: 在web開發(fā)中的樹狀視圖技術 樹型數(shù)據(jù)模型在現(xiàn)實生活中應用相當廣泛，從超市的商品分類到政府的組織結構無不都是”樹形”的。在實際的項目開發(fā)中也經(jīng)常需要顯示這種結構。比如，在樹狀視圖上給一個單位的職工賦予系統(tǒng)操作權限。          在web開發(fā)中實現(xiàn)樹狀顯示在技術上主要有以下兩種方案 1...  閱讀全文

posted @ 2009-05-04 13:14 溪石 閱讀(5319) | 評論 (7) | 編輯 收藏