需求
當(dāng)需要同時(shí)ping/telnet多個(gè)ip時(shí),可以通過引入ping包/telnet包實(shí)現(xiàn),也可以通過go調(diào)用cmd命令實(shí)現(xiàn),不過后者調(diào)用效率較差,所以這里選擇ping包和telnet包
還有就是高并發(fā)的問題,可以通過shell腳本或者go實(shí)現(xiàn)高并發(fā),所以我選擇的用go自帶的協(xié)程實(shí)現(xiàn),但是如果要同時(shí)處理1000+個(gè)ip,考慮到機(jī)器的性能,需要ratelimit控制開辟的go協(xié)程數(shù)量,這里主要寫一下我的建議和淌過的坑
ping
參考鏈接: https://github.com/sparrc/go-ping
import "github.com/sparrc/go-ping"
import "time"
func (p *Ping) doPing(timeout time.Duration, count int, ip string) (err error) {
pinger, cmdErr := ping.NewPinger(ip)
if cmdErr != nil {
glog.Error("Failed to ping " + p.ipAddr)
err = cmdErr
return
}
pinger.Count = count
pinger.Interval = time.Second
pinger.Timeout = timeout
// true的話,代表是標(biāo)準(zhǔn)的icmp包,false代表可以丟包類似udp
pinger.SetPrivileged(false)
// 執(zhí)行
pinger.Run()
// 獲取ping后的返回信息
stats := pinger.Statistics()
//延遲
latency = float64(stats.AvgRtt)
// 標(biāo)準(zhǔn)的往返總時(shí)間
jitter = float64(stats.StdDevRtt)
//丟包率
packetLoss = stats.PacketLoss
return
}
注意: pinger.Run() 這里執(zhí)行的時(shí)候是阻塞的,如果并發(fā)量大的時(shí)候,程序會(huì)卡死在這里,所以當(dāng)有高并發(fā)的需求時(shí)建議如下處理:
go pinger.Run()
time.Sleep(timeout)
telnet
package main
import (
"github.com/reiver/go-telnet"
)
func doTelnet(ip string, port int) {
var caller telnet.Caller = telnet.StandardCaller
address := ip + ":"+ strconv.Itoa(port)
// DialToAndCall 檢查連通性并且調(diào)用
telnet.DialToAndCall(address, caller)
}
}
bug出現(xiàn)報(bào)錯(cuò):
lookup tcp/: nodename nor servname provided, or not known
解決:
修改string(port)為strconv.Itoa(port)
DialToAndCall這種方式telnet無法設(shè)置超時(shí)時(shí)間,默認(rèn)的超時(shí)時(shí)間有1分鐘,所以使用DialTimeout這個(gè)方式實(shí)現(xiàn)telnet
import "net"
func doTelnet(ip string, ports []string) map[string]string {
// 檢查 emqx 1883, 8083, 8080, 18083 端口
results := make(map[string]string)
for _, port := range ports {
address := net.JoinHostPort(ip, port)
// 3 秒超時(shí)
conn, err := net.DialTimeout("tcp", address, 3*time.Second)
if err != nil {
results[port] = "failed"
} else {
if conn != nil {
results[port] = "success"
_ = conn.Close()
} else {
results[port] = "failed"
}
}
}
return results
}
shell高并發(fā)
本質(zhì)就是讀取ip.txt文件里的ip,然后調(diào)用ping方法,實(shí)現(xiàn)高并發(fā)也是借助遍歷所有的ip然后同一交給操作系統(tǒng)去處理高并發(fā)
while read ip
do
{
doPing(ip)
}
done ip.txt
go高并發(fā)限速
import (
"context"
"fmt"
"log"
"time"
"sync"
"golang.org/x/time/rate"
)
func Limit(ips []string)([]string, []string, error) {
//第一個(gè)參數(shù)是每秒鐘最大的并發(fā)數(shù),第二個(gè)參數(shù)是桶的容量,第一次的時(shí)候每秒可執(zhí)行的數(shù)量就是桶的容量,建議這兩個(gè)值都寫成一樣的
r := rate.NewLimiter(10, 10)
ctx := context.Background()
wg := sync.WaitGroup{}
wg.Add(len(ips))
lock := sync.Mutex{}
var success []string
var fail []string
defer wg.Done()
for _,ip:=range ips{
//每次消耗2個(gè),放入一個(gè),消耗完了還會(huì)放進(jìn)去,如果初始是5個(gè),所以這段代碼再執(zhí)行到第4次的時(shí)候筒里面就空了,如果當(dāng)前不夠取兩個(gè)了,本次就不取,再放一個(gè)進(jìn)去,然后返回false
err := r.WaitN(ctx, 2)
if err != nil {
log.Fatal(err)
}
go func(ip string) {
defer func() {
wg.Done()
}()
err := doPing(time.Second, 2, ip)
lock.Lock()
defer lock.Unlock()
if err != nil {
fail = append(fail, ip)
return
} else {
success = append(success, ip)
}
}(ip)
}
// wait等待所有g(shù)o協(xié)程結(jié)束
wg.wait()
return success,fail,nil
}
func main() {
ips := [2]string{"192.168.1.1","192.168.1.2"}
success,fail,err := Limit(ips)
if err != nil {
fmt.Printf("ping error")
}
}
這里注意一個(gè)并發(fā)實(shí)現(xiàn)的坑,在for循環(huán)里使用goroutine時(shí)要把遍歷的參數(shù)傳進(jìn)去才能保證每個(gè)遍歷的參數(shù)都被執(zhí)行,否則只能執(zhí)行一次
(拓展)管道、死鎖
先看個(gè)例子:
func main() {
go print() // 啟動(dòng)一個(gè)goroutine
print()
}
func print() {
fmt.Println("*******************")
}
輸出結(jié)果:
*******************
沒錯(cuò),只有一行,因?yàn)楫?dāng)go開辟一個(gè)協(xié)程想去執(zhí)行print方法時(shí),主函數(shù)已經(jīng)執(zhí)行完print并打印出來,所以goroutine還沒有機(jī)會(huì)執(zhí)行程序就已經(jīng)結(jié)束了,解決這個(gè)問題可是在主函數(shù)里加time.sleep讓主函數(shù)等待goroutine執(zhí)行完,也可以使用WaitGroup.wait等待goroutine執(zhí)行完,還有一種就是信道
信道分無緩沖信道和緩沖信道
無緩沖信道
無緩沖信道也就是定義長度為0的信道,存入一個(gè)數(shù)據(jù),從無緩沖信道取數(shù)據(jù),若信道中無數(shù)據(jù),就會(huì)阻塞,還可能引發(fā)死鎖,同樣數(shù)據(jù)進(jìn)入無緩沖信道, 如果沒有其他goroutine來拿走這個(gè)數(shù)據(jù),也會(huì)阻塞,記住無緩沖數(shù)據(jù)并不存儲(chǔ)數(shù)據(jù)
func main() {
var channel chan string = make(chan string)
go func(message string) {
channel- message // 存消息
}("Ping!")
fmt.Println(-messages) // 取消息
}
緩存信道
顧名思義,緩存信道可以存儲(chǔ)數(shù)據(jù),goroutine之間不會(huì)發(fā)生阻塞,for循環(huán)讀取信道中的數(shù)據(jù)時(shí),一定要判斷當(dāng)管道中不存在數(shù)據(jù)時(shí)的情況,否則會(huì)發(fā)生死鎖,看個(gè)例子
channel := make(chan int, 3)
channel - 1
channel - 2
channel - 3
// 顯式關(guān)閉信道
close(channel)
for v := range channel {
fmt.Println(v)
// 如果現(xiàn)有數(shù)據(jù)量為0,跳出循環(huán),與顯式關(guān)閉隧道效果一樣,選一個(gè)即可
if len(ch) = 0 {
break
}
}
但是這里有個(gè)問題,信道中數(shù)據(jù)是可以隨時(shí)存入的,所以我們遍歷的時(shí)候無法確定目前的個(gè)數(shù)就是信道的總個(gè)數(shù),所以推薦使用select監(jiān)聽信道
// 創(chuàng)建一個(gè)計(jì)時(shí)信道
timeout := time.After(1 * time.Second)
// 監(jiān)聽3個(gè)信道的數(shù)據(jù)
select {
case v1 := - c1: fmt.Printf("received %d from c1", v1)
case v2 := - c2: fmt.Printf("received %d from c2", v2)
case v3 := - c3: fmt.Printf("received %d from c3", v3)
case - timeout:
is_timeout = true // 超時(shí)
break
}
}
以上為個(gè)人經(jīng)驗(yàn),希望能給大家一個(gè)參考,也希望大家多多支持腳本之家。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教。
您可能感興趣的文章:- Java進(jìn)階之高并發(fā)核心Selector詳解
- java的多線程高并發(fā)詳解
- MySQL高并發(fā)生成唯一訂單號的方法實(shí)現(xiàn)
- PHP解決高并發(fā)的優(yōu)化方案實(shí)例
- Redis處理高并發(fā)機(jī)制原理及實(shí)例解析
- C#請求唯一性校驗(yàn)支持高并發(fā)的實(shí)現(xiàn)方法
- Tomcat+Mysql高并發(fā)配置優(yōu)化講解
- Redis高并發(fā)問題的解決方法
- 如何理解軟件系統(tǒng)的高并發(fā)