Golang多線程排序?qū)崿F(xiàn)快速高效地處理大規(guī)模數(shù)據(jù)

更新時間：2023年05月11日 10:10:20 作者：Luyoungs

Golang多線程排序是一種快速高效地處理大規(guī)模數(shù)據(jù)的方法，通過使用Golang的協(xié)程和通道，可以將排序任務分配到多個線程中并行處理，提高了排序的效率和速度，需要詳細了解可以參考下文

前言

本案例實現(xiàn)一個多線程排序算法，能夠?qū)o定的整數(shù)數(shù)組進行排序，使用 goroutines 對其進行并發(fā)化優(yōu)化。

隨機數(shù)生成器

func randProduce(randNums chan []int, wg *sync.WaitGroup) {
	for i := 0; i < 100; i++ {
		go rand1(randNums, wg)
	}
}
func rand1(randNums chan []int, wg *sync.WaitGroup) {
	r := rand.New(rand.NewSource(time.Now().Unix()))
	int1000 := make([]int, 1000000)
	for i := 0; i < 1000000; i++ {
		int1000[i] = r.Intn(1000000)
	}
	randNums <- int1000
	wg.Done()
}

使用goroutines并發(fā)地對各個子數(shù)組進行排序

func sort0(randNums chan []int, sortNums chan []int, wg *sync.WaitGroup) {
	for i := 0; i < 100; i++ {
		go sort2(randNums, sortNums, wg)
	}
}
func sort2(randNums chan []int, sortNums chan []int, wg *sync.WaitGroup) {
	int1000_Old := <-randNums
	sort.Ints(int1000_Old)
	sortNums <- int1000_Old
	wg.Done()
}

合并已排序的子數(shù)組得到最終排序結(jié)果

func mergeAll(sortNums chan []int, wg *sync.WaitGroup) []int {
	defer wg.Done()
	temp2 := <-sortNums
	var temp1 []int
	for i := 1; i <= 99; i++ {
		temp1 = make([]int, 1000000*i+1000000)
		copy(temp1, temp2)
		temp1 = merge(temp1, 1000000*i+1000000, <-sortNums, 1000000)
		temp2 = make([]int, 1000000*i+1000000)
		copy(temp2, temp1)
	}
	return temp2
}
func merge(nums1 []int, m int, nums2 []int, n int) []int {
	temp := make([]int, m)
	copy(temp, nums1)
	t, j := 0, 0 //t為temp的索引，j為nums2的索引
	for i := 0; i < len(nums1); i++ {
		if t >= len(temp) {
			nums1[i] = nums2[j]
			j++
			continue
		}
		if j >= n {
			nums1[i] = temp[t]
			t++
			continue
		}
		if nums2[j] <= temp[t] {
			nums1[i] = nums2[j]
			j++
		} else {
			nums1[i] = temp[t]
			t++
		}
	}
	return nums1
}

main 函數(shù)控制流程

func main() {
	fmt.Println("開始運行!")
	start := time.Now() // 獲取當前時間
	wg := sync.WaitGroup{}
	wg.Add(201)
	randNums := make(chan []int, 100)
	sortNUms := make(chan []int, 100)
	go randProduce(randNums, &wg)
	go sort0(randNums, sortNUms, &wg)
	go mergeAll(sortNUms, &wg)
	wg.Wait()
	// fmt.Println(l)
	elapsed := time.Since(start)
	fmt.Println("該函數(shù)執(zhí)行完成耗時：", elapsed)
}

思路

本案例采用了兩個 channel，分別存儲產(chǎn)生的的隨機數(shù)slice和排好順序的 slice，每一個 slice大小為 100 萬，一共一百個 slice，也就是一億個數(shù)據(jù)。

randNums := make(chan []int, 100)
	sortNUms := make(chan []int, 100)

程序一邊產(chǎn)生隨機數(shù)，一邊將產(chǎn)生的隨機數(shù)randNums發(fā)送到 sort 函數(shù)進行排序，排好順序后將數(shù)據(jù)發(fā)送到sortNUms。這兩個流程可以并行計算，因此：

go randProduce(randNums, &wg)
	go sort0(randNums, sortNUms, &wg)

合并也可以參與到并行計算之中，多加一個信號量就好：

go mergeAll(sortNUms, &wg)

運行結(jié)果：

(base) luliang@shenjian Sort % go build SortRoutine.go

(base) luliang@shenjian Sort % ./SortRoutine

開始運行!

該函數(shù)執(zhí)行完成耗時： 50.317081625s

性能比較

可以寫一個單線程的排序，但是數(shù)據(jù)產(chǎn)生還是多線程的：

package main
import (
	"fmt"
	"math/rand"
	"sort"
	"time"
)
func main() {
	fmt.Println("開始運行!")
	start := time.Now() // 獲取當前時間
	randNums := make(chan int, 10000)
	go randProduce1(randNums)
	randNums1 := make([]int, 100000000)
	for i := 0; i < 100000000; i++ {
		randNums1[i] = <-randNums
	}
	sort.Ints(randNums1)
	elapsed := time.Since(start)
	fmt.Println("該函數(shù)執(zhí)行完成耗時：", elapsed)
}
func randProduce1(randNums chan int) {
	for i := 0; i < 10000; i++ {
		go rand2(randNums)
	}
}
func rand2(randNums chan int) {
	r := rand.New(rand.NewSource(time.Now().Unix()))
	for i := 0; i < 10000; i++ {
		randNums <- r.Intn(10000000)
	}
}

運行結(jié)果為：