大量连接时使用 使用epoll管理 or golang 多协程

package main

import (
    "log"
    "net"
    "reflect"
    "sync"
    "golang.org/x/sys/unix"
    "syscall"

)

type epoll struct {
    fd          int
    connections map[int]net.Conn
    lock        *sync.RWMutex
}
func MkEpoll() (*epoll, error) {
    fd, err := unix.EpollCreate1(0)
    if err != nil {
        return nil, er
    }
    return &epoll{
        fd:          fd,
        lock:        &sync.RWMutex{},
        connections: make(map[int]net.Conn),
    }, nil
}
func (e *epoll) Add(conn net.Conn) error {
    // Extract file descriptor associated with the connection
    fd := socketFD(conn)
    err := unix.EpollCtl(e.fd, syscall.EPOLL_CTL_ADD, fd, &unix.EpollEvent{Events: unix.POLLIN | unix.POLLHUP, Fd: int32(fd)})
    if err != nil {
        return er
    }
    e.lock.Lock()
    defer e.lock.Unlock()
    e.connections[fd] = conn
    if len(e.connections)%100 == 0 {
        log.Printf("total number of connections: %v", len(e.connections))
    }
    return nil
}
func (e *epoll) Remove(conn net.Conn) error {
    fd := socketFD(conn)
    err := unix.EpollCtl(e.fd, syscall.EPOLL_CTL_DEL, fd, nil)
    if err != nil {
        return er
    }
    e.lock.Lock()
    defer e.lock.Unlock()
    delete(e.connections, fd)
    if len(e.connections)%100 == 0 {
        log.Printf("total number of connections: %v", len(e.connections))
    }
    return nil
}
func (e *epoll) Wait() ([]net.Conn, error) {
    events := make([]unix.EpollEvent, 100)
    n, err := unix.EpollWait(e.fd, events, 100)
    if err != nil {
        return nil, er
    }
    e.lock.RLock()
    defer e.lock.RUnlock()
    var connections []net.Conn
    for i := 0; i < n; i++ {
        conn := e.connections[int(events[i].Fd)]
        connections = append(connections, conn)
    }
    return connections, nil
}
func socketFD(conn net.Conn) int {
    //tls := reflect.TypeOf(conn.UnderlyingConn()) == reflect.TypeOf(&tls.Conn{})
    // Extract the file descriptor associated with the connection
    //connVal := reflect.Indirect(reflect.ValueOf(conn)).FieldByName("conn").Elem()
    tcpConn := reflect.Indirect(reflect.ValueOf(conn)).FieldByName("conn")
    //if tls {
    //    tcpConn = reflect.Indirect(tcpConn.Elem())
    //}
    fdVal := tcpConn.FieldByName("fd")
    pfdVal := reflect.Indirect(fdVal).FieldByName("pfd")
    return int(pfdVal.FieldByName("Sysfd").Int())
}

package main

import (
    "log"
    "net"
    "strconv"
    "time"
)

const RobotCount = 100

var SerAddr *net.TCPAdd

var robots map[int]*Robot

func StartTcpserver(){
    listener, _ := net.ListenTCP("tcp", SerAddr)
    log.Println("tcp server started.")
    for {
        conn, _ := listener.Accept()
        go func(conn net.Conn) {
            for {
                msg := "response..." + strconv.Itoa(idx)
                conn.Write([]byte(msg))
                time.Sleep(time.Second*3)
            }
        }(conn)
    }
}

func StartRobots(){
    time.Sleep(time.Second)
    robots = make(map[int]*Robot, RobotCount)
    for k:=0;k<RobotCount;k++ {
        r := Robot{}
        r.Init(k)
        err := r.Start()
        if err != nil {
            break
        }
        if err != nil {
            log.Println("epoller add conn failed:", err)
        }
        fd := socketFD(r.conn)
        robots[fd] = &
    }
}

func RobotLoigc() {
    for {
        for _, r := range robots {
            r.ProcessMsg()
        }
        time.Sleep(time.Microsecond*20)
    }
}


func main() {
    SerAddr, _ = net.ResolveTCPAddr("tcp", ":8888")

    // 开启服务器
    go StartTcpserver()

    // 初始化epoll
    epoller, err := MkEpoll()
    if err != nil {
        log.Println("new epoll failed:", err)
        return
    }

    // 开启机器人客户端
    StartRobots()

    // 将机器人conn交给epoll管理
    for _, r := range robots {
        epoller.Add(r.conn)
    }

    // 消息处理
    go RobotLoigc()

    // 等待epoll通知
    go func(){
        for {
            connections, err := epoller.Wait()
            if err != nil {
                log.Printf("failed to epoll wait %v", err)
                continue
            }
            for _, conn := range connections {
                if conn == nil {
                    break
                }
                var buf = make([]byte, 1024)

                n, err := conn.Read(buf)
                if err != nil {
                    if err := epoller.Remove(conn); err != nil {
                        log.Printf("failed to remove %v", err)
                    }
                    conn.Close()
                }
                if n > 0 {
                    fd := socketFD(conn)
                    if r, ok := robots[fd];ok{
                        r.ReceiveMsg(buf)
                    }
                }
            }
        }
    }()

    time.Sleep(time.Hour)
}

package main

import (
    "log"
    "net"
)

const msgBufLen = 10

type Robot struct {
    id int
    conn net.Conn
    recvMsg chan []byte
}


func (r *Robot) Init(id int) {
    r.id = id
    r.recvMsg = make(chan []byte, msgBufLen)
}

func (r *Robot) Start() error {
    conn, err := net.DialTCP ("tcp", nil, SerAddr)
    if err != nil {
        log.Println("connect to server failed.")
        return er
    }
    r.conn = conn
    return nil
}

func (r *Robot) ReceiveMsg(msg []byte)  bool {
    if len(r.recvMsg) >= msgBufLen {
        return  false
    }
    r.recvMsg <- msg
    return true
}

func (r *Robot) ProcessMsg() {
    if len(r.recvMsg) > 0 {
        select {
        case msg:=<-r.recvMsg:
            log.Println(r.id, "receiv msg:", string(msg))
        }
    }
}