go中的關鍵字-defer

發佈日期: 2020-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網頁設計已成為網頁設計推薦首選

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

發佈日期: 2020-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-11 作者: admin

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

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

【其他文章推薦】

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

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

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

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

發佈日期: 2020-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網頁設計已成為網頁設計推薦首選

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

發佈日期: 2020-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網頁設計已成為網頁設計推薦首選

泰國最大能源公司與六家車企簽訂協定進軍電動汽車領域

發佈日期: 2020-03-06 作者: admin

據報導，泰國最大的能源公司“泰國國家石油公司”近日與六家主要汽車製造商簽訂協定，將聯合開發電動汽車項目，並配合該專案建成第一批電動車輛充電站。
雙方聯合發表的聲明稱，根據協定，國營的泰國國家石油公司將負責建造由20個充電站組成的網狀系統，汽車製造商承諾研發電動車輛，並提升電車在泰國市場上的知名度和可信度。據悉，這六家企業分別是寶馬集團、賓士汽車、三菱汽車、日產汽車、保時捷汽車和沃爾沃集團在泰國的分部。據悉，泰國國家石油公司自2012年起開始研發電動車輛技術，目前已經運行4個電動車輛充電站。到目前為止，影響電動車輛生產和銷售的一大因素是充電站數量的不足。為刺激這一產業的發展，泰國政府已向製造電動車輛部件的生產商發放稅收鼓勵，包括發動機和電池生產商。文章來源：環球網

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

【其他文章推薦】

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

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

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

中國：「十三五」期間將力推新能源汽車技術創新

發佈日期: 2020-03-05 作者: admin

近日，以「安全保障、創新引領」為主題的中國電動汽車百人會夏季論壇召開。科技部副部長陰和俊透露，科技部在「十三五」期間將大力推進新能源汽車技術創新發展，目標是到2020年建立完善的電動汽車動力系統科技體系和產業鏈，為我國新能源汽車產業發展提供更有力的科技支撐。
作為低碳化、資訊化和智慧化的最佳平臺，新能源汽車正成為新一輪科技革命的重要載體。2016年1—6月，全中國新能源汽車累計產銷量均超過17萬輛，較2015年同期大幅提高。然而，在新能源汽車示範推廣應用過程中，暴露出充電基礎設施建設不足、車輛安全隱患增加等問題。陰和俊介紹，從2015年下半年起，科技部聯合財政部、工信部等組織實施「十三五」國家重點研發計畫新能源汽車試點專項，從基礎科學問題、共性核心關鍵、動力系統技術、集成開發與示範四個層次，重點對動力電池與電池的管理系統、電機驅動與電力電池總成、電動汽車智能化、燃料電池動力系統、插電增程式混合動力系統、純電動系統六個方向進行研發部署，以完善中國新能源汽車研發體系。文章來源：科技日報

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

【其他文章推薦】

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

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

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

村田製作所開發電動汽車用絕緣型DC-DC轉換器

發佈日期: 2020-03-04 作者: admin

村田製作所開發出了用於純電動汽車、混合動力車、插電式混合動力車等的絕緣型DC-DC轉換器模“MYIS系列”，將於2016年12月開始量產。該產品適用於採用雙電池系統的EV、HEV及PHEV。最大輸出功率為3W。輸入電壓範圍為＋6～16V。外形尺寸為16mm×27mm×8mm，支持表面封裝。符合車載電子部件品質標準“AEC-Q”。
電源電路拓撲為反激式。配備有反激式轉換器控制IC、電壓檢測IC、功率MOSFET、絕緣變壓器及輸入輸出電容器等。絕緣耐壓為1800VAC。備有輸出電壓不同的4款產品，分別是輸出電壓為＋3.3V的“MYISA3R308PSPQ”、輸出電壓為＋5.0V的“MYISA005R6PSPQ”、輸出電壓為＋12V的“MYISA012R3PSPQ”以及輸出電壓為＋15V的“MYISA015R2PSPQ”。最大輸出電流方面，＋3.3V產品為0.8A，＋5.0V產品為0.6A，＋12V產品為0.25A，＋15V產品為0.2A。輸出電壓的誤差為±5％。開關頻率為300kHz。轉換效率方面，＋3.3V產品為70％，其他三款產品為74％。工作溫度範圍在-40～＋105℃。樣品價格為1000日元。文章來源：日經技術線上mp

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

【其他文章推薦】

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

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

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

2020 年電動車市場表現不一，中國預期疲弱、歐洲強勁

發佈日期: 2020-03-03 作者: admin

普氏能源資訊 7 日報導，標普全球普氏分析（S&P Global Platts Analytics）最新報告，2020 年全球電動車市場可能表現不一，中國市場可能仍在掙扎，歐洲市場則可望迎來動能。電動車市場認為是未來銅需求的一大增長市場，因電動車的銅用量可達傳統汽車 4 倍以上。去年中國取消部分電動車補貼，導致電動車銷售降溫，也令用於汽油車觸媒轉化器的鈀金價格受提振，全年寫下六成漲幅，因純電動車不需要觸媒轉化器。7 日紐約鈀金期貨收盤首度站上每盎司 2,000 美元。

報告指出，最新數據顯示，去年 11 月中國電動車銷售年減 41%，2020 年也將有進一步下滑的風險。中國汽車工業協會數據顯示，11 月新能源汽車產銷中，純電動車產銷分別為 9.6 萬輛和 8.1 萬輛，分別年減 29.6% 和 41.2%。協會預計，2020 年中國汽車市場全年銷量預估達 2,531 萬輛，相較 2019 年預估值下滑 2%，直到 2021 年有可能還是負增長或基本持平；整體而言，預計中國車市約 2022 年才會恢復增長，到 2023~2025 年可能會有年均 4% 左右增長。

另一方面，歐洲市場的電動車銷售較強勁，11 月電動車銷售年增 63%，1~11 月電動車銷售也年增 47%。報告認為，2020 年歐洲電動車市場仍將維持強勁增長，主要因為各國的排放法規趨嚴影響。英國汽車生產與貿易商協會 6 日公布數據顯示，2019 年，英國純電動車掛牌數量創新高，達 37,850 輛，較前一年度大增 144%。不過，去年電動車仍僅占國車市 1.6% 份額，相比英國政府計劃未來 10 年將電動車市占提高至 50%~70%，還有很遙遠的距離。

為了應對全球暖化，各國都有更嚴格的汽車排放標準，這為汽車市場向電動車過渡提供有力支撐。歐盟 2018 年底通過新目標，將在 2050 年達成「氣候中和」的經濟體，意即溫室氣體零排放，比原先所定的 2050 年溫室氣體排放較 1990 年減少 80% 更激進。為了達成此目標，歐盟將更廣泛使用可再生能源、電動交通，甚至「負碳排放」工業。歐盟執委會也將重新審視歐盟的法規，必要時將再更新以期達成此目標。

全球最大銅產國智利央行 7 日公布數據，2019 年 12 月，智利出口額年增 1.3% 至 66.4 億美元，其中銅出口額年減 2.7% 至 34.5 億美元。2019 年，智利的銅出口額年減 8.2% 至 334 億美元。智利去年 10 月爆發全國性大規模抗議事件，最終造成 27 人死亡，因此 12 月整體出口額出現增長仍優於市場預期。

（本文內容由授權使用。首圖來源：）

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

【其他文章推薦】

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

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

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

IEA: 2016全球電動車銷售200萬輛，成長60%

發佈日期: 2020-03-02 作者: admin

國際能源署（IEA）數據顯示，電動車市場成長十分快速，2016 年全球累計銷量約為200 萬輛，較前年比大幅增長了60%，儘管從整體汽車產業來看電動車仍僅占有極小比例，但IEA 預估至2020 年時，全球累計銷售量將達到900~2,000 萬輛。

在所有市場中，又以中國的成長幅度最為明顯，在2016 年累計銷量中，中國就佔了65 萬台，超過美國的56 萬輛，更佔了全球市場的32%，IEA 認為中國能達到這樣的成長，很大程度是受到環境政策的成功推動。

印度在上週宣布全國電動車計畫，計畫至2030 年時只販售電動車，主要原因就是對於空氣汙染的考量，IEA 指出，中國及歐美許多國家在內也正推動類似措施，除了一般汽車以外，也計畫將巴士、貨車、卡車的電動車比例提升至30%。

電動車在近年能有這樣的大幅成長，主要得感謝充電設施的逐漸普及，與消費者興趣的提升。由於汽車排廢醜聞不斷，刺激了針對電動車的投資，再加上特斯拉推出Model 系列車款後，將電動車變成新潮的象徵，將電動車帶往流行，逐漸得到消費者認可。

BP 經濟學家Spencer Dale 估計，至2035 年時，「新潮」因素將有望帶動電動車銷售額達到4.5 億美元。

為了達到「巴黎協定」對於氣候變化定義的目標——將全球暖化導致的溫度上升幅度限制在攝氏2 度以內，IEA 估計，至2040 年時，全球將需要6 億部電動汽車。

（合作媒體：。圖片出處：public domain CC0）

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

【其他文章推薦】

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

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

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