golang实现异步并发sokect

package main
import (
	"fmt"
	"net"
	"os"
	"time"
)
const (
	MAX_CONN_NUM = 5
)
//echo server Goroutine
func EchoFunc(conn net.Conn) {
	defer conn.Close()
	buf := make([]byte, 1024)
	for {
		_, err := conn.Read(buf)
		if err != nil {
			//println("Error reading:", err.Error())
			return
		}
		//send reply
		_, err = conn.Write(buf)
		if err != nil {
			//println("Error send reply:", err.Error())
			return
		}
	}
}
//initial listener and run
func main() {
	listener, err := net.Listen("tcp", "0.0.0.0:8088")
	if err != nil {
		fmt.Println("error listening:", err.Error())
		os.Exit(1)
	}
	defer listener.Close()
	fmt.Printf("running ...\n")
	var cur_conn_num int = 0
	conn_chan := make(chan net.Conn)
	ch_conn_change := make(chan int)
	go func() {
		for conn_change := range ch_conn_change {
			cur_conn_num += conn_change
		}
	}()
	go func() {
		for _ = range time.Tick(1e8) {
			fmt.Printf("cur conn num: %f\n", cur_conn_num)
		}
	}()
	for i := 0; i < MAX_CONN_NUM; i++ {
		go func() {
			for conn := range conn_chan {
				ch_conn_change <- 1
				EchoFunc(conn)
				ch_conn_change <- -1
			}
		}()
	}
	for {
		conn, err := listener.Accept()
		if err != nil {
			println("Error accept:", err.Error())
			return
		}
		conn_chan <- conn
	}
}

再看这段代码使用传统思维实行方式:

//
//A echo server with max-connections limit and interval connection show
//
package main
import (
 "fmt"
 "net"
 "os"
 "time"
)
const (
 MAX_CONN_NUM = 5
)
//echo server Goroutine
func EchoFunc(conn net.Conn, conn_close_flag chan int) {
 defer conn.Close()
 defer func() {
 conn_close_flag <- -1
 }()
 buf := make([]byte, 1024)
 for {
 _, err := conn.Read(buf)
 if err != nil {
 //println("Error reading:", err.Error())
 return
 }
 //send reply
 _, err = conn.Write(buf)
 if err != nil {
 //println("Error send reply:", err.Error())
 return
 }
 }
}
//initial listener and run
func main() {
 listener, err := net.Listen("tcp", "0.0.0.0:8088")
 if err != nil {
 println("error listening:", err.Error())
 os.Exit(1)
 }
 defer listener.Close()
 fmt.Printf("running ...\n")
 var cur_conn_num float64 = 0
 ch_conn_change := make(chan int, MAX_CONN_NUM)
 tick := time.Tick(1e8)
 for {
 //read all close flags berfor accept new connection
 //TODO: better code to handle batch close?
 readmore := 1
 for readmore > 0 {
 select {
 case conn_change := <-ch_conn_change:
 cur_conn_num = cur_conn_num + float64(conn_change)
 default:
 readmore = 0
 }
 }
 //FIXME: tick block by listener.Accept()
 select {
 case <-tick:
 fmt.Printf("cur conn num: %f\n", cur_conn_num)
 default:
 }
 if cur_conn_num >= MAX_CONN_NUM {
 //reach MAX_CONN_NUM, waiting for exist connection close
 time.Sleep(time.Second)
 } else {
 //accept new connetion
 conn, err := listener.Accept()
 if err != nil {
 println("Error accept:", err.Error())
 return
 }
 cur_conn_num++
 go EchoFunc(conn, ch_conn_change)
 }
 }
}

这个案例中,golang通过多个goroutine + channel堵塞做到传统的顺序执行模式。

代码来自google group的讨论:https://groups.google.com/forum/#!topic/golang-china/q4pFH-AGnfs

本文来自:开源中国博客

感谢作者:yunfound

查看原文:golang实现异步并发sokect

• 编辑
• 预览