自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

使用Go語(yǔ)言，25秒讀取16GB文件

作者：佚名
開(kāi)發(fā) 后端
我們將使用Go語(yǔ)言，從一個(gè)大小為16GB的.txt或.log文件中提取日志。

 當(dāng)今世界的任何計(jì)算機(jī)系統(tǒng)每天都會(huì)生成大量的日志或數(shù)據(jù)。隨著系統(tǒng)的發(fā)展，將調(diào)試數(shù)據(jù)存儲(chǔ)到數(shù)據(jù)庫(kù)中是不可行的，因?yàn)樗鼈兪遣豢勺兊?，并且只能用于分析和解決故障。所以大部分公司傾向于將日志存儲(chǔ)在文件中，而這些文件通常位于本地磁盤中。

[[410141]]

我們將使用Go語(yǔ)言，從一個(gè)大小為16GB的.txt或.log文件中提取日志。

讓我們開(kāi)始編碼……

首先，我們打開(kāi)文件。對(duì)于任何文件的IO，我們都將使用標(biāo)準(zhǔn)的Go os.File。 

  1. f, err := os.Open(fileName)  
  2.  if err != nil {  
  3.    fmt.Println("cannot able to read the file", err)  
  4.    return  
  5.  }  
  6. // UPDATE: close after checking error  
  7. defer file.Close()  //Do not forget to close the file 

打開(kāi)文件后，我們有以下兩個(gè)選項(xiàng)可以選擇：

逐行讀取文件，這有助于減少內(nèi)存緊張，但需要更多的時(shí)間。一次將整個(gè)文件讀入內(nèi)存并處理該文件，這將消耗更多內(nèi)存，但會(huì)顯著減少時(shí)間。

由于文件太大，即16 GB，因此無(wú)法將整個(gè)文件加載到內(nèi)存中。但是第一種選擇對(duì)我們來(lái)說(shuō)也是不可行的，因?yàn)槲覀兿Ｍ趲酌腌妰?nèi)處理文件。

但你猜怎么著，還有第三種選擇。瞧……相比于將整個(gè)文件加載到內(nèi)存中，在Go語(yǔ)言中，我們還可以使用bufio.NewReader()將文件分塊加載。 

  1. r := bufio.NewReader(f)  
  2. for {  
  3. buf := make([]byte,4*1024) //the chunk size  
  4. n, err := r.Read(buf) //loading chunk into buffer  
  5.    bufbuf = buf[:n]  
  6. if n == 0 { 
  7.       if err != nil {  
  8.        fmt.Println(err)  
  9.        break  
  10.      }  
  11.      if err == io.EOF {  
  12.        break  
  13.      }  
  14.      return err  
  15.   }  
  16. } 

一旦我們將文件分塊，我們就可以分叉一個(gè)線程，即Go routine，同時(shí)處理多個(gè)文件區(qū)塊。上述代碼將修改為： 

  1. //sync pools to reuse the memory and decrease the preassure on Garbage Collector  
  2. linesPool := sync.Pool{New: func() interface{} {  
  3.         lines := make([]byte, 500*1024)  
  4.         return lines  
  5. }}  
  6. stringPool := sync.Pool{New: func() interface{} {  
  7.           lines := ""  
  8.           return lines  
  9. }}  
  10. slicePool := sync.Pool{New: func() interface{} {  
  11.            lines := make([]string, 100)  
  12.            return lines  
  13. }}  
  14. r := bufio.NewReader(f)  
  15. var wg sync.WaitGroup //wait group to keep track off all threads  
  16. for {   
  17.       buf := linesPool.Get().([]byte)  
  18.      n, err := r.Read(buf)  
  19.      bufbuf = buf[:n]  
  20. if n == 0 {  
  21.         if err != nil {  
  22.             fmt.Println(err)  
  23.             break  
  24.         }  
  25.         if err == io.EOF {  
  26.             break  
  27.         }  
  28.         return err  
  29.      }  
  30. nextUntillNewline, err := r.ReadBytes('\n')//read entire line     
  31.       if err != io.EOF { 
  32.          buf = append(buf, nextUntillNewline...)  
  33.      }     
  34.       wg.Add(1)  
  35.      go func() {       
  36.          //process each chunk concurrently  
  37.         //start -> log start time, end -> log end time        
  38.          ProcessChunk(buf, &linesPool, &stringPool, &slicePool,     start, end)  
  39. wg.Done()    
  40.       }()  
  41. }  
  42. wg.Wait()  
  43. } 

上面的代碼，引入了兩個(gè)優(yōu)化點(diǎn)：

sync.Pool是一個(gè)強(qiáng)大的對(duì)象池，可以重用對(duì)象來(lái)減輕垃圾收集器的壓力。我們將重用各個(gè)分片的內(nèi)存，以減少內(nèi)存消耗，大大加快我們的工作。Go Routines幫助我們同時(shí)處理緩沖區(qū)塊，這大大提高了處理速度。

現(xiàn)在讓我們實(shí)現(xiàn)ProcessChunk函數(shù)，它將處理以下格式的日志行。 

  1. 2020-01-31T20:12:38.1234Z, Some Field, Other Field, And so on, Till new line,...\n 

我們將根據(jù)命令行提供的時(shí)間戳提取日志。 

  1. func ProcessChunk(chunk []byte, linesPool *sync.Pool, stringPool *sync.Pool, slicePool *sync.Pool, start time.Time, end time.Time) {  
  2. //another wait group to process every chunk further                            
  3.        var wg2 sync.WaitGroup  
  4. logs := stringPool.Get().(string)  
  5. logs = string(chunk)  
  6. linesPool.Put(chunk) //put back the chunk in pool  
  7. //split the string by "\n", so that we have slice of logs  
  8.       logsSlice := strings.Split(logs, "\n")  
  9. stringPool.Put(logs) //put back the string pool  
  10. chunkSize := 100 //process the bunch of 100 logs in thread  
  11. n := len(logsSlice)  
  12. noOfThread := n / chunkSize  
  13. if n%chunkSize != 0 { //check for overflow   
  14.          noOfThread++  
  15.       }  
  16. length := len(logsSlice)  
  17. //traverse the chunk  
  18.      for i := 0; i < length; i += chunkSize {         
  19.           wg2.Add(1) 
  20. //process each chunk in saperate chunk  
  21.          go func(s int, e int) {  
  22.             for i:= s; i<e;i++{  
  23.                text := logsSlice[i]  
  24. if len(text) == 0 {  
  25.                   continue  
  26.                }            
  27.              logParts := strings.SplitN(text, ",", 2)  
  28.             logCreationTimeString := logParts[0]  
  29.             logCreationTime, err := time.Parse("2006-01-  02T15:04:05.0000Z", logCreationTimeString)  
  30. if err != nil {  
  31.                  fmt.Printf("\n Could not able to parse the time :%s       for log : %v", logCreationTimeString, text) 
  32.                   return  
  33.             }  
  34. // check if log's timestamp is inbetween our desired period  
  35.           if logCreationTime.After(start) && logCreationTime.Before(end) { 
  36.             fmt.Println(text)  
  37.            }  
  38.         }  
  39.         textSlice = nil 
  40.          wg2.Done()   
  41.       }(i*chunkSize, int(math.Min(float64((i+1)*chunkSize), float64(len(logsSlice)))))  
  42.    //passing the indexes for processing  
  43. }    
  44.    wg2.Wait() //wait for a chunk to finish  
  45.    logsSlice = nil  
  46. } 

對(duì)上面的代碼進(jìn)行基準(zhǔn)測(cè)試。以16 GB的日志文件為例，提取日志所需的時(shí)間約為25秒。

完整的代碼示例如下： 

  1. func main() {  
  2.  s := time.Now()  
  3.  args := os.Args[1:]  
  4.  if len(args) != 6 { // for format  LogExtractor.exe -f "From Time" -t "To Time" -i "Log file directory location"  
  5.   fmt.Println("Please give proper command line arguments")  
  6.   return  
  7.  }  
  8.  startTimeArg := args[1]  
  9.  finishTimeArg := args[3]  
  10.  fileName := args[5]  
  11.  file, err := os.Open(fileName)  
  12.  if err != nil {  
  13.   fmt.Println("cannot able to read the file", err)  
  14.   return  
  15.  }  
  16.  defer file.Close() //close after checking err  
  17.  queryStartTime, err := time.Parse("2006-01-02T15:04:05.0000Z", startTimeArg)  
  18.  if err != nil { 
  19.    fmt.Println("Could not able to parse the start time", startTimeArg)  
  20.   return  
  21.  }  
  22.  queryFinishTime, err := time.Parse("2006-01-02T15:04:05.0000Z", finishTimeArg)  
  23.  if err != nil {  
  24.   fmt.Println("Could not able to parse the finish time", finishTimeArg)  
  25.   return  
  26.  }  
  27.  filestat, err := file.Stat()  
  28.  if err != nil {  
  29.   fmt.Println("Could not able to get the file stat")  
  30.   return  
  31.  } 
  32.  fileSize := filestat.Size()  
  33.  offset := fileSize - 1  
  34.  lastLineSize := 0  
  35.  for {  
  36.   b := make([]byte, 1)  
  37.   n, err := file.ReadAt(b, offset)  
  38.   if err != nil {  
  39.    fmt.Println("Error reading file ", err)  
  40.    break  
  41.   }  
  42.   char := string(b[0])  
  43.   if char == "\n" { 
  44.     break  
  45.   }  
  46.   offset--  
  47.   lastLineSize += n  
  48.  }  
  49.  lastLine := make([]byte, lastLineSize)  
  50.  _, err = file.ReadAt(lastLine, offset+1) 
  51.  if err != nil {  
  52.   fmt.Println("Could not able to read last line with offset", offset, "and lastline size", lastLineSize)  
  53.   return  
  54.  }  
  55.  logSlice := strings.SplitN(string(lastLine), ",", 2)  
  56.  logCreationTimeString := logSlice[0]  
  57.  lastLogCreationTime, err := time.Parse("2006-01-02T15:04:05.0000Z", logCreationTimeString)  
  58.  if err != nil {  
  59.   fmt.Println("can not able to parse time : ", err)  
  60.  }  
  61.  if lastLogCreationTime.After(queryStartTime) && lastLogCreationTime.Before(queryFinishTime) {  
  62.   Process(file, queryStartTime, queryFinishTime) 
  63.   }  
  64.  fmt.Println("\nTime taken - ", time.Since(s))  
  65. } 
  66. func Process(f *os.File, start time.Time, end time.Time) error {  
  67.  linesPool := sync.Pool{New: func() interface{} {  
  68.   lines := make([]byte, 250*1024)  
  69.   return lines  
  70.  }} 
  71.  stringPool := sync.Pool{New: func() interface{} {  
  72.   lines := ""  
  73.   return lines  
  74.  }}  
  75.  r := bufio.NewReader(f)  
  76.  var wg sync.WaitGroup  
  77.  for {  
  78.   buf := linesPool.Get().([]byte)  
  79.   n, err := r.Read(buf)  
  80.   bufbuf = buf[:n]   
  81.   if n == 0 { 
  82.     if err != nil {  
  83.     fmt.Println(err)  
  84.     break  
  85.    }  
  86.    if err == io.EOF {  
  87.     break  
  88.    }  
  89.    return err  
  90.   }  
  91.   nextUntillNewline, err := r.ReadBytes('\n')  
  92.   if err != io.EOF {  
  93.    buf = append(buf, nextUntillNewline...)  
  94.   }  
  95.   wg.Add(1)  
  96.   go func() {  
  97.    ProcessChunk(buf, &linesPool, &stringPool, start, end)  
  98.    wg.Done()  
  99.   }()  
  100.  }  
  101.  wg.Wait()  
  102.  return nil  
  103. }   
  104. func ProcessChunk(chunk []byte, linesPool *sync.Pool, stringPool *sync.Pool, start time.Time, end time.Time) { 
  105.   var wg2 sync.WaitGroup  
  106.  logs := stringPool.Get().(string)  
  107.  logs = string(chunk)  
  108.  linesPool.Put(chunk)  
  109.  logsSlice := strings.Split(logs, "\n")  
  110.  stringPool.Put(logs) 
  111.  chunkSize := 300  
  112.  n := len(logsSlice)  
  113.  noOfThread := n / chunkSize  
  114.  if n%chunkSize != 0 {  
  115.   noOfThread++ 
  116.  }  
  117.  for i := 0; i < (noOfThread); i++ {  
  118.   wg2.Add(1)  
  119.   go func(s int, e int) {  
  120.    defer wg2.Done() //to avaoid deadlocks  
  121.    for i := s; i < e; i++ {  
  122.     text := logsSlice[i]  
  123.     if len(text) == 0 {  
  124.      continue  
  125.     }  
  126.     logSlice := strings.SplitN(text, ",", 2)  
  127.     logCreationTimeString := logSlice[0]  
  128.     logCreationTime, err := time.Parse("2006-01-02T15:04:05.0000Z", logCreationTimeString)  
  129.     if err != nil {  
  130.      fmt.Printf("\n Could not able to parse the time :%s for log : %v", logCreationTimeString, text)  
  131.      return  
  132.     }  
  133.     if logCreationTime.After(start) && logCreationTime.Before(end) {  
  134.      //fmt.Println(text)  
  135.     }  
  136.    }  
  137.   }(i*chunkSize, int(math.Min(float64((i+1)*chunkSize), float64(len(logsSlice)))))  
  138.  }  
  139.  wg2.Wait()  
  140.  logsSlice = nil  
  141. }  

 

責(zé)任編輯：龐桂玉 來(lái)源： 運(yùn)維派
Go語(yǔ)言16GB文件代碼
相關(guān)推薦

2021-06-28 11:15:22

Go語(yǔ)言16GB文件

2023-06-12 07:20:39

2011-07-04 10:06:49

webOS 3.0TouchPad

2015-07-13 11:20:01

iPhone內(nèi)存蘋果

2012-01-12 12:56:16

思科博科16Gb

2023-05-22 07:06:29

RTX 40608GB公版卡

2023-07-04 07:08:27

內(nèi)存游戲容量

2014-04-11 14:35:16

16GB顯卡AMD

2023-07-24 08:14:04

8GB版本RTX

2019-11-06 15:16:12

16GB8GB內(nèi)存

2012-06-04 14:13:48

盛大手機(jī)

2023-05-17 07:28:18

電腦16GB內(nèi)存

2020-04-13 19:13:23

手機(jī)內(nèi)存 PC

2013-07-10 14:41:46

內(nèi)存

2019-05-28 15:32:56

筆記本Max-Q電腦

2012-07-04 10:00:04

16GB DDR4服務(wù)器內(nèi)存

2019-07-01 10:30:27

存儲(chǔ)技術(shù)容器

2011-07-11 14:31:45

臺(tái)式機(jī)評(píng)測(cè)

2023-05-19 08:01:57

Go 語(yǔ)言map

2021-08-02 15:02:37

Go Excelize 開(kāi)發(fā)
點(diǎn)贊
收藏

51CTO技術(shù)棧公眾號(hào)