6次优化,数据同步效率提升72倍

//192.168.10.20 on /mnt/source/test type cifs (rw,relatime,cache=struct,uid=0,noforceuid,gid=0,noforcegid,addr=192.168.10.20,file_mode=0755,dir_mode=0755,soft,nounix,serverino,mapposix,rsize=1048576,wsize=1048576,echo_interval=60,actimeo=1)

func Consumer(cfgInfo map[string]string) {
  // 创建RabbitMQ连接
  conn, connErr := tools.RabbitMQConn()
  // 关闭RabbitMQ连接
  defer conn.Close()

  // 获取并发数
  concurrentNum, _ := strconv.Atoi(cfgInfo["concurrent"])

  for qname, v := range queues {
    mainWg.Add(1)
    go func(queueName string) {
      defer mainWg.Done()
      // 创建通道
      chann, channErr := tools.CreateChann(conn)
      chann.Qos(concurrentNum, 0, false)

      // 关闭通道
      defer chann.Close()
      // 创建队列
      que, queueErr := tools.CreateQueue(chann, queueName)

      // 定义一个消费者
      msgs, comErr := chann.Consume(
        que.Name, // queue
        "",       // consumer
        false,    // auto-ack
        false,    // exclusive
        false,    // no-local
        false,    // no-wait
        nil,      // args
      )

      go func() {
        // 并发消费文件
        for d := range msgs {

          msg := string(d.Body)
          localfile := v["sourcePath"] + msg
          alyunfile := v["targetPath"] + msg + "/"
                    ossutilErr := PushDir(localfile, alyunfile)
          if ossutilErr == nil {
            consumerLogger.Info("Successfully ossutil sync file: " + msg)
            consumerLogger.Info(string(out))
          } else {
            consumerLogger.Fatal("Failed ossutil sync file!")
          }
          d.Ack(false)
        }
      }()
      // 等待
      select {}
    }(qname)
  }
  mainWg.Wait()
}

for d := range msgs {
  mainWg.Add(1)
  go func(d amqp.Delivery) {
    defer mainWg.Done()
    msg := string(d.Body)
    localfile := v["sourcePath"] + msg
    alyunfile := v["targetPath"] + msg + "/"
        ossutilErr := PushDir(localfile, alyunfile)

    if ossutilErr == nil {
      consumerLogger.Infof("Successfully ossutil sync dir: %s\n", msg)
    } else {
      consumerLogger.Fatalf("Failed ossutil sync dir %s, ERROR: %s\n", msg, ossutilErr)
    }
    if err := d.Ack(false); err != nil {
      consumerLogger.Fatalf("Failed rabbitmq ack! ERROR:%v\n", err)
    } else {
      consumerLogger.Info("Successfully rabbitmq Ack message: " + msg)
    }
  }(d)
}

./gosync -r producer -t 20240108  
panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x1 pc=0x5fcba1]

goroutine 2403 [running]:
gopkg.in/ini%2ev1.(*File).SectionsByName(0x0, {0x0?, 0xc000020d00?})
        /Users/edy/go/pkg/mod/gopkg.in/ini.v1@v1.67.0/file.go:156 +0x61
gopkg.in/ini%2ev1.(*File).GetSection(...)
        /Users/edy/go/pkg/mod/gopkg.in/ini.v1@v1.67.0/file.go:137
gopkg.in/ini%2ev1.(*File).Section(0xc00270cd20?, {0x0, 0x0})
        /Users/edy/go/pkg/mod/gopkg.in/ini.v1@v1.67.0/file.go:175 +0x26
gosync/rabbitmq.recursiveDir.func1({0x6ec978, 0xc00018ad40}, 0x0?)
        /Users/edy/gosync/rabbitmq/producer.go:90 +0x234
created by gosync/rabbitmq.recursiveDir in goroutine 6
        /Users/edy/gosync/rabbitmq/producer.go:80 +0x28e

for _, dir := range dirs {
    files, getfileErr := ioutil.ReadDir(classsdir + dir.Name())
    if getfileErr == nil {
      producerLogger.Info("Obtaining a File success!")
    } else {
      producerLogger.Fatal("Obtaining a File failed!")
    }
    // d := strings.TrimLeft(classsdir+dir.Name(), sourcePath)
    d := (classsdir + dir.Name())[len(sourcePath):]

    // 判断是否同步目录: 不存在、有增量则进行同步
    if cfg.Section("").HasKey(d) {
      val := cfg.Section("").Key(d).String()
      nums, _ := strconv.Atoi(val)

      if nums < len(files) {
        cfg.Section("").Key(d).SetValue(strconv.Itoa(len(files)))
        cfg.SaveTo(syncfile)
        producerFiles(producerLogger, d, ch, q, bucket)
      }
    } else {
      cfg.Section("").Key(d).SetValue("0")
      cfg.SaveTo(syncfile)
      producerFiles(producerLogger, d, ch, q, bucket)
    }
  }

  for _, dir := range dirs {
    wg.Add(1)
    go func(dir os.DirEntry) {
      defer wg.Done()
      localDir := classsdir + dir.Name()
      files, getfileErr := os.ReadDir(localDir)
      if getfileErr != nil {
        producerLogger.Fatalf("Obtaining a File failed! dirname: %s, ERROR: %s\n", dir.Name(), getfileErr)
      }

      localFileSum := len(files)
      // 去掉前面的统一的本地路径，保留只上云的路径，把此路径推送到MQ队列
      d := (classsdir + dir.Name())[len(sourcePath):]

      // 判断是否同步目录: 不存在、有增量则进行同步
      exists, err := cli.Exists(ctx, d).Result()
      if err != nil {
        producerLogger.Fatalf("redis: Exists failed! ERROR: %s", err)
      }

      // 判断redis 中key是否存在，返回值为0则不存在，不存在立即把目录发送到MQ，并设置key
      if exists == 0 {
        err := ProducerFiles(producerLogger, d, ch, q)
        if err != nil {
          producerLogger.Fatalf("rabbitMQ: send key failed! ERROR:%s", err)
          return
        }
        err = cli.Set(ctx, d, localFileSum, 0).Err()
        if err != nil {
          producerLogger.Fatalf("redis: set key failed! ERROR:%s", err)
          return
        }
      }

      // 查询redis 中key 的值，跟本地目录下的文件数做对比，如果本地大则发布MQ消息
      val, err := cli.Get(ctx, d).Result()
      if err != nil {
        producerLogger.Fatalf("redis: get key failed! ERROR:%s", err)
      }
      num, err := strconv.Atoi(val)
      if err != nil {
        producerLogger.Fatalf("redis: key to int failed! ERROR:%s", err)
      }
      if localFileSum > num {
        err := ProducerFiles(producerLogger, d, ch, q)
        if err != nil {
          producerLogger.Fatalf("rabbitMQ: send key failed! ERROR:%s", err)
          return
        }
        err = cli.Set(ctx, d, localFileSum, 0).Err()
        if err != nil {
          producerLogger.Fatalf("redis: set key failed! ERROR:%s", err)
        }
      }
    }(dir)
  }