台北網頁設計 | 買了電動機車後才知道!電動車的優點與缺點

電動車需求看好，2016年美國銷量可達20萬輛

2020-03-162020-03-16 admin

電動車市場越趨火熱，在許多國家甚至能經常撞見電動車在街頭奔走。根據美國市場研究機構Navigant Research 近日發布的研究指出，預計從2016 年起，美國電動車市場每年將成長62%，而今年銷量可望達到近20 萬台。

Navigant Research 指出，定價低於4 萬美元的長程純電動車（BEVs），舉例來說，定價3.7 萬美元、預計於今年秋季上市的雪佛蘭（Chevrolet）Bolt 車款，將是造成電動車市場成長的關鍵。此外，特斯拉（Tesla）的純電動車Model X，以及雪佛蘭Volt 、豐田（Toyota）的Prius Prime 與即將推出的三菱（Mitsubishi）Outlander 等插電式混合動力汽車（PHEV），都將引爆美國市場對電動車需求的熱潮。

隨著電動車續航力提高、定價逐漸下滑，加上各家車廠紛紛加入這塊兵家之地，使消費者購買電動車的選擇相對變多，對於消費者來說，現階段要購買一台好用的電動車變得容易許多。2015 年時，就已有9 種不同的電動車種類，從2 人座到標準運動休旅車（SUV）大小的種類都有，而當年度所售出的插電式電動車中，約6 成為如日產（Nissan）Leaf 與特斯拉Model S 等中型或大型車款。

Navigant Research 與美國能源部認為，插電式電動車市場真正開啟是在2011 年，當時僅有4 種車款可供選擇，而現在，從純電動小車Fiat 500e 到富豪（Volvo）XC90 插電式混合動力車，一共有29 種車款可供選擇。

2011 年至今，電動車銷量已從當年的2 萬台，增加到12 萬台，等於成長了6 倍，不過，若以全美的車輛銷售量來看，這個數字依然只占了美國車輛銷量不到1%，就且待Navigant Research 對於電動車2016 年的銷量預測是否真能達標了。

（本文授權自《》──〈〉。照片來源：shared by CC 2.0）

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

※評比前十大台北網頁設計、台北網站設計公司知名案例作品心得分享

※智慧手機時代的來臨，RWD網頁設計已成為網頁設計推薦首選

上海市半年內推出八項新能源汽車政策

2020-03-162020-03-16 admin

2016上半年，中國與新能源汽車相關的中央及地方各類政策已經推出了80餘項，包括新能源汽車及充電基礎設施推廣規劃、補貼標準、充電費、充電服務費標準、專項資金補貼、及指標交通等配套優惠政策。雖然政策密集，但2016上半年新能源汽車產業的發展依然掣肘於政策的完善性。 2016年上半年，上海市共推出了8項新能源汽車相關政策，包括購買和使用新能源汽車暫行辦法、操作流程、補貼標準，充電基礎設施建設規劃、補貼標準，新能源汽車專項資金，支援新能源貨車推廣計畫，促進新能源汽車分時租賃業務發展指南，以及嘉定區補貼方案。這些政策在內容上覆蓋了從生產、銷售、購買、使用整個新能源汽車生產鏈，也對技術研發給予專項資金的鼓勵支援，同時覆蓋了新能源汽車的分時租賃業務。相對比2015年上海市新能源汽車的補貼標準，2016年上海的補貼標準在乘用車、客車及專用車領域都有所消退，同時，上海市還首次提出了補貼「按量退坡（減少）」的概念，即超過一定量後補貼會繼續減少，這體現了上海由補貼向市場引導新能源汽車發展的策略。除了補貼標準的降低外，上海市對補貼的審批也更加嚴格，對廠商的銷售資格及消費者的購買資格都做出了嚴格要求。除此之外，上海市的推廣政策還照顧到了分時租賃業務，通過支持運營車輛額度需求及相關補貼政策鼓勵新能源汽車分時租賃業務發展。
文章來源：第一電動網（中國大陸）

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※想知道網站建置、網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計及後台網頁設計

※不管是台北網頁設計公司、台中網頁設計公司，全省皆有專員為您服務

※Google地圖已可更新顯示潭子電動車充電站設置地點!!

※帶您來看台北網站建置，台北網頁設計，各種案例分享

go中的關鍵字-defer

2020-03-132020-03-13 admin

1. defer的使用

　　defer 延遲調用。我們先來看一下，有defer關鍵字的代碼執行順序：

1 func main() {
2     defer func() {
3         fmt.Println("1號輸出")
4     }()
5     defer func() {
6         fmt.Println("2號輸出")
7     }()
8 }

　　輸出結果：

1 2號出來
2 1號出來

　　結論：多個defer的執行順序是倒序執行（同入棧先進后出）。

　　由例子可以看出來，defer有延遲生效的作用，先使用defer的語句延遲到最後執行。

1.1 defer與返回值之間的順序

 1 func defertest() int
 2 
 3 func main() {
 4     fmt.Println("main:", defertest())
 5 }
 6 
 7 func defertest() int {
 8     var i int
 9     defer func() {
10         i++
11         fmt.Println("defer2的值:", i)
12     }()
13     defer func() {
14         i++
15         fmt.Println("defer1的值:", i)
16     }()
17     return i
18 }

　　輸出結果：

1 defer1的值: 1
2 defer2的值: 2
3 main: 0

　　結論：return最先執行->return負責將結果寫入返回值中->接着defer開始執行一些收尾工作->最後函數攜帶當前返回值退出

　　return的時候已經先將返回值給定義下來了，就是0，由於i是在函數內部聲明所以即使在defer中進行了++操作，也不會影響return的時候做的決定。

 1 func test() (i int)
 2 
 3 func main() {
 4     fmt.Println("main:", test())
 5 }
 6 
 7 func test() (i int) {
 8     defer func() {
 9         i++
10         fmt.Println("defer2的值:", i)
11     }()
12     defer func() {
13         i++
14         fmt.Println("defer1的值:", i)
15     }()
16     return i
17 }

　　詳解：由於返回值提前聲明了，所以在return的時候決定的返回值還是0，但是後面兩個defer執行後進行了兩次++，將i的值變為2，待defer執行完后，函數將i值進行了返回。

2. defer定義和執行

 1 func test(i *int) int {
 2     return *i
 3 }
 4 
 5 func main(){
 6     var i = 1
 7 
 8     // defer定義的時候test(&i)的值就已經定了，是1，後面就不會變了
 9     defer fmt.Println("i1 ="  , test(&i))
10     i++
11 
12     // defer定義的時候test(&i)的值就已經定了，是2，後面就不會變了
13     defer fmt.Println("i2 ="  , test(&i))
14 
15     // defer定義的時候，i就已經確定了是一個指針類型，地址上的值變了，這裏跟着變
16     defer func(i *int) {
17         fmt.Println("i3 ="  , *i)
18     }(&i)
19 
20     // defer定義的時候i的值就已經定了，是2，後面就不會變了
21     defer func(i int) {
22         //defer 在定義的時候就定了
23         fmt.Println("i4 ="  , i)
24     }(i)
25 
26     defer func() {
27         // 地址，所以後續跟着變
28         var c = &i
29         fmt.Println("i5 ="  , *c)
30     }()
31 
32     // 執行了 i=11 后才調用，此時i值已是11
33     defer func() {
34         fmt.Println("i6 ="  , i)
35     }()
36 
37     i = 11
38 }

　　結論：會先將defer后函數的參數部分的值(或者地址)給先下來【你可以理解為()裡頭的會先確定】，後面函數執行完，才會執行defer后函數的{}中的邏輯。

例題分析

 1 //例子1
 2 func f() (result int) {
 3     defer func() {
 4         result++
 5     }()
 6     return 0
 7 }
 8 //例子2
 9 func f() (r int) {
10      t := 5
11      defer func() {
12        t = t + 5
13      }()
14      return t
15 }
16 //例子3
17 func f() (r int) {
18     defer func(r int) {
19           r = r + 5
20     }(r)
21     return 1
22 }

　　例1的正確答案不是0，例2的正確答案不是10，例3的正確答案不是6……

　　這裏先說一下返回值。defer是在return之前執行的。這條規則毋庸置疑，但最重要的一點是要明白，return xxx這一條語句並不是一條原子指令!

　　函數返回的過程：先給返回值賦值，然後調用defer表達式，最後才是返回到調用函數中。defer表達式可能會在設置函數返回值之後，且在返回到調用函數之前去修改返回值，使最終的函數返回值與你想象的不一致。

　　return xxx 可被改寫成:

1 返回值 = xxx
2 調用defer函數
3 空的return

　　所以例子也可以改寫成：

 1 //例1
 2 func f() (result int) {
 3      result = 0  //return語句不是一條原子調用，return xxx其實是賦值＋ret指令
 4      func() { //defer被插入到return之前執行，也就是賦返回值和ret指令之間
 5          result++
 6      }()
 7      return
 8 }
 9 //例2
10 func f() (r int) {
11      t := 5
12      r = t //賦值指令
13      func() {        //defer被插入到賦值與返回之間執行，這個例子中返回值r沒被修改過
14          t = t + 5
15      }
16      return        //空的return指令
17 }
18 例3
19 func f() (r int) {
20      r = 1  //給返回值賦值
21      func(r int) {        //這裏改的r是傳值傳進去的r，不會改變要返回的那個r值
22           r = r + 5
23      }(r)
24      return        //空的return
25 }

　　所以例1的結果是1，例2的結果是5，例3的結果是1.

3. defer內部原理

　　從例子開始看：

1 packmage main
2 
3 import()
4 
5 func main() {
6   defer println("這是一個測試")
7 }

　　反編譯一下看看：

 1   src $ go build -o test test.go
 2   src $ go tool objdump -s "main\.main" test

 1 TEXT main.main(SB) /Users/tushanshan/go/src/test3.go
 2   test3.go:5        0x104ea70        65488b0c2530000000      MOVQ GS:0x30, CX
 3   test3.go:5        0x104ea79        483b6110                CMPQ 0x10(CX), SP
 4   test3.go:5        0x104ea7d        765f                    JBE 0x104eade
 5   test3.go:5        0x104ea7f        4883ec28                SUBQ $0x28, SP
 6   test3.go:5        0x104ea83        48896c2420              MOVQ BP, 0x20(SP)
 7   test3.go:5        0x104ea88        488d6c2420              LEAQ 0x20(SP), BP
 8   test3.go:6        0x104ea8d        c7042410000000          MOVL $0x10, 0(SP)
 9   test3.go:6        0x104ea94        488d05e5290200          LEAQ go.func.*+57(SB), AX
10   test3.go:6        0x104ea9b        4889442408              MOVQ AX, 0x8(SP)
11   test3.go:6        0x104eaa0        488d05e6e50100          LEAQ go.string.*+173(SB), AX
12   test3.go:6        0x104eaa7        4889442410              MOVQ AX, 0x10(SP)
13   test3.go:6        0x104eaac        48c744241804000000      MOVQ $0x4, 0x18(SP)
14   test3.go:6        0x104eab5        e8b631fdff              CALL runtime.deferproc(SB)
15   test3.go:6        0x104eaba        85c0                    TESTL AX, AX
16   test3.go:6        0x104eabc        7510                    JNE 0x104eace
17   test3.go:7        0x104eabe        90                      NOPL
18   test3.go:7        0x104eabf        e83c3afdff              CALL runtime.deferreturn(SB)
19   test3.go:7        0x104eac4        488b6c2420              MOVQ 0x20(SP), BP
20   test3.go:7        0x104eac9        4883c428                ADDQ $0x28, SP
21   test3.go:7        0x104eacd        c3                      RET
22   test3.go:6        0x104eace        90                      NOPL
23   test3.go:6        0x104eacf        e82c3afdff              CALL runtime.deferreturn(SB)
24   test3.go:6        0x104ead4        488b6c2420              MOVQ 0x20(SP), BP
25   test3.go:6        0x104ead9        4883c428                ADDQ $0x28, SP
26   test3.go:6        0x104eadd        c3                      RET
27   test3.go:5        0x104eade        e8cd84ffff              CALL runtime.morestack_noctxt(SB)
28   test3.go:5        0x104eae3        eb8b                    JMP main.main(SB)
29   :-1               0x104eae5        cc                      INT $0x3
30   :-1               0x104eae6        cc                      INT $0x3
31   :-1               0x104eae7        cc                      INT $0x3

　　編譯器將defer處理成兩個函數調用 deferproc 定義一個延遲調用對象，然後在函數結束前通過 deferreturn 完成最終調用。在defer出現的地方，插入了指令call runtime.deferproc，然後在函數返回之前的地方，插入指令call runtime.deferreturn。

內部結構

 1 //defer
 2 type _defer struct {
 3    siz     int32   // 參數的大小
 4    started bool    // 是否執行過了
 5    sp      uintptr // sp at time of defer
 6    pc      uintptr
 7    fn      *funcval 
 8    _panic  *_panic // defer中的panic
 9    link    *_defer // defer鏈表，函數執行流程中的defer，會通過 link這個 屬性進行串聯
10 }
11 //panic
12 type _panic struct {
13    argp      unsafe.Pointer // pointer to arguments of deferred call run during panic; cannot move - known to liblink
14    arg       interface{}    // argument to panic
15    link      *_panic        // link to earlier panic
16    recovered bool           // whether this panic is over
17    aborted   bool           // the panic was aborted
18 }
19 //g
20 type g struct {
21    _panic         *_panic // panic組成的鏈表
22    _defer         *_defer // defer組成的先進后出的鏈表，同棧
23 }

　　因為 defer panic 都是綁定在運行的g上的，這裏也說一下g中與 defer panic相關的屬性

　　再把defer, panic, recover放一起看一下：

1 func main() {
2     defer func() {
3         recover()
4     }()
5     panic("error")
6 }

　　反編譯結果：

1 go build -gcflags=all="-N -l" main.go
2 go tool objdump -s "main.main" main

1 go tool objdump -s "main\.main" main | grep CALL
2   main.go:4             0x4548d0                e81b00fdff              CALL runtime.deferproc(SB)              
3   main.go:7             0x4548f2                e8b90cfdff              CALL runtime.gopanic(SB)                
4   main.go:4             0x4548fa                e88108fdff              CALL runtime.deferreturn(SB)            
5   main.go:3             0x454909                e85282ffff              CALL runtime.morestack_noctxt(SB)       
6   main.go:5             0x4549a6                e8d511fdff              CALL runtime.gorecover(SB)              
7   main.go:4             0x4549b5                e8a681ffff              CALL runtime.morestack_noctxt(SB)

　　defer 關鍵字首先會調用 runtime.deferproc 定義一個延遲調用對象，然後再函數結束前，調用 runtime.deferreturn 來完成 defer 定義的函數的調用

　　panic 函數就會調用 runtime.gopanic 來實現相關的邏輯

　　recover 則調用 runtime.gorecover 來實現 recover 的功能

deferproc

　　根據 defer 關鍵字後面定義的函數 fn 以及參數的size，來創建一個延遲執行的函數，並將這個延遲函數，掛在到當前g的 _defer 的鏈表上，下面是deferproc的實現：

 1 func deferproc(siz int32, fn *funcval) { // arguments of fn follow fn
 2    sp := getcallersp()
 3    argp := uintptr(unsafe.Pointer(&fn)) + unsafe.Sizeof(fn)
 4    callerpc := getcallerpc()
 5    // 獲取一個_defer對象， 並放入g._defer鏈表的頭部
 6    d := newdefer(siz)
 7      // 設置defer的fn pc sp等，後面調用
 8    d.fn = fn
 9    d.pc = callerpc
10    d.sp = sp
11    switch siz {
12    case 0:
13       // Do nothing.
14    case sys.PtrSize:
15       // _defer 後面的內存 存儲 argp的地址信息
16       *(*uintptr)(deferArgs(d)) = *(*uintptr)(unsafe.Pointer(argp))
17    default:
18       // 如果不是指針類型的參數，把參數拷貝到 _defer 的後面的內存空間
19       memmove(deferArgs(d), unsafe.Pointer(argp), uintptr(siz))
20    }
21    return0()
22 }

　　通過newproc 獲取一個 _defer 的對象，並加入到當前g的 _defer 鏈表的頭部，然後再把參數或參數的指針拷貝到獲取到的 _defer對象的後面的內存空間。

　　再看看newdefer 的實現：

 1 func newdefer(siz int32) *_defer {
 2    var d *_defer
 3    // 根據 size 通過deferclass判斷應該分配的 sizeclass，就類似於 內存分配預先確定好幾個sizeclass，然後根據size確定sizeclass，找對應的緩存的內存塊
 4    sc := deferclass(uintptr(siz))
 5    gp := getg()
 6    // 如果sizeclass在既定的sizeclass範圍內，去g綁定的p上找
 7    if sc < uintptr(len(p{}.deferpool)) {
 8       pp := gp.m.p.ptr()
 9       if len(pp.deferpool[sc]) == 0 && sched.deferpool[sc] != nil {
10          // 當前sizeclass的緩存數量==0，且不為nil，從sched上獲取一批緩存
11          systemstack(func() {
12             lock(&sched.deferlock)
13             for len(pp.deferpool[sc]) < cap(pp.deferpool[sc])/2 && sched.deferpool[sc] != nil {
14                d := sched.deferpool[sc]
15                sched.deferpool[sc] = d.link
16                d.link = nil
17                pp.deferpool[sc] = append(pp.deferpool[sc], d)
18             }
19             unlock(&sched.deferlock)
20          })
21       }
22       // 如果從sched獲取之後，sizeclass對應的緩存不為空，分配
23       if n := len(pp.deferpool[sc]); n > 0 {
24          d = pp.deferpool[sc][n-1]
25          pp.deferpool[sc][n-1] = nil
26          pp.deferpool[sc] = pp.deferpool[sc][:n-1]
27       }
28    }
29    // p和sched都沒有找到 或者 沒有對應的sizeclass，直接分配
30    if d == nil {
31       // Allocate new defer+args.
32       systemstack(func() {
33          total := roundupsize(totaldefersize(uintptr(siz)))
34          d = (*_defer)(mallocgc(total, deferType, true))
35       })
36    }
37    d.siz = siz
38    // 插入到g._defer的鏈表頭
39    d.link = gp._defer
40    gp._defer = d
41    return d
42 }

　　newdefer的作用是獲取一個_defer對象，並推入 g._defer鏈表的頭部。根據size獲取sizeclass，對sizeclass進行分類緩存，這是內存分配時的思想，先去p上分配，然後批量從全局 sched上獲取到本地緩存，這種二級緩存的思想真的在go源碼的各個部分都有。

deferreturn

 1 func deferreturn(arg0 uintptr) {
 2    gp := getg()
 3    // 獲取g defer鏈表的第一個defer，也是最後一個聲明的defer
 4    d := gp._defer
 5    // 沒有defer，就不需要干什麼事了
 6    if d == nil {
 7       return
 8    }
 9    sp := getcallersp()
10    // 如果defer的sp與callersp不匹配，說明defer不對應，有可能是調用了其他棧幀的延遲函數
11    if d.sp != sp {
12       return
13    }
14    // 根據d.siz，把原先存儲的參數信息獲取並存儲到arg0裏面
15    switch d.siz {
16    case 0:
17       // Do nothing.
18    case sys.PtrSize:
19       *(*uintptr)(unsafe.Pointer(&arg0)) = *(*uintptr)(deferArgs(d))
20    default:
21       memmove(unsafe.Pointer(&arg0), deferArgs(d), uintptr(d.siz))
22    }
23    fn := d.fn
24    d.fn = nil
25    // defer用過了就釋放了，
26    gp._defer = d.link
27    freedefer(d)
28    // 跳轉到執行defer
29    jmpdefer(fn, uintptr(unsafe.Pointer(&arg0)))
30 }

freedefer

　　釋放defer用到的函數，應該跟調度器、內存分配的思想是一樣的。

 1 func freedefer(d *_defer) {
 2    // 判斷defer的sizeclass
 3    sc := deferclass(uintptr(d.siz))
 4    // 超出既定的sizeclass範圍的話，就是直接分配的內存，那就不管了
 5    if sc >= uintptr(len(p{}.deferpool)) {
 6       return
 7    }
 8    pp := getg().m.p.ptr()
 9    // p本地sizeclass對應的緩衝區滿了，批量轉移一半到全局sched
10    if len(pp.deferpool[sc]) == cap(pp.deferpool[sc]) {
11       // 使用g0來轉移
12       systemstack(func() {
13          var first, last *_defer
14          for len(pp.deferpool[sc]) > cap(pp.deferpool[sc])/2 {
15             n := len(pp.deferpool[sc])
16             d := pp.deferpool[sc][n-1]
17             pp.deferpool[sc][n-1] = nil
18             pp.deferpool[sc] = pp.deferpool[sc][:n-1]
19             // 先將需要轉移的那批defer對象串成一個鏈表
20             if first == nil {
21                first = d
22             } else {
23                last.link = d
24             }
25             last = d
26          }
27          lock(&sched.deferlock)
28          // 把這個鏈表放到sched.deferpool對應sizeclass的鏈表頭
29          last.link = sched.deferpool[sc]
30          sched.deferpool[sc] = first
31          unlock(&sched.deferlock)
32       })
33    }
34    // 清空當前要釋放的defer的屬性
35    d.siz = 0
36    d.started = false
37    d.sp = 0
38    d.pc = 0
39    d.link = nil
40 
41    pp.deferpool[sc] = append(pp.deferpool[sc], d)
42 }

gopanic

 1 func gopanic(e interface{}) {
 2    gp := getg()
 3 
 4    var p _panic
 5    p.arg = e
 6    p.link = gp._panic
 7    gp._panic = (*_panic)(noescape(unsafe.Pointer(&p)))
 8 
 9    atomic.Xadd(&runningPanicDefers, 1)
10    // 依次執行 g._defer鏈表的defer對象
11    for {
12       d := gp._defer
13       if d == nil {
14          break
15       }
16 
17       // If defer was started by earlier panic or Goexit (and, since we're back here, that triggered a new panic),
18       // take defer off list. The earlier panic or Goexit will not continue running.
19       // 正常情況下，defer執行完成之後都會被移除，既然這個defer沒有移除，原因只有兩種： 1. 這個defer裏面引發了panic 2. 這個defer裏面引發了 runtime.Goexit，但是這個defer已經執行過了，需要移除，如果引發這個defer沒有被移除是第一個原因，那麼這個panic也需要移除，因為這個panic也執行過了，這裏給panic增加標誌位，以待後續移除
20       if d.started {
21          if d._panic != nil {
22             d._panic.aborted = true
23          }
24          d._panic = nil
25          d.fn = nil
26          gp._defer = d.link
27          freedefer(d)
28          continue
29       }
30       d.started = true
31 
32       // Record the panic that is running the defer.
33       // If there is a new panic during the deferred call, that panic
34       // will find d in the list and will mark d._panic (this panic) aborted.
35       // 把當前的panic 綁定到這個defer上面，defer裏面有可能panic，這種情況下就會進入到 上面d.started 的邏輯裏面，然後把當前的panic終止掉，因為已經執行過了 
36       d._panic = (*_panic)(noescape(unsafe.Pointer(&p)))
37       // 執行defer.fn
38       p.argp = unsafe.Pointer(getargp(0))
39       reflectcall(nil, unsafe.Pointer(d.fn), deferArgs(d), uint32(d.siz), uint32(d.siz))
40       p.argp = nil
41 
42       // reflectcall did not panic. Remove d.
43       if gp._defer != d {
44          throw("bad defer entry in panic")
45       }
46       // 解決defer與panic的綁定關係，因為 defer函數已經執行完了，如果有panic或Goexit就不會執行到這裏了
47       d._panic = nil
48       d.fn = nil
49       gp._defer = d.link
50 
51       // trigger shrinkage to test stack copy. See stack_test.go:TestStackPanic
52       //GC()
53 
54       pc := d.pc
55       sp := unsafe.Pointer(d.sp) // must be pointer so it gets adjusted during stack copy
56       freedefer(d)
57       // panic被recover了，就不需要繼續panic了，繼續執行剩餘的代碼
58       if p.recovered {
59          atomic.Xadd(&runningPanicDefers, -1)
60 
61          gp._panic = p.link
62          // Aborted panics are marked but remain on the g.panic list.
63          // Remove them from the list.
64          // 從panic鏈表中移除aborted的panic，下面解釋
65          for gp._panic != nil && gp._panic.aborted {
66             gp._panic = gp._panic.link
67          }
68          if gp._panic == nil { // must be done with signal
69             gp.sig = 0
70          }
71          // Pass information about recovering frame to recovery.
72          gp.sigcode0 = uintptr(sp)
73          gp.sigcode1 = pc
74          // 調用recovery， 恢復當前g的調度執行
75          mcall(recovery)
76          throw("recovery failed") // mcall should not return
77       }
78    }
79      // 打印panic信息
80    preprintpanics(gp._panic)
81      // panic
82    fatalpanic(gp._panic) // should not return
83    *(*int)(nil) = 0      // not reached
84 }

　　看下裏面gp._panic.aborted 的作用：

 1 func main() {
 2    defer func() { // defer1
 3       recover()
 4    }()
 5    panic1()
 6 }
 7 
 8 func panic1() {
 9    defer func() {  // defer2
10       panic("error1") // panic2
11    }()
12    panic("error")  // panic1
13 }

　　執行順序詳解：

當執行到 panic("error") 時

　　g._defer鏈表： g._defer->defer2->defer1

　　g._panic鏈表：g._panic->panic1

當執行到 panic("error1") 時

　　g._defer鏈表： g._defer->defer2->defer1

　　g._panic鏈表：g._panic->panic2->panic1

繼續執行到 defer1 函數內部，進行recover()
此時會去恢復 panic2 引起的 panic， panic2.recovered = true，應該順着g._panic鏈表繼續處理下一個panic了，但是我們可以發現 panic1 已經執行過了，這也就是下面的代碼的邏輯了，去掉已經執行過的panic

1 for gp._panic != nil && gp._panic.aborted {
2    gp._panic = gp._panic.link
3 }

panic的邏輯：

　　程序在遇到panic的時候，就不再繼續執行下去了，先把當前panic 掛載到 g._panic 鏈表上，開始遍歷當前g的g._defer鏈表，然後執行_defer對象定義的函數等，如果 defer函數在調用過程中又發生了 panic，則又執行到了 gopanic函數，最後，循環打印所有panic的信息，並退出當前g。然而，如果調用defer的過程中，遇到了recover，則繼續進行調度（mcall(recovery)）。

recovery

 1 func recovery(gp *g) {
 2    // Info about defer passed in G struct.
 3    sp := gp.sigcode0
 4    pc := gp.sigcode1
 5    // Make the deferproc for this d return again,
 6    // this time returning 1.  The calling function will
 7    // jump to the standard return epilogue.
 8    // 記錄defer返回的sp pc
 9    gp.sched.sp = sp
10    gp.sched.pc = pc
11    gp.sched.lr = 0
12    gp.sched.ret = 1
13    // 重新恢復執行調度
14    gogo(&gp.sched)
15 }

gorecover

　　gorecovery 僅僅只是設置了 g._panic.recovered 的標誌位

 1 func gorecover(argp uintptr) interface{} {
 2    gp := getg()
 3    p := gp._panic
 4    // 需要根據 argp的地址，判斷是否在defer函數中被調用
 5    if p != nil && !p.recovered && argp == uintptr(p.argp) {
 6       // 設置標誌位，上面gopanic中會對這個標誌位做判斷
 7       p.recovered = true
 8       return p.arg
 9    }
10    return nil
11 }

goexit

　　當手動調用 runtime.Goexit() 退出的時候，defer函數也會執行:

 1 func Goexit() {
 2     // Run all deferred functions for the current goroutine.
 3     // This code is similar to gopanic, see that implementation
 4     // for detailed comments.
 5     gp := getg()
 6   // 遍歷defer鏈表
 7     for {
 8         d := gp._defer
 9         if d == nil {
10             break
11         }
12     // 如果 defer已經執行過了，與defer綁定的panic 終止掉
13         if d.started {
14             if d._panic != nil {
15                 d._panic.aborted = true
16                 d._panic = nil
17             }
18             d.fn = nil
19       // 從defer鏈表中移除
20             gp._defer = d.link
21       // 釋放defer
22             freedefer(d)
23             continue
24         }
25     // 調用defer內部函數
26         d.started = true
27         reflectcall(nil, unsafe.Pointer(d.fn), deferArgs(d), uint32(d.siz), uint32(d.siz))
28         if gp._defer != d {
29             throw("bad defer entry in Goexit")
30         }
31         d._panic = nil
32         d.fn = nil
33         gp._defer = d.link
34         freedefer(d)
35         // Note: we ignore recovers here because Goexit isn't a panic
36     }
37   // 調用goexit0，清除當前g的屬性，重新進入調度
38     goexit1()
39 }

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

【其他文章推薦】

※USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

※評比前十大台北網頁設計、台北網站設計公司知名案例作品心得分享

※智慧手機時代的來臨，RWD網頁設計已成為網頁設計推薦首選

nodejs入門之模塊

2020-03-132020-03-13 admin

nodejs模塊語法與開閉原則
nodejs模塊的底層實現

一、nodejs模塊語法與開閉原則

關於nodejs模塊我在之前的兩篇博客中都有涉及，但都沒有對nodejs模塊的底層做做任何探討，但是為了使相關內容更方便查看比對理解，這裏還是先引入一下之前兩篇博客的連接：

1.1 exports、module.exports、require()實現模塊導出導入：

 1 //示例一：導出原始值數據
 2 //a.js--用於導出數據
 3 let a = 123;
 4 module.exports.a=a;
 5 //inde.js--用於導入a模塊的數據
 6 let aModule = require('./a.js');
 7 console.log(aModule.a); //123
 8 
 9 //示例二：導出引用值數據
10 //a.js--同上
11 function foo(val){ 
12     console.log(val);
13 }
14 module.exports.foo = foo;
15 //index.js--同上
16 let aModule = require('./a.js');
17 let str = "this is 'index' module"
18 aModule.foo(str); //this is 'index' module
19 
20 //示例三：導出混合數據
21 a.js--同上
22 let a = 123;
23 function foo(val){ 
24     console.log(val);
25 }
26 module.exports = {
27     a:a,
28     foo:foo
29 }
30 //inde.js--同上
31 let aModule = require('./a.js');
32 let str = "this is 'index' module"
33 console.log(aModule.a);//123
34 aModule.foo(str); //this is 'index' module

在上面這些示例中，沒有演示exports的導出，暫時可以把它看作與同等於module.exports，例如：

 1 //a.js -- 導出模塊
 2 let a = 123;
 3 function foo(val){ 
 4     console.log(val);
 5 }
 6 exports.a = a;
 7 exports.foo = foo;
 8 
 9 //inde.js -- 引用模塊a
10 let aModule = require('./a.js');
11 let str = "this is 'index' module"
12 console.log(aModule.a);//123
13 aModule.foo(str); //this is 'index' module

但是使用exports導出模塊不能這麼寫：

 1 //a.js
 2 let a = 123;
 3 function foo(val){ 
 4     console.log(val);
 5 }
 6 exports = {
 7     a:a,
 8     foo:foo
 9 }
10 
11 //index.js
12 let aModule = require('./a.js');
13 let str = "this is 'index' module"
14 console.log(aModule);// {} -- 一個空對象

至於為什麼不能這麼寫，暫時不在這裏闡述，下一節關於nodejs模塊底層實現會具體的分析介紹，這裏先來介紹nodejs模塊的一個設計思想。

1.2 nodejs模塊的開閉原則設計實現

1 //a.js -- 導出模塊
2 let num = 123;
3 let str = "this is module 'a'";
4 exports.a = a;
5 
6 //index.js -- 引用模塊a
7 let aModule = require('./a.js');
8 console.log(aModule.num);//123
9 console.log(aModule.str);//undefined

這裏你會發現只有被exports執行了導出的num成員才能被正常導出，而str成員沒有被執行導出，在依賴a.js模塊的index.js中是不能引用到a.js模塊中的str成員。可能你會說這不是很正常嗎？都沒有導出怎麼引用呢？

不錯，這是一個非常正常情況，因為語法就告訴了我們，要想引用一個模塊的成員就必須先在被引用的模塊中導出該成員。然而這裏要討論的當然不會是導出與引用這個問題，而是模塊給我實現了一個非常友好的設計，假設我現在在a.js中有成員str，在index.js模塊中也有成員str，這回衝突嗎？顯然是不會的，即使在a.js中導出str並且在index.js中引用a.js模塊，因為index.js要使用a.js模塊的成員str，需要使用接收模塊變量aModule.str來使用。

 1 //a.js
 2 let num = 123;
 3 let str = "this is module 'a'";
 4 exports.num = num;
 5 exports.str = str;
 6 
 7 //index.js
 8 let aModule = require('./a.js');
 9 let str = "this is module 'index'"
10 console.log(aModule.num);//123
11 console.log(aModule.str);//this is module 'a'
12 console.log(str);//this is module 'index'

基於開閉原則的設計方式，封閉可以讓模塊的內部實現隱藏起來，開放又可以友好的實現模塊之間的相互依賴，這相對於之前我們常用的回調函數解決方案，程序設計變得更清晰，代碼復用變得更靈活，更關鍵的是還解決了js中一個非常棘手的問題——命名衝突問題，上面的示例就是最好的證明。這裏需要拋出一個問題，看示例：

1 //下面這種寫法有什麼問題？
2 //a.js
3 let num = 123;
4 module.exports = num;
5 
6 //index.js
7 let aModule = require('./a.js');
8 let str = "this is module 'index'"
9 console.log(aModule);//123

這種寫法不會報錯，也能正常達到目前的需求，如果從能解決目前的功能需求角度來說，它沒錯。但是開閉原則的重要思想就是讓模塊保持相對封閉，又有更好的拓展性，這樣寫顯然不合適，比如就上面的代碼寫完上線以後，業務又出現了一個新的需求需要a.js模塊導出一個成員str，這時候顯然需要同時更改a.js模塊和index.js模塊，即使新需求不需要index.js來實現也是需要改的。所以維持模塊的開閉原則是良好的編碼風格。

二、nodejs模塊的底層實現原理

2.1 module.exports與exports的區別：

//a.js
console.log(module.exports == exports);//true

//然後在控制台直接執行a.js模塊
node a.js

實際上它們是沒有區別的，那為什麼在之前的exports不能直接等於一個對象，而module.exports可以呢？這關乎於js的引用值指向問題：

當export被賦值一個對象時，就發生了一下變化：

這時候我們可以確定node不會導出exports，因為前面的示例已經說明了這一點，但是值得我們繼續思考的是，node模塊是依據module.exports、exports、還是它們指向的初始對象呢？這裏你肯定會說是module.exports，因為前面已經有示例是module.exports指向一個新的對象被成功導出，但是我並不覺得前面那些示例能說服我，比如下面這種情況：

 1 //a.js模塊
 2 let num = 123;
 3 function foo(val){
 4     console.log(val);
 5 }
 6 module.exports = {
 7     num:num
 8 }
 9 exports = {
10     foo:foo
11 }
12 //index.js模塊
13 let aModule = require('./a.js');
14 console.log(aModule);//這裡會打印出什麼？

我們現不測試也不猜測，先通過下面的示圖來看下現在的a.js模塊中module.exports、exports、以及它們兩初始指向的空對象的關係圖：

這時候我們來看一下index.js執行會輸出什麼？

{ num: 123 }

所以從這個結果可以看出，最後require()最後導入的是被引用模塊的module.exports。探討到這裏的時候並沒有到達node模塊的終點，我們這裏module.exports、exports、require()是從哪裡來的？node系統內置變量？還是別的？

2.2 node模塊的底層實現原理

這部分的內容其實也沒有太多可以說的，就前面提出來的問題其實有一個方式就可以讓你一目瞭然，只需要在一個js文件中編寫以下代碼，然後使用node執行這個js文件就可以了：

1 console.log(require);      // 一個方法
2 console.log(module);       //  一個對象
3 console.log(exports);      //  一個空對象
4 console.log(__dirname);    //  當前模塊所在路徑
5 console.log(__filename);   //  當前文件的路徑

這時因為node模塊實際上底層是被放到一個立即執行函數內（不要在乎xyz這個名稱，因為我也不知道node底層到底用的什麼名稱），這些變量其實就是這個函數的參數，這個函數大概是一下形式：

1 function xyz(module.exports,require,module,__filename,__dirname){
2     //...
3     //  這裏就是我們在模塊中寫入的代碼
4     //...
5     return module.exports;
6 }

通過上面的推斷就可以得到下面這樣的結果：

1 console.log(module.exports == arguments[0]);//true
2 console.log(require == arguments[1]);//true
3 console.log(module == arguments[2]);//true
4 console.log(__filename == arguments[3]);//true
5 console.log(__dirname == arguments[4]);//true

通過執行這段打印代碼也確實可以得到這樣的結果，到這裏又有一個值得我們關注的內容，就是每個模塊的module參數：

 1 console.log(module);
 2 Module {
 3     id: '.',//當前模塊的id都是'.'，在後面的parent和children裏面的模塊對象上的id就是的對應模塊的filename
 4     exports: {},//這裡是模塊導出對象
 5     parent: null,//這裡是當前模塊被那些模塊引用的模塊對象列表，意思是當前模塊作為那些模塊的父級模塊
 6     filename:''，//這裡是當前文件路徑的絕對路徑
 7     loaded: false,//模塊加載狀態，如果在模塊內部輸出module對象它永遠都會是false，因為只有這個模塊加載完成之後才會被修改成true
 8     children: [
 9         // 這裡是引用模塊module對象列表，意思是當前模塊作為了那些模塊的子模塊
10     ],
11     paths:[ 
12         // 這裡是外部模塊包的路徑列表，從最近的路徑（模塊所在同級路徑）到系統盤路徑所有的node_modules文件夾路徑
13      ] 
14     }

到這裡有可能你還會問為什麼底層實現裏面只有module.exports，沒有export，這個解釋起來真的費勁，下面這一行代碼幫你搞定：

let exports = module.exports;

這篇博客主要介紹了node模塊的內部內容，並未就node模塊基於commonjs規範做任何介紹，是因為在之前的博客中已經有了非常全面的解析，詳細參考博客開始時的連接，關於node模塊加載相關內容也是在那篇博客。

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

【其他文章推薦】

※想知道網站建置、網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計及後台網頁設計

※不管是台北網頁設計公司、台中網頁設計公司，全省皆有專員為您服務

※Google地圖已可更新顯示潭子電動車充電站設置地點!!

※帶您來看台北網站建置，台北網頁設計，各種案例分享

中國國產電動汽車續航超500km 遠超特斯拉

2020-03-122020-03-12 admin

由中國中能東道集團自主研發，與四川野馬汽車公司聯合生產的「野馬u能純電動汽車」在大連進行了路演。行駛502公里，仍有餘電22%，這遠遠優於領先國際先進水準的特斯拉（400公里）和比亞迪（260公里），其續航能力是目前中國國內普通純電動汽車的3倍。
該純電動汽車不依賴專用充電站或充電樁，220v民用電即可完成充電。該車已進入國家工信部目錄並實現量產。中能東道集團已規劃建設5個電動汽車「動力總成核新部件產業基地」，可實現年產50萬套目標。 2014年大連市政府專門出臺檔要求2016年全市公交、公務、郵政、環衛、出租等公共領域新能源汽車置換率不低於30%；個人購買使用新能源汽車，除享受由中央和地方財政分別給予總車款60%左右的政策性補貼外，還可享受免車輛購置稅和車輛不限行、電價優惠等鼓勵政策。2016年底大連市區將實現充電樁網路化建設，基本可滿足新能源汽車充電需求。中能東道集團已完成全國31個省管理服務中心佈局，並在大連設立運營中心。野馬u能純電動汽車即將落戶大連。文章來源：新浪新聞看點

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

※評比前十大台北網頁設計、台北網站設計公司知名案例作品心得分享

※智慧手機時代的來臨，RWD網頁設計已成為網頁設計推薦首選

大眾汽車集團「以人為本」戰略再推15款電動汽車

2020-03-112020-03-11 admin

大眾汽車集團攜手旗下大眾汽車、奧迪、斯柯達、西雅特、保時捷、賓利六大品牌及合資企業一汽大眾、上海大眾，聯合發佈了』以人為本」的中國戰略。
該戰略基於三個重要支柱——客戶、員工與社會，這三大支柱將促成創新且滿足消費者需求的車型。作為該戰略的首項活動，大眾汽車集團攜旗下各品牌與合資公司共同向中國家庭捐贈5000多個安全坐椅。作為中國戰略的一部分，大眾汽車集團將在未來三年中不斷推進節能環保車型的生產，加強技術創新，推出一系列低排放車型。據大眾汽車集團總裁兼CEO海茲曼介紹，大眾未來每一款新車型的燃油效率都將比其上一代車型提高20%。大眾汽車集團已宣佈其新能源汽車戰略，計畫在2018年之前再推出超過15款電動車新車型，並從2016年開始國產其中多款車型。到2018年，大眾汽車集團在中國的國產車型將超過35款。文章來源：新京報

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

※評比前十大台北網頁設計、台北網站設計公司知名案例作品心得分享

※智慧手機時代的來臨，RWD網頁設計已成為網頁設計推薦首選

ABB超級電容「閃充」技術，在美獲首筆電動巴士訂單

2020-03-112020-03-11 admin

充電慢、交換電池麻煩，這是全電動車的兩難，尤其是若要發展大眾運輸，全電動公車可不能總是停在路邊充電，為了解決這個問題，機電大廠ABB 推出以超級電容（Supercapacitors）「閃充」技術，如今已經接到第一筆訂單。

2016 年7 月中，美國政府期望推動讓未來的電動車能在10 分鐘內充電完成，不過，其實不用等到未來，ABB 現有的技術就已經讓10 分鐘看起來顯得落伍，因為超級電容可以600kW，在15 秒內閃充，提供電動公車行駛到下一個充電站的電力，而若是需完全充飽電池，也只需要幾分鐘。ABB 已經小規模測試此技術數年。

如今，ABB 接到第一筆商業訂單，來自於公車營運商Tosa，將於日內瓦市郊到機場的公車路線上行駛。這款全電動公車行駛時與一般公車無異，不過總共50 個公車站中，有13 個公車站是充電站，當公車進站時，車頂上會伸出接收器，與車站的充電裝置接觸，在15 秒內閃充，取得可行駛到下一個充電站的電力，而在公車總站，則有200kW充電器，以3~5 分鐘時間將電池完全充飽。

充電站採用超級電容，超級電容的能源密度雖然不如鋰電池，但可緩緩充能，於短時間內一次釋放，除了縮短電動公車的充電時間，使得公車能在靠站時閃充以外，也能減輕電網負擔，因為用電量是平均分攤在緩緩充能的階段，而不是一接上充電座時就突然大增電力需求，充完又需求突降，造成電網調度上的困擾。另一方面，由於分散了負擔，整個充電站可以用較低成本的低電流規劃打造。

這種方式也利於應用太陽能，公車站可設置太陽能發電，為超級電容緩緩充能，充電時間之中萬一有雲飄過，一時中斷供電，也不會產生太大影響。而由於公車班次最多時間是在白天，因此太陽能可發揮相當效用。

ABB 先前將超級電容應用於電車煞車回電，2013 年起，ABB 與Tosa 測試超級電容應用於全電動公車，目前測試結果，其成本已經與柴油車相當，未來隨著相關技術成熟，電池價格進一步下降，將有可能比柴油車更便宜，除了價格以外，全電動公車還有無碳排放、無廢氣空氣污染、噪音低的好處，有助於改善市區生活品質。

這款全電動巴士預定於2018 年全面營運。

來源：

（本文授權轉載自《》──〈〉。圖片來源：）

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※想知道網站建置、網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計及後台網頁設計

※不管是台北網頁設計公司、台中網頁設計公司，全省皆有專員為您服務

※Google地圖已可更新顯示潭子電動車充電站設置地點!!

※帶您來看台北網站建置，台北網頁設計，各種案例分享

新加坡贈比亞迪100張電動計程車牌照打造智慧交通

2020-03-102020-03-10 admin

近日，新加坡政府向比亞迪頒發100個電動計程車牌照，意味著比亞迪正式成為有史以來進入新加坡計程車市場的第一家中國企業。
據悉，此次頒發的100個計程車牌照將用於e6純電動車。宏達同新加坡計程車私人有限公司將負責這100台e6計程車的日常運營和管理，首批車輛預計在今年9月上路運營，100台計程車計畫于明年一季度前全部投入服務。屆時，比亞迪將與其新加坡合作夥伴對這支e6車隊進行即時資料採集、分析和處理，聯合研發自動駕駛電動汽車技術和智慧交通管理系統，以支援新加坡政府打造智慧交通。目前，比亞迪在新加坡的車型由最初的e6逐步擴展到電動巴士、叉車和電動商用車等多種車型，並與GrabCar等交通出行領軍企業展開合作。今年5月30日，比亞迪還攜手SMART集團在新加坡開啟了全球首家純電動車體驗中心，並簽署在新加坡聯合推廣1,000輛e6的合作諒解備忘錄。文章來源：環球網

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

※評比前十大台北網頁設計、台北網站設計公司知名案例作品心得分享

※智慧手機時代的來臨，RWD網頁設計已成為網頁設計推薦首選

車用電池市場變化，傳日產將出售電池事業

2020-03-102020-03-10 admin

日經新聞5日報導，日產汽車（Nissan）將退出車用電池事業，計畫出售和NEC共同設立的合資子公司「Automotive Energy Supply（以下簡稱AES）」，且除了日本電池廠之外、日產已和多家中國廠商展開出售協商；另外，除了AES之外，日產也計畫出售在美國及英國單獨營運的電池生產事業，且據悉多家中國廠商已表達有意收購的意願。預估日產在評估出售額、員工雇用等條件之後，會在今年內敲定出售對象。

報導指出，AES為日產、NEC分別出資51%、49%於2007年設立的公司，主要生產日產電動車Leaf或油電混合車（HV）所需的鋰離子電池，於車用鋰離子電池市場的市佔率僅次於Panasonic位居第2位，2015年度營收為366億日圓，不過因日產研判比起自家生產、向外部電池廠商採購電池較能進一步調降車輛價格，故決定出售ASE持股。

每日新聞5日報導，日產汽車計畫出售ASE持股，且已和Panasonic以及南韓LG化學（LG Chem）等海外廠商展開協商，而據關係人士透露，出售額預估將達數百億日圓。

日經新聞於6日追加報導指出，隨著日產有意退出車用電池事業，NEC也考慮跟進出售ASE持股，而ASE為全球第2大車用鋰離子電池廠，故日產/NEC出售ASE的舉動恐將對全球車用鋰離子電池的勢力版圖帶來巨大變動。

日經指出，供應鋰離子電池給特斯拉（Tesla）的Panasonic目前正面臨中國廠商的猛烈追趕、導致市佔萎縮，根據調查公司Techno Systems Research指出，2014年Panasonic車用鋰離子電池全球市佔率高達47%，而2015年雖維持首位、但市佔率萎縮至34%；反觀「其他廠商」市佔率從14%飆增至33%，而在「其他廠商」中、比亞迪（BYD）等中國廠商佔了大多數。

另外，LG化學、三星SDI等南韓廠商也在後猛追，且日本村田製作所也收購Sony鋰離子電池事業、今後可能將進軍車用市場，故未來日中韓於車用電池市場的競爭料將更趨激烈。

Sony 7月28日宣布，已和村田製作所締結意向確認書，計畫將電池事業出售給村田，雙方預計在2016年10月中旬簽訂具法律約束力的最終契約，且在獲得相關當局同意之後，目標在2017年3月底完成出售手續。

（本文內容由授權提供）

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※想知道網站建置、網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計及後台網頁設計

※不管是台北網頁設計公司、台中網頁設計公司，全省皆有專員為您服務

※Google地圖已可更新顯示潭子電動車充電站設置地點!!

※帶您來看台北網站建置，台北網頁設計，各種案例分享

【Flume】Flume基礎之安裝與使用

2020-03-092020-03-09 admin

1、Flume簡介

(1) Flume提供一個分佈式的，可靠的，對大數據量的日誌進行高效收集、聚集、移動的服務，Flume只能在Unix環境下運行。

(2) Flume基於流式架構，容錯性強，也很靈活簡單。

(3) Flume、Kafka用來實時進行數據收集，Spark、Flink用來實時處理數據，impala用來實時查詢。

2、Flume角色

2.1 Source

用於採集數據，Source是產生數據流的地方，同時Source會將產生的數據流傳輸到Channel，這個有點類似於Java IO部分的Channel。

2.2 Channel

用於橋接Sources和Sinks，類似於一個隊列。

2.3 Sink

從Channel收集數據，將數據寫到目標源(可以是下一個Source，也可以是HDFS或者HBase)。

2.4 Event

傳輸單元，Flume數據傳輸的基本單元，以事件的形式將數據從源頭送至目的地。

3、Flume傳輸過程

source監控某個文件或數據流，數據源產生新的數據，拿到該數據后，將數據封裝在一個Event中，並put到channel后commit提交，channel隊列先進先出，sink去channel隊列中拉取數據，然後寫入到HDFS或其他目標源中。

4、Flume安裝與部署

4.1 上傳包

將flume的gz包上傳到/opt/soft/目錄下；

[root@bigdata111 conf]# rz

若不支持rz命令，則用yum安裝lrzsz命令：

查詢含有rz的yum源，由結果可見，yum源中含有lrzsz.x86_64包；

[root@bigdata111 soft]# yum search rzsz
已加載插件：fastestmirror
Loading mirror speeds from cached hostfile
 * base: mirrors.tuna.tsinghua.edu.cn
 * extras: mirrors.aliyun.com
 * updates: mirrors.aliyun.com
============================================================================================================================= N/S matched: rzsz ==============================================================================================================================
lrzsz.x86_64 : The lrz and lsz modem communications programs

  名稱和簡介匹配 only，使用“search all”試試。

安裝rz命令

[root@bigdata111 soft]# yum -y install lrzsz

4.2 解壓包

將flume解壓到/opt/module/目錄下，並改短名字flume-1.8.0：

[root@bigdata111 soft]# tar -zvxf apache-flume-1.8.0-bin.tar.gz -C /opt/module
[root@bigdata111 module]# mv apache-flume-1.8.0-bin flume-1.8.0

4.3 配置參數

切換到/opt/module/flume-1.8.0/conf目錄，將flume-env.sh.template文件名改為：flume-env.sh

[root@bigdata111 module]# mv flume-env.sh.template flume-env.sh

查詢JAVA_HOME的值；

[root@bigdata111 conf]# echo $JAVA_HOME
/opt/module/jdk1.8.0_144

編輯flume-env.sh，將文件內容中的JAVA_HOME的值修改為上面查到的；

export JAVA_HOME=/opt/module/jdk1.8.0_144

4.4 配置環境變量

在/etc/profile末尾添加flume的家路徑

export FLUME_HOME=/opt/module/flume-1.8.0
export PATH=$PATH:$FLUME_HOME/bin

4.5 驗證flume成功與否

在xshell客戶端下，輸入flu，按tab鍵，看是否能夠自動補全：flume-ng

如果可以自動補全，則代表安裝flume成功，否則失敗。

[root@bigdata112 opt]# flume-ng
Error: Unknown or unspecified command ''

Usage: /opt/module/flume-1.8.0/bin/flume-ng <command> [options]...

commands:
  help                      display this help text
  agent                     run a Flume agent
  avro-client               run an avro Flume client
  version                   show Flume version info
............

4.6 配置其他兩台機器

利用scp命令，配置其他兩台機器；

首先，將flume目錄分發到bigdata112，bigdata113

[root@bigdata111 ~]# scp -r /opt/module/flume-1.8.0/ root@bigdata112:/opt/module/
[root@bigdata111 ~]# scp -r /opt/module/flume-1.8.0/ root@bigdata113:/opt/module/

其次，將/etc/profile環境變量文件分發到bigdata112，bigdata113

[root@bigdata111 ~]# scp -r /etc/profile root@bigdata112:/etc/
[root@bigdata111 ~]# scp -r /etc/profile root@bigdata113:/etc/

最後，在bigdata112，bigdata113上分別刷新環境變量

[root@bigdata112 opt]# source /etc/profile
[root@bigdata113 opt]# source /etc/profile

5、Flume案例

5.1 監控端口數據

目標：Flume監控一端Console，另一端Console發送消息，使被監控端實時显示。

5.1.1 安裝telnet命令

[root@bigdata111 conf]# yum -y install telnet

5.1.2 創建Agent配置文件

在flume根目錄下，新建一個myconf目錄，用於存放自定義conf配置文件；

新建flume-telnet.conf文件，文件內容如下：

# 定義agent
# <自定義agent名>.sources=<自定義source名稱>
a1.sources = r1
# <自定義agent名>.sinks=<自定義sink名稱>
a1.sinks = k1
# <自定義agent名>.channels=<自定義channel名稱>
a1.channels = c1

# 定義source
# <agent名>.sources.<source名稱>.type = 源類型
a1.sources.r1.type = netcat
# <agent名>.sources.<source名稱>.bind = 數據來源服務器
a1.sources.r1.bind = bigdata111
# <agent名>.sources.<source名稱>.port = 自定義未被佔用的端口
a1.sources.r1.port = 44445

# 定義sink
# <agent名>.sinks.<sink名稱>.type = 下沉到目標源的類型
a1.sinks.k1.type = logger

# 定義channel
# <agent名>.channels.<channel名稱>.type = channel的類型
a1.channels.c1.type = memory
# <agent名>.channels.<channel名稱>.capacity = 最大容量
a1.channels.c1.capacity = 1000
# transactionCapacity<=capacity
a1.channels.c1.transactionCapacity = 1000                  

# 雙向鏈接
# <agent名>.sources.<source名稱>.channels = channel名稱
a1.sources.r1.channels = c1
# <agent名>.sinks.<sink名稱>.channel = channel名稱
a1.sinks.k1.channel = c1

5.1.3 啟動flume配置文件

[root@bigdata111 conf]# flume-ng agent --conf /opt/module/flume-1.8.0/conf/ --name a1 --conf-file /opt/module/flume-1.8.0/conf/flume-telnet.conf -Dflume.root.logger==INFO,console

可以簡寫為：

[root@bigdata111 conf]# flume-ng agent --c /opt/module/flume-1.8.0/conf/ --n a1 --f /opt/module/flume-1.8.0/conf/flume-telnet.conf -Dflume.root.logger==INFO,console

5.1.4 發送測試數據

通過其他機器向bigdata111的44445端口發送數據

[root@bigdata112 ~]# telnet bigdata111 44445
Trying 192.168.1.111...
Connected to bigdata111.
Escape character is '^]'.
echo aaaa
OK
echo aaaa
OK
echo bbbbbbbbb
OK

運行結果如圖：

5.2 實時讀取本地文件到HDFS

5.2.1 創建Agent配置文件

創建flume-hdfs配置文件

# 1 agent  若同時運行兩個agent，則agent名字需要改變，比如下面a2
a2.sources = r2
a2.sinks = k2
a2.channels = c2

# 2 source 
# 因監控linux本地文件，執行shell命令，所以type為exec；
a2.sources.r2.type = exec
# 監控的文件路徑
a2.sources.r2.command = tail -F /opt/test.log
a2.sources.r2.shell = /bin/bash -c

# 3 sink
# 數據下沉到目標源hdfs
a2.sinks.k2.type = hdfs
# 如果集群為HA模式，則路徑為active的namenode地址，普通分佈式集群，直接寫namenode所在地址即可。
a2.sinks.k2.hdfs.path = hdfs://bigdata111:9000/flume/%Y%m%d/%H
#上傳文件的前綴
a2.sinks.k2.hdfs.filePrefix = logs-
#是否按照時間滾動文件夾
a2.sinks.k2.hdfs.round = true
#多少時間單位創建一個新的文件夾
a2.sinks.k2.hdfs.roundValue = 1
#重新定義時間單位
a2.sinks.k2.hdfs.roundUnit = hour
#是否使用本地時間戳
a2.sinks.k2.hdfs.useLocalTimeStamp = true
#積攢多少個Event才flush到HDFS一次
a2.sinks.k2.hdfs.batchSize = 1000
#設置文件類型，可支持壓縮
a2.sinks.k2.hdfs.fileType = DataStream
#多久生成一個新的文件
a2.sinks.k2.hdfs.rollInterval = 600
#設置每個文件的滾動大小
a2.sinks.k2.hdfs.rollSize = 134217700
#文件的滾動與Event數量無關
a2.sinks.k2.hdfs.rollCount = 0
#最小副本數
a2.sinks.k2.hdfs.minBlockReplicas = 1

# 定義channel
a2.channels.c2.type = memory
a2.channels.c2.capacity = 1000
a2.channels.c2.transactionCapacity = 1000

# 雙向鏈接綁定
a2.sources.r2.channels = c2
a2.sinks.k2.channel = c2

5.2.2 啟動flume配置文件

[root@bigdata111 flume-1.8.0]# flume-ng agent --conf /opt/module/flume-1.8.0/conf/ --name a2 --conf-file /opt/module/flume-1.8.0/myconf/flume-hdfs.conf

5.2.3 發送文件內容

[root@bigdata111 opt]# echo kjalksdjglkajsdg2333333333333333asdgasdgasdg >> test.log
[root@bigdata111 opt]# echo kjalksdjglkajsdg2333333333333333asdgasdgasdg >> test.log
[root@bigdata111 opt]# echo kjalksdjglkajsdg2333333333333333asdgasdgasdg >> test.log
[root@bigdata111 opt]# echo kjalksdjglkajsdg2333333333333333asdgasdgasdg >> test.log

運行結果：

5.3 實時讀取目錄文件到HDFS

目標：使用flume監聽整個目錄的文件

5.3.1 創建Agent配置文件

創建agent配置文件，命名為：flume-dir.conf，文件內容如下：

#1 Agent
a3.sources = r3
a3.sinks = k3
a3.channels = c3

#2 source
#監控目錄的類型
a3.sources.r3.type = spooldir
#監控目錄的路徑
a3.sources.r3.spoolDir = /opt/module/flume1.8.0/upload
#哪個文件上傳hdfs，然後給這個文件添加一個後綴
a3.sources.r3.fileSuffix = .COMPLETED
a3.sources.r3.fileHeader = true
#忽略所有以.tmp結尾的文件，不上傳（可選）
a3.sources.r3.ignorePattern = ([^ ]*\.tmp)

# 3 sink
a3.sinks.k3.type = hdfs
a3.sinks.k3.hdfs.path = hdfs://bigdata111:9000/flume/%H
#上傳文件的前綴
a3.sinks.k3.hdfs.filePrefix = upload-
#是否按照時間滾動文件夾
a3.sinks.k3.hdfs.round = true
#多少時間單位創建一個新的文件夾
a3.sinks.k3.hdfs.roundValue = 1
#重新定義時間單位
a3.sinks.k3.hdfs.roundUnit = hour
#是否使用本地時間戳
a3.sinks.k3.hdfs.useLocalTimeStamp = true
#積攢多少個Event才flush到HDFS一次
a3.sinks.k3.hdfs.batchSize = 100
#設置文件類型，可支持壓縮
a3.sinks.k3.hdfs.fileType = DataStream
#多久生成一個新的文件
a3.sinks.k3.hdfs.rollInterval = 600
#設置每個文件的滾動大小大概是128M
a3.sinks.k3.hdfs.rollSize = 134217700
#文件的滾動與Event數量無關
a3.sinks.k3.hdfs.rollCount = 0
#最小副本數
a3.sinks.k3.hdfs.minBlockReplicas = 1

# Use a channel which buffers events in memory
a3.channels.c3.type = memory
a3.channels.c3.capacity = 1000
a3.channels.c3.transactionCapacity = 100

# Bind the source and sink to the channel
a3.sources.r3.channels = c3
a3.sinks.k3.channel = c3

溫馨提示：

1) 不要在監控目錄中創建並持續修改文件

2) 上傳完成的文件會以.COMPLETED結尾

3) 被監控文件夾每500毫秒掃描一次文件變動

5.3.2 啟動flume配置文件

[root@bigdata111 myconf]# flume-ng agent --conf /opt/module/flume-1.8.0/conf/ --name a3 --conf-file /opt/module/flume-1.8.0/myconf/flume-dir.conf

5.3.3 上傳文件到upload目錄

[root@bigdata111 opt]# mkdir upload
[root@bigdata111 opt]# ls
module  soft  test.log  upload
[root@bigdata111 opt]# mv test.log upload/
[root@bigdata111 opt]# ls
module  soft  upload
[root@bigdata111 opt]# vi test1.log
[root@bigdata111 opt]# mv test1.log upload/

運行如圖：

5.4 扇出例子01

扇出：數據用於多個地方。（簡單理解：一個數據源對應多個channel，sink,並且輸出到多個目標源）

例子01示意圖：

目標：在flume1裏面接收數據，然後數據下沉到兩個不同目標源（控制台和HDFS）

5.4.1 創建Agent配置文件

在myconf目錄下，新建一個flume-fanout1.conf文件，內容配置如下：

# 定義agent
a1.sources=c1
a1.channels=k1 k2
a1.sinks=s1 s2

# 定義source
a1.sources.c1.type=exec
a1.sources.c1.command=tail -F /opt/test.log
a1.sources.c1.shell=/bin/bash -c

# 將數據流複製給多個channel
a1.sources.r1.selector.type=replicating

# 定義channel1
a1.channels.k1.type=memory
a1.channels.k1.capacity = 1000
a1.channels.k1.transactionCapacity=1000

# 定義channel2
a1.channels.k2.type=memory
a1.channels.k2.capacity = 1000
a1.channels.k2.transactionCapacity=1000

# 定義sink1
a1.sinks.s1.type=logger

# 定義sink2
a1.sinks.s2.type=hdfs
a1.sinks.s2.hdfs.path = hdfs://bigdata111:9000/flume/%Y%m%d/%H
# 上傳文件的前綴
a1.sinks.s2.hdfs.filePrefix = logs-
# 是否按照時間滾動文件夾
a1.sinks.s2.hdfs.round = true
# 多少時間單位創建一個新的文件夾
a1.sinks.s2.hdfs.roundValue = 1
# 重新定義時間單位
a1.sinks.s2.hdfs.roundUnit = hour
# 是否使用本地時間戳
a1.sinks.s2.hdfs.useLocalTimeStamp = true
# 積攢多少個Event才flush到HDFS一次
a1.sinks.s2.hdfs.batchSize = 1000
# 設置文件類型，可支持壓縮
a1.sinks.s2.hdfs.fileType = DataStream
# 多久生成一個新的文件
a1.sinks.s2.hdfs.rollInterval = 600
# 設置每個文件的滾動大小
a1.sinks.s2.hdfs.rollSize = 134217700
# 文件的滾動與Event數量無關
a1.sinks.s2.hdfs.rollCount = 0
# 最小副本數
a1.sinks.s2.hdfs.minBlockReplicas = 1

# 雙向鏈接
a1.sources.c1.channels = k1 k2
a1.sinks.s1.channel=k1
a1.sinks.s2.channel=k2

5.4.2 啟動flume配置文件

[root@bigdata111 myconf]# flume-ng agent --conf /opt/module/flume-1.8.0/conf/ --name a1 --conf-file /opt/module/flume-1.8.0/myconf/flume-fanout1.conf -Dflume.root.logger==INFO,console

5.4.3 向文件添加內容

切換到/opt/目錄下，新建test.log文件，然後動態添加內容，觀察控制台輸出以及web的hdfs文件

[root@bigdata111 opt]# touch test.log
[root@bigdata111 opt]# touch test.log
[root@bigdata111 opt]# echo 'china' >>test.log 
[root@bigdata111 opt]# echo 'hello world' >>test.log
[root@bigdata111 opt]# echo 'nihao' >> test.log

控制台輸出如下：

web頁面結果：

5.5 扇出例子02

目標：flume1監控文件，然後將變動數據分別傳給flume2和flume3，flume2的數據下沉到HDFS；flume3的數據下沉到本地文件；

5.5.1 創建flume1配置文件

在bigdata111上的myconf目錄下，新建agent配置文件：flume-fanout1.conf;

flume1用於監控某文件的變動，同時產生兩個channel和兩個sink，分別輸送給flume2，flume3;

文件內容如下：

# 配置agent
a1.sources = c1
a1.channels = k1 k2
a1.sinks = s1 s2

# 定義source
a1.sources.c1.type=exec
a1.sources.c1.command=tail -F /opt/test.log
a1.sources.c1.shell=/bin/bash -c

# 將數據流複製給多個channel
a1.sources.c1.selector.type=replicating

# 定義channel1
a1.channels.k1.type=memory
a1.channels.k1.capacity = 1000
a1.channels.k1.transactionCapacity=1000

# 定義channel2
a1.channels.k2.type=memory
a1.channels.k2.capacity = 1000
a1.channels.k2.transactionCapacity=1000

# 定義sink1
a1.sinks.s1.type = avro
a1.sinks.s1.hostname = bigdata112
a1.sinks.s1.port = 4402

# 定義sink2
a1.sinks.s2.type = avro
a1.sinks.s2.hostname = bigdata113
a1.sinks.s2.port = 4402

# 雙向鏈接
a1.sources.c1.channels = k1 k2
a1.sinks.s1.channel=k1
a1.sinks.s2.channel=k2

5.5.2 創建flume2配置文件

在bigdata112的myconf目錄下，新建agent配置文件：flume-fanout2.conf

接收flume1的event數據，然後產生一個channel和一個sink，最後將數據下沉到hdfs

文件內容如下：

# 配置agent 不同agent之間，agent名不相同，但是source，channel,sink名可以相同
a2.sources = c2
a2.channels = k2
a2.sinks = s2

# 定義source
a2.sources.c2.type=avro
a2.sources.c2.bind = bigdata112
a2.sources.c2.port = 4402

# 定義channel
a2.channels.k2.type=memory
a2.channels.k2.capacity = 1000
a2.channels.k2.transactionCapacity=1000

# 定義sink
a2.sinks.s2.type = hdfs
a2.sinks.s2.hdfs.path=hdfs://bigdata111:9000/flume2/%H
#上傳文件的前綴
a2.sinks.s2.hdfs.filePrefix = flume2-
#是否按照時間滾動文件夾
a2.sinks.s2.hdfs.round = true
#多少時間單位創建一個新的文件夾
a2.sinks.s2.hdfs.roundValue = 1
#重新定義時間單位
a2.sinks.s2.hdfs.roundUnit = hour
#是否使用本地時間戳
a2.sinks.s2.hdfs.useLocalTimeStamp = true
#積攢多少個Event才flush到HDFS一次
a2.sinks.s2.hdfs.batchSize = 100
#設置文件類型，可支持壓縮
a2.sinks.s2.hdfs.fileType = DataStream
#多久生成一個新的文件
a2.sinks.s2.hdfs.rollInterval = 600
#設置每個文件的滾動大小大概是128M
a2.sinks.s2.hdfs.rollSize = 134217700
#文件的滾動與Event數量無關
a2.sinks.s2.hdfs.rollCount = 0
#最小副本數
a2.sinks.s2.hdfs.minBlockReplicas = 1

# 雙向鏈接
a2.sources.c2.channels = k2
a2.sinks.s2.channel=k2

5.5.3 創建flume3配置文件

在bigdata113的myconf目錄下，新建agent配置文件：flume-fanout3.conf

接收flume1的event數據，然後產生一個channel和一個sink，最後將數據下沉到本地/opt/flume3

文件內容如下：

# 配置agent
a3.sources = c3
a3.channels = k3
a3.sinks = s3

# 定義source
a3.sources.c3.type=avro
a3.sources.c3.bind = bigdata113
a3.sources.c3.port = 4402

# 定義channel
a3.channels.k3.type=memory
a3.channels.k3.capacity = 1000
a3.channels.k3.transactionCapacity=1000

# 定義sink
a3.sinks.s3.type = file_roll
# 提示：本地此目錄必須先建好，程序不會自動創建該目錄
a3.sinks.s3.sink.directory=/opt/flume3

# 雙向鏈接
a3.sources.c3.channels = k3
a3.sinks.s3.channel=k3

5.5.4 啟動三台機器配置文件

bigdata111:

[root@bigdata111 myconf]# flume-ng agent --conf /opt/module/flume-1.8.0/conf/ --name a1 --conf-file /opt/module/flume-1.8.0/myconf/flume-fanout1.conf -Dflume.root.logger==INFO,console

bigdata112:

[root@bigdata112 myconf]# flume-ng agent --conf /opt/module/flume-1.8.0/conf/ --name a2 --conf-file /opt/module/flume-1.8.0/myconf/flume-fanout2.conf

bigdata113:

[root@bigdata113 myconf]# flume-ng agent --conf /opt/module/flume-1.8.0/conf/ --name a3 --conf-file /opt/module/flume-1.8.0/myconf/flume-fanout3.conf

運行結果如圖：

bigdata112:

bigdata113:

5.6 扇入例子

5.6.1 創建flume1配置文件

flume1(agent1)監控端口數據變化，將數據sink到flume3(agent3);

在myconf目錄下新建agent文件：flume-fanin-1.conf

配置內容如下：

# 配置agent
a1.sources = c1
a1.channels = k1
a1.sinks = s1

# 配置source
a1.sources.c1.type = netcat
a1.sources.c1.bind = bigdata111
a1.sources.c1.port = 6666

# 配置sink
a1.sinks.s1.type=avro
a1.sinks.s1.hostname=bigdata113
a1.sinks.s1.port=5008

# 配置channel
a1.channels.k1.type=memory
a1.channels.k1.capacity=1000
a1.channels.k1.transactionCapacity=1000

# 雙向綁定
a1.sources.c1.channels = k1
a1.sinks.s1.channel = k1

5.6.2 創建flume2配置文件

flume2(agent2)監控本地文件變化，將數據sink到flume3(agent3);

在myconf目錄下新建agent文件：flume-fanin-2.conf

配置內容如下：

# 配置agent
a2.sources = c1
a2.channels = k1
a2.sinks = s1

# 配置source
a2.sources.c1.type = exec
a2.sources.c1.command = tail -F /opt/ceshi.log
a2.sources.c1.shell=/bin/bash -c

# 配置sink
a2.sinks.s1.type=avro
a2.sinks.s1.hostname=bigdata113
a2.sinks.s1.port=5008

# 配置channel
a2.channels.k1.type=memory
a2.channels.k1.capacity=1000
a2.channels.k1.transactionCapacity=1000

# 雙向綁定
a2.sources.c1.channels = k1
a2.sinks.s1.channel = k1

5.6.3 創建flume3配置文件

flume3(agent3)接收flume1和flume2的數據，將數據sink到HDFS ；

在myconf目錄下新建agent文件：flume-fanin-3.conf

配置內容如下：

# 配置agent
a3.sources = c1
a3.channels = k1
a3.sinks = s1

# 配置source
a3.sources.c1.type = avro
a3.sources.c1.bind = bigdata113
a3.sources.c1.port = 5008

# 配置sink
a3.sinks.s1.type=hdfs
a3.sinks.s1.hdfs.path=hdfs://bigdata111:9000/flume3/%H
# 上傳文件的前綴
a3.sinks.s1.hdfs.filePrefix = flume3-
# 是否按照時間滾動文件夾
a3.sinks.s1.hdfs.round = true
# 多少時間單位創建一個新的文件夾
a3.sinks.s1.hdfs.roundValue = 1
# 重新定義時間單位
a3.sinks.s1.hdfs.roundUnit = hour
# 是否使用本地時間戳
a3.sinks.s1.hdfs.useLocalTimeStamp = true
# 積攢多少個Event才flush到HDFS一次
a3.sinks.s1.hdfs.batchSize = 1000
# 設置文件類型，可支持壓縮
a3.sinks.s1.hdfs.fileType = DataStream
# 多久生成一個新的文件
a3.sinks.s1.hdfs.rollInterval = 600
# 設置每個文件的滾動大小大概是128M
a3.sinks.s1.hdfs.rollSize = 134217700
# 文件的滾動與Event數量無關
a3.sinks.s1.hdfs.rollCount = 0
# 最小冗餘數
a3.sinks.s1.hdfs.minBlockReplicas = 1

# 配置channel
a3.channels.k1.type=memory
a3.channels.k1.capacity=1000
a3.channels.k1.transactionCapacity=1000

# 雙向綁定
a3.sources.c1.channels = k1
a3.sinks.s1.channel = k1

5.6.4 啟動三個flume配置文件

flume1:

[root@bigdata111 myconf]# flume-ng agent -c ../conf/ -n a1 -f flume-fanout1.conf -Dflume.root.logger==INFO,console

flume2:

[root@bigdata112 myconf]# flume-ng agent -c ../conf/ -n a2 -f flume-fanout2.conf

flume3:

[root@bigdata113 myconf]# flume-ng agent -c ../conf/ -n a3 -f flume-fanout3.conf

5.6.5 操作端口與文件

新開xshell選項卡，鏈接bigdata111服務器，然後執行telnet命令：

[root@bigdata111 ~]# telnet bigdata111 6666
Trying 192.168.1.111...
Connected to bigdata111.
Escape character is '^]'.
english
OK
chinese
OK
hello
OK
.net
OK
php
OK
java
OK

新開xshell選項卡，鏈接bigdata112服務器，然後向/opt/ceshi.log添加新內容：

[root@bigdata112 ~]# cd /opt/
[root@bigdata112 opt]# ls
ceshi.log  ha  module  soft  zookeeper.out
[root@bigdata112 opt]# cat ceshi.log
start-log-in
end-log
[root@bigdata112 opt]# echo `date` >> ceshi.log
[root@bigdata112 opt]# echo "end-log" >> ceshi.log
[root@bigdata112 opt]# cat ceshi.log 
start-log-in
end-log
2019年 09月 07日 星期六 23:36:03 CST
end-log

5.6.6 显示運行結果

web頁面結果：

hdfs的文件內容：

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

※評比前十大台北網頁設計、台北網站設計公司知名案例作品心得分享

※智慧手機時代的來臨，RWD網頁設計已成為網頁設計推薦首選