[Go]TCP服务中增加消息队列与工作池

之前的处理中每一个连接都会创建一个主groutine,每个连接中的主groutine中创建出读groutine 和写groutine

每个连接处理业务再单独开出一个groutine,这样如果有10万并发的连接 ， 将会出现30万groutine ，其中读写占20万阻塞住的 ， 不占用资源。处理业务的有10万groutine ，会不停的切换 ， 比较占有cpu资源 ， 现在把处理业务的groutine限制住 ，创建出一个工作池，里面存的是每个worker ，每个worker groutine去读取自己对应的channel,这个channel是个有缓存的channel作为消息队列使用

package snet

import (
    "bufio"
    fmtlogmath/randnettime"
)

type Conn struct {
    IP             string
    Port           uint32
    TCPConn        *net.TCPConn
    MsgChan        chan []byte
    ExitChan       chan bool
    Closed         
    WorkerPool     []chan []
    WorkerPoolSize uint32
    PreWorkerQueue uint32
}

func NewConn(IP string,Port uint32,WorkerPoolSize uint32) *Conn {
    s := &Conn{
        IP:             IP,Port:           Port,MsgChan:        make(chan []),ExitChan:       make(chan make([]chan [],WorkerPoolSize),WorkerPoolSize: WorkerPoolSize,PreWorkerQueue: 1024Conn) Start() {
    log.Printf(%s:%d start...\ntcp4",fmt.Sprintf(%s:%dif err != nil {
            log.Println(resolve tcp addr err listen tcp err 0
        for {
            conn,err := listener.AcceptTCP()
             nil {
                log.Println(accept tcp err  conn
            go c.StartRead()
            go c.StartWrite()
            connid++
        }
    }()
    select {}
}
func (c *Conn) StartRead() {
    log.Println(read groutine is waiting)
    defer c.Stop()
    defer log.Println(read groutine exit)
    reader := bufio.NewReader(c.TCPConn)
     {
        lineBytes,err := reader.ReadBytes('\n')
        startread read bytes error  len(lineBytes)
        line := lineBytes[:len-1]
        log.Println(start read from client (line))
        if c.WorkerPoolSize>0{
            c.SendMsgToWorker(line)
        }else{
            go c.HandleMsg(line)
        }
    }
}
func (c *Conn) StartWrite() {
    log.Println(write groutine is waiting)
    defer log.Println(write groutine exit)
     {
        case data := <-c.MsgChan:
            if _,err := c.TCPConn.Write(data); err !=startwrite conn write error start write from server (data))
        case <-c.ExitChan:
            return
        }
    }
}
func (c *Conn) HandleMsg(data []) {
    res := res:%s(data))
    c.MsgChan <- [](res)
}
func (c *Conn) SendMsgToWorker(data []) {
    rand.Seed(time.Now().UnixNano())
    workerId := rand.Intn(int(c.WorkerPoolSize))
    c.WorkerPool[workerId] <- data
}
func (c *Conn) StartWorkerPool() {
    for i := 0; i < int(c.WorkerPoolSize); i++ {
        c.WorkerPool[i] = int,queue chan []) {
    log.Println(start one worker groutine is waiting:queue:
            c.HandleMsg(data)
            log.Println(one worker groutine is finshed:Conn) Stop() {
    if c.Closed {
        return
    }
    c.Closed = true
    c.ExitChan <- 

    c.TCPConn.Close()
    close(c.ExitChan)
    close(c.MsgChan)
}