您好,登錄后才能下訂單哦!
這篇文章將為大家詳細講解有關怎么設置golang網絡通信超時,文章內容質量較高,因此小編分享給大家做個參考,希望大家閱讀完這篇文章后對相關知識有一定的了解。
1.超時設置
1.1 連接超時
func DialTimeout(network, address string, timeout time.Duration) (Conn, error)
第三個參數timeout可以用來設置連接超時設置。
如果超過timeout的指定的時間,連接沒有完成,會返回超時錯誤。
1.2 讀寫超時
在Conn定義中,包括讀寫的超時時間設置。
type Conn interface { // SetDeadline sets the read and write deadlines associated // with the connection. It is equivalent to calling both // SetReadDeadline and SetWriteDeadline. // ... ... SetDeadline(t time.Time) error // SetReadDeadline sets the deadline for future Read calls // and any currently-blocked Read call. // A zero value for t means Read will not time out. SetReadDeadline(t time.Time) error // SetWriteDeadline sets the deadline for future Write calls // and any currently-blocked Write call. // Even if write times out, it may return n > 0, indicating that // some of the data was successfully written. // A zero value for t means Write will not time out. SetWriteDeadline(t time.Time) error }
通過上面的函數說明,可以得知,這里的參數t是一個未來的時間點,所以每次讀或寫之前,都要調用SetXXX重新設置超時時間,
如果只設置一次,就會出現總是超時的問題。
2.1 server
server端監聽連接,如果收到連接請求,就是創建一個goroutine負責這個連接的數據收發。
為了測試超時,我們在寫操作之前,sleep 3s。
package main import ( "net" "log" "time" ) func main() { addr := "0.0.0.0:8080" tcpAddr, err := net.ResolveTCPAddr("tcp",addr) if err != nil { log.Fatalf("net.ResovleTCPAddr fail:%s", addr) } listener, err := net.ListenTCP("tcp", tcpAddr) if err != nil { log.Fatalf("listen %s fail: %s", addr, err) } else { log.Println("listening", addr) } for { conn, err := listener.Accept() if err != nil { log.Println("listener.Accept error:", err) continue } go handleConnection(conn) } } func handleConnection(conn net.Conn) { defer conn.Close() var buffer []byte = []byte("You are welcome. I'm server.") for { time.Sleep(3*time.Second)// sleep 3s n, err := conn.Write(buffer) if err != nil { log.Println("Write error:", err) break } log.Println("send:", n) } log.Println("connetion end") }
2.2 client
client建立連接時,使用的超時時間是3s。
創建連接成功后,設置連接的讀超時。
每次讀之前,都重新設置超時時間。
package main import ( "log" "net" "os" "time" ) func main() { connTimeout := 3*time.Second conn, err := net.DialTimeout("tcp", "127.0.0.1:8080", connTimeout) // 3s timeout if err != nil { log.Println("dial failed:", err) os.Exit(1) } defer conn.Close() readTimeout := 2*time.Second buffer := make([]byte, 512) for { err = conn.SetReadDeadline(time.Now().Add(readTimeout)) // timeout if err != nil { log.Println("setReadDeadline failed:", err) } n, err := conn.Read(buffer) if err != nil { log.Println("Read failed:", err) //break } log.Println("count:", n, "msg:", string(buffer)) } }
輸出結果
2019/05/12 16:18:19 Read failed: read tcp 127.0.0.1:51718->127.0.0.1:8080: i/o timeout 2019/05/12 16:18:19 count: 0 msg: 2019/05/12 16:18:20 count: 28 msg: You are welcome. I'm server. 2019/05/12 16:18:22 Read failed: read tcp 127.0.0.1:51718->127.0.0.1:8080: i/o timeout 2019/05/12 16:18:22 count: 0 msg: You are welcome. I'm server. 2019/05/12 16:18:23 count: 28 msg: You are welcome. I'm server. 2019/05/12 16:18:25 Read failed: read tcp 127.0.0.1:51718->127.0.0.1:8080: i/o timeout 2019/05/12 16:18:25 count: 0 msg: You are welcome. I'm server. 2019/05/12 16:18:26 count: 28 msg: You are welcome. I'm server.
補充:Golang中的并發限制與超時控制
并發
package main import ( "fmt" "time" ) func run(task_id, sleeptime int, ch chan string) { time.Sleep(time.Duration(sleeptime) * time.Second) ch <- fmt.Sprintf("task id %d , sleep %d second", task_id, sleeptime) return } func main() { input := []int{3, 2, 1} ch := make(chan string) startTime := time.Now() fmt.Println("Multirun start") for i, sleeptime := range input { go run(i, sleeptime, ch) } for range input { fmt.Println(<-ch) } endTime := time.Now() fmt.Printf("Multissh finished. Process time %s. Number of tasks is %d", endTime.Sub(startTime), len(input)) }
函數 run() 接受輸入的參數,sleep 若干秒。然后通過 go 關鍵字并發執行,通過 channel 返回結果。
channel 顧名思義,他就是 goroutine 之間通信的“管道"。管道中的數據流通,實際上是 goroutine 之間的一種內存共享。我們通過他可以在 goroutine 之間交互數據。
ch <- xxx // 向 channel 寫入數據
<- ch // 從 channel 中讀取數據
channel 分為無緩沖(unbuffered)和緩沖(buffered)兩種。例如剛才我們通過如下方式創建了一個無緩沖的 channel。
ch := make(chan string)
channel 的緩沖,我們一會再說,先看看剛才看看執行的結果。
三個 goroutine `分別 sleep 了 3,2,1秒。但總耗時只有 3 秒。所以并發生效了,go 的并發就是這么簡單。
按序返回
剛才的示例中,我執行任務的順序是 0,1,2。但是從 channel 中返回的順序卻是 2,1,0。這很好理解,因為 task 2 執行的最快嘛,所以先返回了進入了 channel,task 1 次之,task 0 最慢。
如果我們希望按照任務執行的順序依次返回數據呢?可以通過一個 channel 數組(好吧,應該叫切片)來做,比如這樣
package main import ( "fmt" "time" ) func run(task_id, sleeptime int, ch chan string) { time.Sleep(time.Duration(sleeptime) * time.Second) ch <- fmt.Sprintf("task id %d , sleep %d second", task_id, sleeptime) return } func main() { input := []int{3, 2, 1} chs := make([]chan string, len(input)) startTime := time.Now() fmt.Println("Multirun start") for i, sleeptime := range input { chs[i] = make(chan string) go run(i, sleeptime, chs[i]) } for _, ch := range chs { fmt.Println(<-ch) } endTime := time.Now() fmt.Printf("Multissh finished. Process time %s. Number of tasks is %d", endTime.Sub(startTime), len(input)) }
超時控制
剛才的例子里我們沒有考慮超時。然而如果某個 goroutine 運行時間太長了,那很肯定會拖累主 goroutine 被阻塞住,整個程序就掛起在那兒了。因此我們需要有超時的控制。
通常我們可以通過select + time.After 來進行超時檢查,例如這樣,我們增加一個函數 Run() ,在 Run() 中執行 go run() 。并通過 select + time.After 進行超時判斷。
package main import ( "fmt" "time" ) func Run(task_id, sleeptime, timeout int, ch chan string) { ch_run := make(chan string) go run(task_id, sleeptime, ch_run) select { case re := <-ch_run: ch <- re case <-time.After(time.Duration(timeout) * time.Second): re := fmt.Sprintf("task id %d , timeout", task_id) ch <- re } } func run(task_id, sleeptime int, ch chan string) { time.Sleep(time.Duration(sleeptime) * time.Second) ch <- fmt.Sprintf("task id %d , sleep %d second", task_id, sleeptime) return } func main() { input := []int{3, 2, 1} timeout := 2 chs := make([]chan string, len(input)) startTime := time.Now() fmt.Println("Multirun start") for i, sleeptime := range input { chs[i] = make(chan string) go Run(i, sleeptime, timeout, chs[i]) } for _, ch := range chs { fmt.Println(<-ch) } endTime := time.Now() fmt.Printf("Multissh finished. Process time %s. Number of task is %d", endTime.Sub(startTime), len(input)) }
運行結果,task 0 和 task 1 已然超時
并發限制
如果任務數量太多,不加以限制的并發開啟 goroutine 的話,可能會過多的占用資源,服務器可能會爆炸。所以實際環境中并發限制也是一定要做的。
一種常見的做法就是利用 channel 的緩沖機制。我們分別創建一個帶緩沖和不帶緩沖的 channel 看看
ch := make(chan string) // 這是一個無緩沖的 channel,或者說緩沖區長度是 0
ch := make(chan string, 1) // 這是一個帶緩沖的 channel, 緩沖區長度是 1
這兩者的區別在于,如果 channel 沒有緩沖,或者緩沖區滿了。goroutine 會自動阻塞,直到 channel 里的數據被讀走為止。舉個例子
package main import ( "fmt" ) func main() { ch := make(chan string) ch <- "123" fmt.Println(<-ch) }
這段代碼執行將報錯
fatal error: all goroutines are asleep - deadlock! goroutine 1 [chan send]: main.main() /tmp/sandbox531498664/main.go:9 +0x60 Program exited.
這是因為我們創建的 ch 是一個無緩沖的 channel。因此在執行到 ch<-"123",這個 goroutine 就阻塞了,后面的 fmt.Println(<-ch) 沒有辦法得到執行。所以將會報 deadlock 錯誤。
如果我們改成這樣,程序就可執行
package main import ( "fmt" ) func main() { ch := make(chan string, 1) ch <- "123" fmt.Println(<-ch) }
執行
123
Program exited.
如果我們改成這樣
package main import ( "fmt" ) func main() { ch := make(chan string, 1) ch <- "123" ch <- "123" fmt.Println(<-ch) fmt.Println(<-ch) }
盡管讀取了兩次 channel,但是程序還是會死鎖,因為緩沖區滿了,goroutine 阻塞掛起。第二個 ch<- "123" 是沒有辦法寫入的。
因此,利用 channel 的緩沖設定,我們就可以來實現并發的限制。我們只要在執行并發的同時,往一個帶有緩沖的 channel 里寫入點東西(隨便寫啥,內容不重要)。讓并發的 goroutine 在執行完成后把這個 channel 里的東西給讀走。這樣整個并發的數量就講控制在這個 channel 的緩沖區大小上。
比如我們可以用一個 bool 類型的帶緩沖 channel 作為并發限制的計數器。
然后在并發執行的地方,每創建一個新的 goroutine,都往 chLimit 里塞個東西。
for i, sleeptime := range input { chs[i] = make(chan string, 1) chLimit <- true go limitFunc(chLimit, chs[i], i, sleeptime, timeout) }
這里通過 go 關鍵字并發執行的是新構造的函數。他在執行完原來的 Run() 后,會把 chLimit 的緩沖區里給消費掉一個。
limitFunc := func(chLimit chan bool, ch chan string, task_id, sleeptime, timeout int) { Run(task_id, sleeptime, timeout, ch) <-chLimit }
這樣一來,當創建的 goroutine 數量到達 chLimit 的緩沖區上限后。主 goroutine 就掛起阻塞了,直到這些 goroutine 執行完畢,消費掉了 chLimit 緩沖區中的數據,程序才會繼續創建新的 goroutine。我們并發數量限制的目的也就達到了。
完整示例代碼
package main import ( "fmt" "time" ) func Run(task_id, sleeptime, timeout int, ch chan string) { ch_run := make(chan string) go run(task_id, sleeptime, ch_run) select { case re := <-ch_run: ch <- re case <-time.After(time.Duration(timeout) * time.Second): re := fmt.Sprintf("task id %d , timeout", task_id) ch <- re } } func run(task_id, sleeptime int, ch chan string) { time.Sleep(time.Duration(sleeptime) * time.Second) ch <- fmt.Sprintf("task id %d , sleep %d second", task_id, sleeptime) return } func main() { input := []int{3, 2, 1} timeout := 2 chLimit := make(chan bool, 1) chs := make([]chan string, len(input)) limitFunc := func(chLimit chan bool, ch chan string, task_id, sleeptime, timeout int) { Run(task_id, sleeptime, timeout, ch) <-chLimit } startTime := time.Now() fmt.Println("Multirun start") for i, sleeptime := range input { chs[i] = make(chan string, 1) chLimit <- true go limitFunc(chLimit, chs[i], i, sleeptime, timeout) } for _, ch := range chs { fmt.Println(<-ch) } endTime := time.Now() fmt.Printf("Multissh finished. Process time %s. Number of task is %d", endTime.Sub(startTime), len(input)) }
運行結果
Multirun start task id 0 , timeout task id 1 , timeout task id 2 , sleep 1 second Multissh finished. Process time 5s. Number of task is 3 Program exited.
chLimit 的緩沖是 1。task 0 和 task 1 耗時 2 秒超時。task 2 耗時 1 秒。總耗時 5 秒。并發限制生效了。
如果我們修改并發限制為 2
chLimit := make(chan bool, 2)
運行結果
Multirun start task id 0 , timeout task id 1 , timeout task id 2 , sleep 1 second Multissh finished. Process time 3s. Number of task is 3 Program exited.
task 0 , task 1 并發執行,耗時 2秒。task 2 耗時 1秒。總耗時 3 秒。符合預期。
有沒有注意到代碼里有個地方和之前不同。這里,用了一個帶緩沖的 channel
chs[i] = make(chan string, 1)
還記得上面的例子么。如果 channel 不帶緩沖,那么直到他被消費掉之前,這個 goroutine 都會被阻塞掛起。
然而如果這里的并發限制,也就是 chLimit 生效阻塞了主 goroutine,那么后面消費這些數據的代碼并不會執行到。。。于是就 deadlock 拉!
for _, ch := range chs { fmt.Println(<-ch) }
關于怎么設置golang網絡通信超時就分享到這里了,希望以上內容可以對大家有一定的幫助,可以學到更多知識。如果覺得文章不錯,可以把它分享出去讓更多的人看到。
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。