博客

为了整合这些使用做的一个demo，所以做的比较复杂，并没有太大的实际意义。

只是为了弄懂这些概念练练手。

练手版：

package main

import (
	"fmt"
	"strconv"
	"strings"
	"sync"
)

type MapFunc func(interface{}) interface{}

type Map struct {
	MapF        MapFunc
	in          chan interface{}
	out         chan interface{}
	parallelism uint
}

func NewMap(mapFunc MapFunc, parallelism uint) *Map {
	_map := &Map{
		mapFunc,
		make(chan interface{}),
		make(chan interface{}),
		parallelism,
	}
	go _map.doStream()
	return _map
}

func (m *Map) doStream() {
	sem := make(chan struct{}, m.parallelism)
	for elem := range m.in {
		sem <- struct{}{}
		go func(e interface{}) {
			defer func() { <-sem }()
			trans := m.MapF(e)
			m.out <- trans
		}(elem)
	}
	for i := 0; i < int(m.parallelism); i++ {
		sem <- struct{}{}
	}
	close(m.out)
}

func output(_map *Map) {
	for elem := range _map.out {
		// go func(e interface{}) {
		// 	fmt.Println(elem)
		// }(elem)
		fmt.Println(elem)
	}
	wg.Done()
}

func input(_map *Map) {
	for i := 0; i < 10; i++ {
		_map.in <- "hello world" + strconv.Itoa(i)
	}
	close(_map.in)
	wg.Done()
}

var wg sync.WaitGroup

func main() {
	map1 := NewMap(toUpper, 1)

	wg.Add(2)
	go output(map1)
	go input(map1) 
	wg.Wait()
}

var toUpper = func(in interface{}) interface{} {
	msg := in.(string)
	result := strings.ToUpper(msg)
	return result
}

实际版：

package main

import (
	"fmt"
	"strconv"
	"strings"
	"sync"
)

func trans(ch chan string) {
	for elem := range ch {
		trans := strings.ToUpper(elem)
		fmt.Println(trans)
	}
	wg.Done()
}

func input(ch chan string) {
	for i := 0; i < 10; i++ {
		ch <- "hello world" + strconv.Itoa(i)
	}
	close(ch)
	wg.Done()
}

var wg sync.WaitGroup

func main() {
	ch := make(chan string)
	wg.Add(2)
	go input(ch)
	go trans(ch)
	wg.Wait()
}

这两个代码的结果是一样的：