打算闲的时候写一个tcp_pool.刚开始做,就结束了.因为我走到了死胡同.

来源:互联网 时间:1970-01-01


package gyc_server_deployimport ( "errors" "fmt" "net" "sync" "time")var GlobalLock *sync.RWMutex = new(sync.RWMutex)type IPPool struct { IP []string Maxconn int TimeOut int64}type ConnectionPool struct { IPPool UseConnection map[string]int Lock *sync.RWMutex Connection map[string]chan net.Conn}var returnChan chan net.Conn = make(chan net.Conn, 1000)var exitChan chan bool = make(chan bool, 1)func (this *ConnectionPool) GetConnection(ip string) *NetConnection { if list := this.Connection[ip]; len(list) > 0 { con := NetConnection{new(sync.RWMutex), <-list, returnChan} GlobalLock.Lock() this.UseConnection[ip] = this.UseConnection[ip] + 1 GlobalLock.Unlock() return &con } if this.UseConnection[ip] < this.Maxconn { c, err := dial(ip, this.TimeOut) if err != nil { fmt.Println(err) return nil } con := NetConnection{new(sync.RWMutex), c, returnChan} GlobalLock.Lock() this.UseConnection[ip] = this.UseConnection[ip] + 1 GlobalLock.Unlock() return &con } con := NetConnection{new(sync.RWMutex), <-this.Connection[ip], returnChan} GlobalLock.Lock() this.UseConnection[ip] = this.UseConnection[ip] + 1 GlobalLock.Unlock() return &con}func (this *ConnectionPool) New(ip string) error { if this.Index(ip) { if _, ok := this.Connection[ip]; !ok { ConChan := make(chan net.Conn, this.Maxconn) err := createConList(ip, ConChan, this.TimeOut) if err != nil { return err } this.Lock.Lock() this.Connection[ip] = ConChan this.Lock.Unlock() } return errors.New(fmt.Sprintf("%s: Already exists.", ip)) } ConChan := make(chan net.Conn, this.Maxconn) err := createConList(ip, ConChan, this.TimeOut) if err != nil { return err } if !this.Index(ip) { this.Lock.Lock() this.IP = append(this.IP, ip) this.Connection[ip] = ConChan this.Lock.Unlock() return nil } return errors.New(fmt.Sprintf("%s: Already exists.", ip))}func (this *ConnectionPool) Index(ip string) bool { this.Lock.RLock() defer this.Lock.RUnlock() for _, v := range this.IP { if v == ip { return true } } return false}func (this *ConnectionPool) Close() { exitChan <- true}func (this *IPPool) Init() *ConnectionPool { M := ConnectionPool{Lock: new(sync.RWMutex), Connection: make(map[string]chan net.Conn)} for _, addr := range this.IP { ConChan := make(chan net.Conn, this.Maxconn) err := createConList(addr, ConChan, this.TimeOut) if err != nil { continue } M.Connection[addr] = ConChan } M.IP = this.IP M.TimeOut = this.TimeOut M.Maxconn = this.Maxconn M.UseConnection = make(map[string]int) go RecoveryAndExit(&M) return &M}func createConList(ip string, ConChan chan net.Conn, t int64) error { for i := 0; i < cap(ConChan); i++ { con, err := dial(ip, t) if err != nil { fmt.Println(err) continue } ConChan <- con } if len(ConChan) <= 0 { return errors.New(fmt.Sprintf("%s: can't connection.", ip)) } return nil}func dial(ip string, t int64) (net.Conn, error) { if _, err := net.ResolveTCPAddr("tcp", ip); err != nil { return nil, err } return net.DialTimeout("tcp", ip, time.Second*time.Duration(t))}func RecoveryAndExit(pool *ConnectionPool) { for { select { case con := <-returnChan: if con == nil { GlobalLock.Lock() pool.UseConnection[ip] = pool.UseConnection[ip] - 1 GlobalLock.Unlock() continue } ip := con.RemoteAddr().String() if list, ok := pool.Connection[ip]; ok { GlobalLock.Lock() pool.UseConnection[ip] = pool.UseConnection[ip] - 1 GlobalLock.Unlock() list <- con } else { con.Close() } case exit := <-exitChan: if exit { pool.Lock.Lock() defer pool.Lock.Unlock() for k, v := range pool.Connection { num := len(v) for i := 0; i < num; i++ { con := <-v con.Close() } close(v) delete(pool.Connection, k) } return } } }}type NetConnection struct { lock *sync.RWMutex con net.Conn returnChan chan net.Conn}func (this *NetConnection) Write(b []byte) (int, error) { this.lock.RLock() defer this.lock.RUnlock() return this.con.Write(b)}func (this *NetConnection) Read(b []byte) (int, error) { this.lock.RLock() defer this.lock.RUnlock() return this.con.Read(b)}func (this *NetConnection) LocalAddr() net.Addr { this.lock.RLock() defer this.lock.RUnlock() return this.con.LocalAddr()}func (this *NetConnection) RemoteAddr() net.Addr { this.lock.RLock() defer this.lock.RUnlock() return this.con.RemoteAddr()}func (this *NetConnection) Close() { this.lock.Lock() defer this.lock.Unlock() this.con = nil this.returnChan <- this.con}



相关阅读:
Top