摘錄自2020年11月5日中央社報導

日本香川縣三豊市的一處養雞場從11月1日到4日突然有約3800隻雞死亡，經檢驗後驗出H5型禽流感病毒，該養雞場將撲殺約33萬隻雞。

香川縣政府今天上午召開對策本部會議，決定養雞場飼養的約33萬隻雞將進行撲殺。

日本的養雞場上次發生禽流感疫情，是2018年1月香川縣讚岐市的養雞場驗出H5型禽流感病毒。

日本首相菅義偉今天上午聽取報告後，指示相關單位要提醒家禽業者對此高度警戒，並輔導與支援業者預防、收集現場情報；農林水產省與相關部會緊密合作，迅速進行防疫措施，以及對國民盡速提供正確的防疫資訊。

國際新聞
日本
禽流感

本站聲明:網站內容來源環境資訊中心https://e-info.org.tw/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※網頁設計公司推薦不同的風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※南投搬家公司費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

※超省錢租車方案

[源碼解析] GroupReduce，GroupCombine和Flink SQL group by

目錄

• [源碼解析] GroupReduce，GroupCombine和Flink SQL group by
  • 0x00 摘要
  • 0x01 緣由
  • 0x02 概念
    • 2.1 GroupReduce
    • 2.2 GroupCombine
    • 2.3 例子
  • 0x03 代碼
  • 0x04 Flink SQL內部翻譯
  • 0x05 JobGraph
  • 0x06 Runtime
    • 6.1 ChainedFlatMapDriver
    • 6.2 GroupReduceCombineDriver
    • 6.3 GroupReduceDriver & ChainedFlatMapDriver
  • 0x07 總結
  • 0x08 參考

0x00 摘要

本文從源碼和實例入手，為大家解析 Flink 中 GroupReduce 和 GroupCombine 的用途。也涉及到了 Flink SQL group by 的內部實現。

0x01 緣由

在前文[源碼解析] Flink的Groupby和reduce究竟做了什麼中，我們剖析了Group和reduce都做了些什麼，也對combine有了一些了解。但是總感覺意猶未盡，因為在Flink還提出了若干新算子，比如GroupReduce和GroupCombine。這幾個算子不搞定，總覺得如鯁在喉，但沒有找到一個良好的例子來進行剖析說明。

本文是筆者在探究Flink SQL UDF問題的一個副產品。起初是為了調試一段sql代碼，結果發現Flink本身給出了一個GroupReduce和GroupCombine使用的完美例子。於是就拿出來和大家共享，一起分析看看究竟如何使用這兩個算子。

請注意：這個例子是Flink SQL，所以本文中將涉及Flink SQL goup by內部實現的知識。

0x02 概念

Flink官方對於這兩個算子的使用說明如下：

2.1 GroupReduce

GroupReduce算子應用在一個已經分組了的DataSet上，其會對每個分組都調用到用戶定義的group-reduce函數。它與Reduce的區別在於用戶定義的函數會立即獲得整個組。

Flink將在組的所有元素上使用Iterable調用用戶自定義函數，並且可以返回任意數量的結果元素。

2.2 GroupCombine

GroupCombine轉換是可組合GroupReduceFunction中組合步驟的通用形式。它在某種意義上被概括為允許將輸入類型 I 組合到任意輸出類型O。與此相對的是，GroupReduce中的組合步驟僅允許從輸入類型 I 到輸出類型 I 的組合。這是因為GroupReduceFunction的 “reduce步驟” 期望自己的輸入類型為 I。

在一些應用中，我們期望在執行附加變換（例如，減小數據大小）之前將DataSet組合成中間格式。這可以通過CombineGroup轉換能以非常低的成本實現。

注意：分組數據集上的GroupCombine在內存中使用貪婪策略執行，該策略可能不會一次處理所有數據，而是以多個步驟處理。它也可以在各個分區上執行，而無需像GroupReduce轉換那樣進行數據交換。這可能會導致輸出的是部分結果，所以GroupCombine是不能替代GroupReduce操作的，儘管它們的操作內容可能看起來都一樣。

2.3 例子

是不是有點暈？還是直接讓代碼來說話吧。以下官方示例演示了如何將CombineGroup和GroupReduce轉換用於WordCount實現。即通過combine操作先對單詞數目進行初步排序，然後通過reduceGroup對combine產生的結果進行最終排序。因為combine進行了初步排序，所以在算子之間傳輸的數據量就少多了。

DataSet<String> input = [..] // The words received as input

// 這裏通過combine操作先對單詞數目進行初步排序，其優勢在於用戶定義的combine函數只調用一次，因為runtime已經把輸入數據一次性都提供給了自定義函數。  
DataSet<Tuple2<String, Integer>> combinedWords = input
  .groupBy(0) // group identical words
  .combineGroup(new GroupCombineFunction<String, Tuple2<String, Integer>() {

    public void combine(Iterable<String> words, Collector<Tuple2<String, Integer>>) { // combine
        String key = null;
        int count = 0;

        for (String word : words) {
            key = word;
            count++;
        }
        // emit tuple with word and count
        out.collect(new Tuple2(key, count));
    }
});

// 這裏對combine的結果進行第二次排序，其優勢在於用戶定義的reduce函數只調用一次，因為runtime已經把輸入數據一次性都提供給了自定義函數。  
DataSet<Tuple2<String, Integer>> output = combinedWords
  .groupBy(0)                              // group by words again
  .reduceGroup(new GroupReduceFunction() { // group reduce with full data exchange

    public void reduce(Iterable<Tuple2<String, Integer>>, Collector<Tuple2<String, Integer>>) {
        String key = null;
        int count = 0;

        for (Tuple2<String, Integer> word : words) {
            key = word;
            count++;
        }
        // emit tuple with word and count
        out.collect(new Tuple2(key, count));
    }
});

看到這裏，有的兄弟已經明白了，這和mapPartition很類似啊，都是runtime做了大量工作。為了讓大家這兩個算子的使用情形有深刻的認識，我們再通過一個sql的例子，向大家展示Flink內部是怎麼應用這兩個算子的，也能看出來他們的強大之處。

0x03 代碼

下面代碼主要參考自 flink 使用問題匯總。我們可以看到這裏通過groupby進行了聚合操作。其中collect方法，類似於mysql的group_concat。

public class UdfExample {
    public static class MapToString extends ScalarFunction {

        public String eval(Map<String, Integer> map) {
            if(map==null || map.size()==0) {
                return "";
            }
            StringBuffer sb=new StringBuffer();
            for(Map.Entry<String, Integer> entity : map.entrySet()) {
                sb.append(entity.getKey()+",");
            }
            String result=sb.toString();
            return result.substring(0, result.length()-1);
        }
    }

    public static void main(String[] args) throws Exception {
        MemSourceBatchOp src = new MemSourceBatchOp(new Object[][]{
                new Object[]{"1", "a", 1L},
                new Object[]{"2", "b33", 2L},
                new Object[]{"2", "CCC", 2L},
                new Object[]{"2", "xyz", 2L},
                new Object[]{"1", "u", 1L}
        }, new String[]{"f0", "f1", "f2"});

        BatchTableEnvironment environment = MLEnvironmentFactory.getDefault().getBatchTableEnvironment();
        Table table = environment.fromDataSet(src.getDataSet());
        environment.registerTable("myTable", table);
        BatchOperator.registerFunction("MapToString",  new MapToString());
        BatchOperator.sqlQuery("select f0, mapToString(collect(f1)) as type from myTable group by f0").print();
    }
}

程序輸出是

f0|type
--|----
1|a,u
2|CCC,b33,xyz

0x04 Flink SQL內部翻譯

這個SQL語句的重點是group by。這個是程序猿經常使用的操作。但是大家有沒有想過這個group by在真實運行起來時候是怎麼操作的呢？針對大數據環境有沒有做了什麼優化呢？其實，Flink正是使用了GroupReduce和GroupCombine來實現並且優化了group by的功能。優化之處在於：

• GroupReduce和GroupCombine的函數調用次數要遠低於正常的reduce算子，如果reduce操作中涉及到頻繁創建額外的對象或者外部資源操作，則會相當省時間。
• 因為combine進行了初步排序，所以在算子之間傳輸的數據量就少多了。

SQL生成Flink的過程十分錯綜複雜，所以我們只能找最關鍵處。其是在 DataSetAggregate.translateToPlan 完成的。我們可以看到，對於SQL語句 “select f0, mapToString(collect(f1)) as type from myTable group by f0”，Flink系統把它翻譯成如下階段，即

• pre-aggregation ：排序 + combine；
• final aggregation ：排序 + reduce；

從之前的文章我們可以知道，groupBy這個其實不是一個算子，它只是排序過程中的一個輔助步驟而已，所以我們重點還是要看combineGroup和reduceGroup。這恰恰是我們想要的完美例子。

input ----> (groupBy + combineGroup) ----> (groupBy + reduceGroup) ----> output

SQL生成的Scala代碼如下，其中 combineGroup在後續中將生成GroupCombineOperator，reduceGroup將生成GroupReduceOperator。

  override def translateToPlan(
      tableEnv: BatchTableEnvImpl,
      queryConfig: BatchQueryConfig): DataSet[Row] = {

    if (grouping.length > 0) {
      // grouped aggregation
      ...... 
      if (preAgg.isDefined) { // 我們的例子是在這裏
        inputDS          
          // pre-aggregation
          .groupBy(grouping: _*)
          .combineGroup(preAgg.get) // 將生成GroupCombineOperator算子
          .returns(preAggType.get)
          .name(aggOpName)
          // final aggregation
          .groupBy(grouping.indices: _*) //將生成GroupReduceOperator算子。
          .reduceGroup(finalAgg.right.get)
          .returns(rowTypeInfo)
          .name(aggOpName)
      } else {
        ......
      }
    }
    else {
      ......
    }
  }
}

// 程序變量打印如下
this = {DataSetAggregate@5207} "Aggregate(groupBy: (f0), select: (f0, COLLECT(f1) AS $f1))"
 cluster = {RelOptCluster@5220} 

0x05 JobGraph

LocalExecutor.execute中會生成JobGraph。JobGraph是提交給 JobManager 的數據結構，是唯一被Flink的數據流引擎所識別的表述作業的數據結構，也正是這一共同的抽象體現了流處理和批處理在運行時的統一。

在生成JobGraph時候，系統得到如下JobVertex。

jobGraph = {JobGraph@5652} "JobGraph(jobId: 6aae8b5e5ad32f588136bef26f8b65f6)"
 taskVertices = {LinkedHashMap@5655}  size = 4

{JobVertexID@5677} "c625209bb7fb9a098807551840aeaa99" -> {InputOutputFormatVertex@5678} "CHAIN DataSource (at initializeDataSource(MemSourceBatchOp.java:98) (org.apache.flink.api.java.io.CollectionInputFormat)) -> FlatMap (select: (f0, f1)) (org.apache.flink.runtime.operators.DataSourceTask)"

{JobVertexID@5679} "b56ace4acd7a2f69ea110a9f262ff80a" -> {JobVertex@5680} "CHAIN GroupReduce (groupBy: (f0), select: (f0, COLLECT(f1) AS $f1)) -> FlatMap (select: (f0, mapToString($f1) AS type)) -> Map (Map at linkFrom(MapBatchOp.java:35)) (org.apache.flink.runtime.operators.BatchTask)"
 
{JobVertexID@5681} "3f5e2a0f700421d80ce85e02a6d9db73" -> {InputOutputFormatVertex@5682} "DataSink (collect()) (org.apache.flink.runtime.operators.DataSinkTask)"
 
{JobVertexID@5683} "ad29dc5b2e0a39ad2cd1d164b6f859f7" -> {JobVertex@5684} "GroupCombine (groupBy: (f0), select: (f0, COLLECT(f1) AS $f1)) (org.apache.flink.runtime.operators.BatchTask)"

我們可以看到，在JobGraph中就生成了對應的兩個算子。其中這裏的FlatMap就是用戶的UDF函數MapToString的映射生成。

GroupCombine (groupBy: (f0), select: (f0, COLLECT(f1) AS $f1))  
  
CHAIN GroupReduce (groupBy: (f0), select: (f0, COLLECT(f1) AS $f1)) -> FlatMap (select: (f0, mapToString($f1) AS type)) -> Map 

0x06 Runtime

最後，讓我們看看runtime會如何處理這兩個算子。

6.1 ChainedFlatMapDriver

首先，Flink會在ChainedFlatMapDriver.collect中對record進行處理，這是從Table中提取數據所必須經歷的，與後續的group by關係不大。

@Override
public void collect(IT record) {
   try {
      this.numRecordsIn.inc();
      this.mapper.flatMap(record, this.outputCollector);
   } catch (Exception ex) {
      throw new ExceptionInChainedStubException(this.taskName, ex);
   }
}

// 這裡能夠看出來，我們獲取了第一列記錄
record = {Row@9317} "1,a,1"
 fields = {Object[3]@9330} 
this.taskName = "FlatMap (select: (f0, f1))"

// 程序堆棧打印如下
collect:80, ChainedFlatMapDriver (org.apache.flink.runtime.operators.chaining)
collect:35, CountingCollector (org.apache.flink.runtime.operators.util.metrics)
invoke:196, DataSourceTask (org.apache.flink.runtime.operators)
doRun:707, Task (org.apache.flink.runtime.taskmanager)
run:532, Task (org.apache.flink.runtime.taskmanager)
run:748, Thread (java.lang)

6.2 GroupReduceCombineDriver

其次，GroupReduceCombineDriver.run()中會進行combine操作。

1. 會通過this.sorter.write(value)把數據寫到排序緩衝區。
2. 會通過sortAndCombineAndRetryWrite(value)進行實際的排序，合併。

因為是系統實現，所以Combine的用戶自定義函數就是由Table API提供的，比如org.apache.flink.table.functions.aggfunctions.CollectAccumulator.accumulate 。

@Override
public void run() throws Exception {
   final MutableObjectIterator<IN> in = this.taskContext.getInput(0);
   final TypeSerializer<IN> serializer = this.serializer;

   if (objectReuseEnabled) {
    .....
   }
   else {
      IN value;
      while (running && (value = in.next()) != null) {
         // try writing to the sorter first
         if (this.sorter.write(value)) {
            continue;
         }

         // do the actual sorting, combining, and data writing
         sortAndCombineAndRetryWrite(value);
      }
   }

   // sort, combine, and send the final batch
   if (running) {
      sortAndCombine();
   }
}

// 程序變量如下
value = {Row@9494} "1,a"
 fields = {Object[2]@9503} 

sortAndCombine是具體排序/合併的過程。

1. 排序是通過 org.apache.flink.runtime.operators.sort.QuickSort 完成的。
2. 合併是通過 org.apache.flink.table.functions.aggfunctions.CollectAccumulator.accumulate 完成的。
3. 給下游是由 org.apache.flink.table.runtime.aggregate.DataSetPreAggFunction.combine 調用 out.collect(output) 完成的。

private void sortAndCombine() throws Exception {
   final InMemorySorter<IN> sorter = this.sorter;
   // 這裏進行實際的排序
   this.sortAlgo.sort(sorter);
   final GroupCombineFunction<IN, OUT> combiner = this.combiner;
   final Collector<OUT> output = this.output;

   // iterate over key groups
   if (objectReuseEnabled) {
			......		
   } else {
      final NonReusingKeyGroupedIterator<IN> keyIter = 
            new NonReusingKeyGroupedIterator<IN>(sorter.getIterator(), this.groupingComparator);
      // 這裡是歸併操作
      while (this.running && keyIter.nextKey()) {
         // combiner.combiner 是用戶定義操作，runtime把某key對應的數據一次性傳給它
         combiner.combine(keyIter.getValues(), output);
      }
   }
}

具體調用棧如下：

accumulate:57, CollectAggFunction (org.apache.flink.table.functions.aggfunctions)
accumulate:-1, DataSetAggregatePrepareMapHelper$5
combine:71, DataSetPreAggFunction (org.apache.flink.table.runtime.aggregate)
sortAndCombine:213, GroupReduceCombineDriver (org.apache.flink.runtime.operators)
run:188, GroupReduceCombineDriver (org.apache.flink.runtime.operators)
run:504, BatchTask (org.apache.flink.runtime.operators)
invoke:369, BatchTask (org.apache.flink.runtime.operators)
doRun:707, Task (org.apache.flink.runtime.taskmanager)
run:532, Task (org.apache.flink.runtime.taskmanager)
run:748, Thread (java.lang)

6.3 GroupReduceDriver & ChainedFlatMapDriver

這兩個放在一起，是因為他們組成了Operator Chain。

GroupReduceDriver.run中完成了reduce。具體reduce 操作是在 org.apache.flink.table.runtime.aggregate.DataSetFinalAggFunction.reduce 完成的，然後在其中直接發送給下游 out.collect(output)。

@Override
public void run() throws Exception {
   // cache references on the stack
   final GroupReduceFunction<IT, OT> stub = this.taskContext.getStub();
 
   if (objectReuseEnabled) {
       ......	
   }
   else {
      final NonReusingKeyGroupedIterator<IT> iter = new NonReusingKeyGroupedIterator<IT>(this.input, this.comparator);
      // run stub implementation
      while (this.running && iter.nextKey()) {
         // stub.reduce 是用戶定義操作，runtime把某key對應的數據一次性傳給它
         stub.reduce(iter.getValues(), output);
      }
   }
}

從前文我們可以，這裏已經配置成了Operator Chain，所以out.collect(output)會調用到CountingCollector。CountingCollector的成員變量collector已經配置成了ChainedFlatMapDriver。

public void collect(OUT record) {
   this.numRecordsOut.inc();
   this.collector.collect(record);
}

this.collector = {ChainedFlatMapDriver@9643} 
 mapper = {FlatMapRunner@9610} 
 config = {TaskConfig@9655} 
 taskName = "FlatMap (select: (f0, mapToString($f1) AS type))"

於是程序就調用到了 ChainedFlatMapDriver.collect。

public void collect(IT record) {
   try {
      this.numRecordsIn.inc();
      this.mapper.flatMap(record, this.outputCollector);
   } catch (Exception ex) {
      throw new ExceptionInChainedStubException(this.taskName, ex);
   }
}

最終調用棧如如下：

eval:21, UdfExample$MapToString (com.alibaba.alink)
flatMap:-1, DataSetCalcRule$14
flatMap:52, FlatMapRunner (org.apache.flink.table.runtime)
flatMap:31, FlatMapRunner (org.apache.flink.table.runtime)
collect:80, ChainedFlatMapDriver (org.apache.flink.runtime.operators.chaining)
collect:35, CountingCollector (org.apache.flink.runtime.operators.util.metrics)
reduce:80, DataSetFinalAggFunction (org.apache.flink.table.runtime.aggregate)
run:131, GroupReduceDriver (org.apache.flink.runtime.operators)
run:504, BatchTask (org.apache.flink.runtime.operators)
invoke:369, BatchTask (org.apache.flink.runtime.operators)
doRun:707, Task (org.apache.flink.runtime.taskmanager)
run:532, Task (org.apache.flink.runtime.taskmanager)
run:748, Thread (java.lang)

0x07 總結

由此我們可以看到：

• GroupReduce，GroupCombine和mapPartition十分類似，都是從系統層面對算子進行優化，把循環操作放到用戶自定義函數來處理。
• 對於group by這個SQL語句，Flink將其翻譯成 GroupReduce + GroupCombine，採用兩階段優化的方式來完成了對大數據下的處理。

0x08 參考

flink 使用問題匯總

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※網頁設計公司推薦不同的風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※南投搬家公司費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

※超省錢租車方案

摘錄自2018年9月9日蘋果日報美國報導

位於美國加州和夏威夷之間有個全球最大的垃圾帶，聚集人類多年來不斷丟棄到大海中的垃圾。為了解決此生態浩劫，由荷蘭24歲青年研發的巨型垃圾蒐集器昨（8）日正式上場，將沿路「撿回」垃圾，還給大海一個乾淨空間。

這個長約600公尺的巨大垃圾搜集器周六從舊金山港口出發，朝著相當於40個台灣面積的太平洋垃圾帶（Great Pacific Garbage Patch）前進。搜集器呈U型漂浮設計，在海上攔截垃圾。裝置以低於水流的速度行進，讓潮汐和海浪產生推力，將海洋垃圾往裝置中心帶去。裝置底下設有網狀裝置，能攔下較小、沉降在水面下的海洋廢棄物。而過濾網產生的推力，則能使魚類和其他生物，得以輕鬆繞過網子底下、通過裝置而不被攔截。被攔下的海洋垃圾，再由團隊派出船隻定期載運回陸地，進行資源回收再利用。

此計畫由荷蘭青年斯萊特（Boyan Slat）所創立的「海洋清理計劃」（Ocean Cleanup Project）推動。該組織已募得3500萬美元（約10億7870萬元台幣）資金，預計在2020年前在太平洋海域上放置60個搜集器檔板。斯萊特說：「我們的目標是在5年內清理5成太平洋垃圾帶中的垃圾。」斯萊特還說，這些漂浮檔板可抵抗惡劣氣候，使用期限長達20年，可望蒐集到9成海面漂浮垃圾。

本站聲明:網站內容來源環境資訊中心https://e-info.org.tw/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※網頁設計公司推薦不同的風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※南投搬家公司費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

※超省錢租車方案

目錄

• 1. 什麼是內存模型？
• 2. JSR 133是關於什麼的?
• 3. 再談指令重排序
• 4.同步都做了什麼？
• 5. final字段在舊的內存模型中為什麼可以改變？
• 6.“初始化安全”與final字段？
• 7. 增強volatile語義
• 8. 修復“double-checked locking”的問題
• 9. 為什麼要關心這些問題？
• 延伸閱讀

1. 什麼是內存模型？

在多處理器系統中，為了提高訪問數據的速度，通常會增加一層或多層高速緩存(越靠近處理器的緩存速度越快)。
但是緩存同時也帶來了許多新的挑戰。比如，當兩個處理器同時讀取同一個內存位置時，看到的結果可能會不一樣？
在處理器維度上，內存模型定義了一些規則來保證當前處理器可以立即看到其他處理器的寫入，以及當前處理器的寫入對其他處理器立即可見。這些規則被稱為緩存一致性協議。
有些多處理器架構實現了強一致性，所有的處理器在同一時刻看到的同一內存位置的值是一樣的。
而其他處理器實現的則是較弱的一致性，需要使用被稱為內存屏障的特殊機器指令使來實現最終一致性（通過刷新緩存或使緩存失效）。
這些內存屏障通常在釋放鎖和獲取鎖時被執行；對於高級語言(如Java)的程序員來說，它們是不可見的。

在強一致性的處理器上,由於減少了對內存屏障的依賴，編寫併發程序會更容易一些。
但是，相反的，近年來處理器設計的趨勢是使用較弱的內存模型，因為放寬對緩存一致性的要求可以使得多處理器系統有更好的伸縮性和更大的內存。

此外，編譯器、緩存或運行時還被允許通過指令重排序改變內存的操作順序(相對於程序所表現的順序)。
例如，編譯器可能會往後移動一個寫入操作，只要移動操作不改變程序的原本語義(as-if-serial語義)，就可以自由進行更改。
再比如，緩存可能會推遲把數據刷回到主內存中，直到它認為時機合適了。
這種靈活的設計，目的都是為了獲得得最佳的性能，
但是在多線程環境下，指令重排會使得跨線程可見性的問題變的更複雜。

為了方便理解，我們來看個代碼示例：

Class Reordering {
  int x = 0, y = 0;
  //thread A
   public void writer() {
        x = 1;
        y = 2;
    }

    //thread B
    public void reader() {
        int r1 = y;
        int r2 = x;
     }
}

假設這段代碼被兩個線程併發執行，線程A執行writer()，線程B執行reader()。
如果線程B在reader()中看到了y=2，那麼直覺上我們會認為它看到的x肯定是1，因為在writer()中x=1在y=2之前 。
然而，發生重排序時y=2會早於x=1執行，此時，實際的執行順序會是這樣的：

y=2；
int r1=y;
int r2=x;
x=1;

結果就是，r1的值是2，r2的值是0。
從線程A的角度看，x=1與y=2哪個先執行結果是一樣的（或者說沒有違反as-if-serial語義），但是在多線程環境下，這種重排序會產生混亂的結果。

我們可以看到，高速緩存和指令重排序提高了效率的同時也引出了新的問題，這顯然使得編寫併發程序變得更加困難。
Java內存模型就是為了解決這類問題，它對多線程之間如何通過內存進行交互做了明確的說明。
更具體點，Java內存模型描述了程序中的變量與實際計算機的存儲設備（包括內存、緩存、寄存器）之間交互的底層細節。
例如，Java提供了volatile、final和 synchronized等工具，用於幫助程序員向編譯器表明對併發程序的要求。
更重要的是，Java內存模型保證這些同步工具可以正確的運行在任何處理器架構上，使Java併發應用做到“Write Once, Run Anywhere”。

相比之下，大多數其他語言(例如C/C++)都沒有提供显示的內存模型。
C程序繼承了處理器的內存模型，這意味着，C語言的併發程序在一個處理器架構中可以正確運行，在另外一個架構中則不一定。

2. JSR 133是關於什麼的?

Java提供的跨平台內存模型是一個雄心勃勃的計劃，在當時是具有開創性的。
但不幸的是，定義一個即直觀又一致的內存模型比預期的要困難得多。
自1997年以來，在《Java語言規範》的第17章關於Java內存模型的定義中發現了一些嚴重的缺陷。
這些缺陷使一些同步工具產生混亂的結果，例如final字段可以被更改。
JSR 133為Java語言定義了一個新的內存模型，修復了舊版內存模型的缺陷（修改了final和volatile的語義)
JSR的主要目標包括不限於這些：

1. 正確同步的語義應該更直觀更簡單。
2. 應該定義不完整或不正確同步的語義，以最小化潛在的安全隱患
3. 程序員應該有足夠的自信推斷出多線程程序如何與內存交互的。
4. 提供一個新的初始化安全性保證(initialization safety)。
   如果一個對象被正確初始化了(初始化期間，對象的引用沒有逃逸，比如構造函數里把this賦值給變量)，那麼所有可以看到該對象引用的線程，都可以看到在構造函數中被賦值的final變量。這不需要使用synchronized或volatile。

3. 再談指令重排序

在許多情況下，出於優化執行效率的目的，數據(實例變量、靜態字段、數組元素等)可以在寄存器、緩存和內存之間以不同於程序中聲明的順序被移動。
例如，線程先寫入字段a，再寫入字段b，並且b的值不依賴a，那麼編譯器就可以自由的對這些操作重新排序，在寫入a之前把b的寫入刷回到內存。
除了編譯器，重排序還可能發生在JIT、緩存、處理器上。
無論發生在哪裡，重排序都必須遵循as-if-serial語義，這意味着在單線程程序中，程序不會覺察到重排序的存在，或者說給單線程程序一種沒有發生過重排序的錯覺。
但是，重排序在沒有同步的多線程程序中會產生影響。在這種程序中，一個線程能夠觀察到其他線程的運行情況，並且可能檢測到變量訪問順序與代碼中指定的順序不一致。
大多數情況下，一個線程不會在乎另一個線程在做什麼，但是，如果有，就是同步的用武之地。

4.同步都做了什麼？

同步有很多面，最為程序員熟知的是它的互斥性，同一時刻只能有一個線程持有monitor。
但是，同步不僅僅是互斥性。同步還能保證一個線程在同步塊中的寫內存操作對其他持有相同monitor的線程立即可見。
當線程退出同步塊時(釋放monitor)，會把緩存中的數據刷回到主內存，使主內存中保持最新的數據。
當線程進入同步塊時(獲取monitor)，會使本地處理器緩存失效，使得變量必須從主內存中重新加載。
我們可以看到，之前的所有寫操作對後來的線程都是可見的。

5. final字段在舊的內存模型中為什麼可以改變？

證明final字段可以改變的最佳示例是String類的實現(JDK 1.4版本)。
String對象包含三個字段：一個字符串數組的引用value、一個記錄數組中開始位置的offset、字符串長度length。
通過這種方式，可以實現多個String/StringBuffer對象共享一個相同的字符串數組，從而避免為每個對象分配額外的空間。
例如，String.substring()通過與原String對象共享一個數組來產生一個新的對象，唯一的不同是length和offset字段。

String s1 = "/usr/tmp";
String s2 = s1.substring(4); 

s2和s1共享一個字符串數組”/usr/tmp”，不同的是s2的offset=4，length=4，s1的offset=0，length=8。
在String的構造函數運行之前，根類Object的構造函數會先初始化所有字段為默認值，包括final的length和offset字段。
當String的構造函數運行時，再把length和offset賦值為期望的值。
但是這一過程，在舊的內存模型中，如果沒有使用同步，另一個線程可能會看到offset的默認值0，然後在看到正確的值4.
結果導致一個迷幻的現象，開始看到字符串s2的內容是’/usr’，然後再看到’/tmp’。
這不符合我們對final語義的認識，但是在舊內存模型中確實存在這樣的問題。
（JDK7開始，改變了substring的實現方式，每次都會創建一個新的對象）

6.“初始化安全”與final字段？

新的內存模型提供一個新初始化安全( initialization safety)保障。
意味着，只要一個對象被正確的構造，那麼所有的線程都會看到這些在構造函數中被賦值的final字段。
“正確”的構造是指在構造函數執行期間，對象的引用沒有發生逃逸。或者說，在構造函數中沒有把該對象的引用賦值給任何變量。

class FinalFieldExample {
  final int x;
  int y;
  static FinalFieldExample f;
  public FinalFieldExample() {
    x = 3;
    y = 4;
  }

  static void writer() {
    f = new FinalFieldExample();
  }

  static void reader() {
    if (f != null) {
      int i = f.x;
      int j = f.y;
    }
  }
}

示例中，初始化安全保證執行reader()方法的線程看到的f.x=3，因為它是final字段，但是不保證能看到y=4，因為它不是final的。
但是如果構造函數像這樣:

public FinalFieldExample() { // bad!
  x = 3;
  y = 4;
  global.obj = this;  //  allowing this to escape
}

初始化安全不能保證讀取global.obj的線程看到的x的值是3，因為對象引用this發生了逃逸。

不僅如此，任何通過final字段(構造函數中被賦值的)可以觸達的變量都可以保證對其他線程可見。
這意味着如果一個final字段包含一個引用，例如ArrayList，除了該字段的引用對其他線程可見，ArrayList中的元素對其他線程也是可見的。

初始化安全增強了final的語義，使其更符合我們對final的直觀感受，任何情況下都不會改變。

7. 增強volatile語義

volatile變量是用於線程之間傳遞狀態的特殊變量，這要求任何線程看到的都是volatile變量的最新值。
為實現可見性，禁止在寄存器中分配它們，還必須確保修改volatile后，要把最新值從緩存刷到內存中。
類似的，在讀取volatile變量之前，必須使高速緩存失效，這樣其他線程會直接讀取主內存中的數據。
在舊的內存模型中，多個volatile變量之間不能互相重排序，但是它們被允許可以與非volatile變量一起重排序，這消弱了volatile作為線程間交流信號的作用。
我們來看個示例：

Map configs;
volatile boolean initialized = false;
. . .
 
// In thread A
configs  =  readConfigFile(fileName);
processConfigOptions( configs);
initialized = true;
. . .
 
// In thread B
while (initialized) {
    // use configs
}

示例中，線程A負責配置數據初始化工作，初始化完成后線程B開始執行。
實際上，volatile變量initialized扮演者守衛者的角色，它表示前置工作已經完成，依賴這些數據的其他線程可以執行了。
但是，當volatile變量與非volatile變量被編譯器放到一起重新排序時，“守衛者”就形同虛設了。
重排序發生時，可能會使readConfigFile()中某個動作在initialized = true之後執行，
那麼，線程B在看到initialized的值為true后，在使用configs對象時，會讀取到沒有被正確初始化的數據。
這是volatile很典型的應用場景，但是在舊的內存模型中卻不能正確的工作。

JSR 133專家組決定在新的內存模型中，不再允許volatile變量與其他任務內存操作一起重排序。
這意味着，volatile變量之前的內存操作不會在其後執行，volatile變量之後的內存操作不會在其前執行。
volatile變量相當於一個屏障，重排序不能越過對volatile的內存操作。（實際上，jvm確實使用了內存屏障指令）
增強volatile語義的副作用也很明顯，禁止重排序會有一定的性能損失。

8. 修復“double-checked locking”的問題

double-checked locking是單例模式的其中一種實現，它支持懶加載且是線程安全的。
大概長這個樣子：

private static Something instance = null;

public Something getInstance() {
  if (instance == null) {
    synchronized (this) {
      if (instance == null)
        instance = new Something();//
    }
  }
  return instance;
}

它通過兩次檢查巧妙的避開了在公共代碼路徑上使用同步，從而避免了同步所帶來的性能開銷。
它唯一的問題就是——不起作用。為什麼呢？
instance的賦值操作會與SomeThing()構造函數中的變量初始化一起被編譯器或緩存重排序，這可能會導致把未完全初始化的對象引用賦值給instance。
現在很多人知道把instance聲明為volatile可以修復這個問題，但是在舊的內存模型(JDK 1.5之前)中並不可行，原因前面有提到，volatile可以與非volatile字段一起重排序。

儘管，新的內存模型修復了double-checked locking的問題，但仍不鼓勵這種實現方式，因為volatile並不是免費的。
相比之下，Initialization On Demand Holder Class更值得被推薦，
它不僅實現了懶加載和線程安全，還提供了更好的性能和更清晰的代碼邏輯。大概長這個樣子：

public class Something {
    private Something() {}
    //static innner class
    private static class LazyHolder {
        static final Something INSTANCE = new Something(); //static  field
    }

    public static Something getInstance() {
        return LazyHolder.INSTANCE;
    }
}

這種實現完全沒有使用同步工具，而是利用了Java語言規範的兩個基本原則，
其一，JVM保證靜態變量的初始化對所有使用該類的線程立即可見；
其二，內部類首次被使用時才會觸發類的初始化，這實現了懶加載。

9. 為什麼要關心這些問題？

併發問題一般不會在測試環境出現，生成環境的併發問題又不容易復現，這兩個特點使得併發問題通常比較棘手。
所以你最好提前花點時間學習併發知識，以確保寫出正確的併發程序。我知道這很困難，但是應該比排查生產環境的併發問題容易的多。

延伸閱讀

1.JSR 133 (Java Memory Model) FAQ,2004
2.volatile關鍵字
3.Double-checked問題
4.內存屏障和volatile語義
5.修復Java內存模型
6.String substring 在jdk7中會創建新的數組
7.Memory Ordering
8.有MESI協議為什麼還需要volatile?
9.Initialization On Demand Holder Class
10.The JSR-133 Cookbook for Compiler Writers

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※網頁設計公司推薦不同的風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※南投搬家公司費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

※超省錢租車方案

環境資訊中心記者 陳文姿報導

本站聲明:網站內容來源環境資訊中心https://e-info.org.tw/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※網頁設計公司推薦不同的風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※南投搬家公司費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

※超省錢租車方案