go 使用线程池做请求限流

//线程池
package myroutine

import (
	"fmt"
	"strconv"
)

/**
 * @ Author      : jgbb
 * @ Date        : Created in 2019/9/4 13:19
 * @ Description : TODO 线程池
 * @ Modified by :
 * @ Version     : 1.0
 */

func Init(poolSize int,name string) *RoutinePool{
	pool := &RoutinePool{
		Queue:make(chan func()),
		PoolSize:poolSize,
		Name:name,
	}
	defer pool.ExeTask()
	return pool
}

type RoutinePool struct {
	//缓存任务
	Queue chan func()
	PoolSize int
	Name string
}


// 添加任务到线程池
func (pool *RoutinePool) AddTask(task func()){
	pool.Queue <- task
}

//执行任务
func (pool *RoutinePool) ExeTask(){
	counter := make(chan int)
	for i:=0;i<pool.PoolSize;i++ {
		go func() {
			j := <- counter//哪条线程
			var count int64= 0//计数（线程跑了多少次）
			var stdout =pool.Name+"\t线程"+strconv.Itoa(j)+"\t"
			for task := range pool.Queue{
				count++
				fmt.Printf("%p\t%s\n",pool,stdout+strconv.FormatInt(count,10))

				task()
			}
		}()

		counter <- i
	}
}

//对server源码修改
const(
   DefaultPoolSize = 10
)

//每个请求对应的线程池
var PoolMap  = make(map[string]*myroutine.RoutinePool)

//golang源码
func (srv *Server) Serve(l net.Listener) error {
	defer l.Close()
	if fn := testHookServerServe; fn != nil {
		fn(srv, l)
	}
	var tempDelay time.Duration // how long to sleep on accept failure

	if err := srv.setupHTTP2_Serve(); err != nil {
		return err
	}

	srv.trackListener(l, true)
	defer srv.trackListener(l, false)

	baseCtx := context.Background() // base is always background, per Issue 16220
	ctx := context.WithValue(baseCtx, ServerContextKey, srv)
	for {
		rw, e := l.Accept()
		if e != nil {
			select {
			case <-srv.getDoneChan():
				return ErrServerClosed
			default:
			}
			if ne, ok := e.(net.Error); ok && ne.Temporary() {
				if tempDelay == 0 {
					tempDelay = 5 * time.Millisecond
				} else {
					tempDelay *= 2
				}
				if max := 1 * time.Second; tempDelay > max {
					tempDelay = max
				}
				srv.logf("http: Accept error: %v; retrying in %v", e, tempDelay)
				time.Sleep(tempDelay)
				continue
			}
			return e
		}
		tempDelay = 0
		c := srv.newConn(rw)
                
                /***********************修改开始*******************************/
		//将c.server(ctx)的处理过程放入线程池中
		//首先需要请求path,根据path获取对应的线程池
		c.r = &connReader{conn: c}
		c.r.setReadLimit(c.server.initialReadLimitSize()) //若不setReadLimit，无法读取到缓冲流中的数据
		c.bufr = newBufioReader(c.r)//用来读取流
		s,err := c.bufr.Peek(100)//缓冲流使用peek(),游标不会进行计数，这样才能流中的数据在后面的处理中复用。否则后续读取流会从游标开始
		news := make([]byte,0)
		for i:=0;i<100;i++ {
			news = append(news,s[i])
			if s[i] == 10 {
                                //10表示换行符，到此获取到所需信息
				break
			}
		}
		if err != nil {
			fmt.Errorf("my err:%v",err)
		}
		newss := string(news)
                //请求path当作线程池名称
		poolName := newss[strings.Index(newss,"/"):strings.LastIndex(newss," ")]


		c.setState(c.rwc, StateNew) // before Serve can return
		//go c.serve(ctx) //源码

		//放入线程池处理请求
		putPoolMap(poolName).AddTask(func() {
			c.serve(ctx)
		})
                /***********************修改结束*******************************/
	}
}

//生成线程池
//-参数1:线程池大小
//-参数2:线程池名称
func PutPoolMap(poolSize int,name string) *myroutine.RoutinePool{
	if _,ok := PoolMap[name]; !ok {
                //如果不存在对应的线程池，则生成一个
		PoolMap[name] = myroutine.Init(poolSize,name)
	}
        //返回对应的线程池
	return PoolMap[name]
}

//默认使用此方法生成线程池
//-参数1:线程池名称
func putPoolMap(name string) *myroutine.RoutinePool{
	return PutPoolMap(DefaultPoolSize,name)
}