p2mocv3pr/handler/ProducerAndComsumer.go

package handler

import (
	"bytes"
	"fmt"
	"objectss/database"
	"objectss/utils"
	"os/exec"
	"sync"
)

var mu sync.Mutex
var ComsuNum int

// 生产者产生 shell的命令，放入管道中等待生产者执行
func Productor(channel chan string, dbw *database.DbWorker) {
	defer close(channel)
	for i := 0; i < len(dbw.Repositories); i++ {
		channel <- fmt.Sprintf("%s", dbw.Repositories[i].Path.String)
	}
}

// 消费者负责执行shell
func Consumer(channel chan string, dbw *database.DbWorker, wg *sync.WaitGroup, osslink string, obsCommand string) {
	for {
		path, ok := <-channel // 此处会阻塞, 如果信道中没有数据的话
		if ok {
			//取出命令执行
			if len(path) != 0 {
				//pathDir := strings.Replace(path, ".git", "", -1)
				//utils.Log.Info("path:", pathDir, "ComsumerNume:", ComsuNum)
				// 拷贝版本库到oss
				ossPath := fmt.Sprintf("%s%s", osslink, path)
				copyExec := fmt.Sprintf("%s %s %s", obsCommand, path, ossPath)

				fmt.Println("shell:", copyExec)
				s, err := Exec_linux_shell(copyExec)
				// 拷贝完成后，更新数据 oss=1
				if err == nil {
					dbw.UpdateRepositoryOssbyPath(path)
				} else {
					
					utils.Log.Errorf("对象存储命令行执行失败,错误为: %s ; 执行命令为： %s  ; 命令返回值: %s ",err,copyExec,s )

				}
			}

			mu.Lock()
			ComsuNum++
			mu.Unlock()
			wg.Done()
		}
	}
}

//执行shell 命令
func Exec_linux_shell(s string) (string, error) {
	//阻塞式的执行外部shell命令的函数,等待执行完毕并返回标准输出
	//函数返回一个*Cmd，用于使用给出的参数执行name指定的程序
	cmd := exec.Command("/bin/bash", "-c", s)

	//读取io.Writer类型的cmd.Stdout，再通过bytes.Buffer(缓冲byte类型的缓冲器)将byte类型转化为string类型(out.String():这是bytes类型提供的接口)
	var out bytes.Buffer
	cmd.Stdout = &out
	//Run执行c包含的命令，并阻塞直到完成。  这里stdout被取出，cmd.Wait()无法正确获取stdin,stdout,stderr，则阻塞在那了
	err := cmd.Run()
	return out.String(), err

}