concurrent_download

concurrent_download.go

package main

import (
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"os"
	"time"
)

type receiver struct {
	seqId int
	buf   io.ReadCloser
}

func main() {

	fileSize := 10485760
	size := 512000 // 500K
	fileUrl := "http://cdntest-10048632.cos.myqcloud.com/10M.bin"

	maxTimes := 0
	if n := fileSize % size; n == 0 {
		maxTimes = int(fileSize / size)
	} else {
		maxTimes = int(fileSize/size) + 1
	}

	fmt.Printf("File: %s\n", fileUrl)
	startTime := time.Now()

	ch := make(chan receiver)
	for i := 0; i < maxTimes; i++ {
		go download(fileUrl, i, i*size, i*size+size, ch)
	}

	// tmp file

	tmpFiles := make([]string, maxTimes)
	// 等待所有任务完成
	for i := 0; i < maxTimes; i++ {
		r := <-ch

		contents, err := ioutil.ReadAll(r.buf)
		r.buf.Close()

		if err != nil {
			panic(err)
		}

		tmpFiles[r.seqId] = fmt.Sprintf("/tmp/download/%d.tmp", r.seqId)

		ioutil.WriteFile(tmpFiles[r.seqId], contents, 0666)
		if err != nil {
			panic(err)
		}
	}

	fmt.Println("Merge tmp files...\n")
	// 合并文件
	file, err := os.OpenFile("./file.tmp", os.O_WRONLY|os.O_CREATE, 0666)
	if err != nil {
		panic(err)
	}

	for _, f := range tmpFiles {
		//fmt.Println(f)
		if b, err := ioutil.ReadFile(f); err != nil {
			panic(err)
			break
		} else {
			if _, err := file.Write(b); err != nil {
				panic(err)
			}
		}

		//os.Remove(f)
	}

	latency := time.Now().Sub(startTime)
	fmt.Printf("Download done! \nLatency:%13v\n", latency)
}

func download(urlStr string, seqId int, rangeFrom int, rangeTo int, ch chan receiver) {
	request, _ := http.NewRequest(http.MethodGet, urlStr, nil)
	request.Header.Add("Range", fmt.Sprintf("bytes=%d-%d", rangeFrom, rangeTo))

	client := &http.Client{}

	if response, err := client.Do(request); err != nil {
		panic(fmt.Sprintf("#%d Error: %s", seqId, err.Error()))
	} else {
		if response.StatusCode != http.StatusPartialContent {
			panic(fmt.Sprintf("#%d Response status invalid", seqId))
		}

		fmt.Printf("#%d\tFinished Range: %d-%d\n", seqId, rangeFrom, rangeTo)
		ch <- receiver{
			seqId: seqId,
			buf:   response.Body,
		}
	}
}