需求
开发语言：golang
目的：并发10000个mqtt连接，循环发送publish信息，当时间戳小于某个值的时候，中止循环，退出连接
publish内容是json格式的，未设置时，有默认值，可以通过golang代码修改json内容
登录信息存取在csv文件中，csv文件有多少列，就并发多少个设备连接
话不多说，直接上代码
main.go
package main import ( "encoding/csv" "encoding/json" "fmt" "os" "strconv" "time" mqtt "github.com/eclipse/paho.mqtt.golang" ) var total int // 读取csv文件里的第3列数据，存入一个string数组里 func readcsv(filename string) []string { var userNameList []string f, _ := os.Open(filename) defer f.Close() w := csv.NewReader(f) data, err := w.ReadAll() if err != nil { fmt.Println(err) } for i := range data { userNameList = append(userNameList, data[i][2]) } return userNameList } func mqttDevice(username string, end_number chan int) { // mqtt设备连接,设置IP地址 opts := mqtt.NewClientOptions().AddBroker("localhost:1883") // 设置连接的用户名密码 opts.SetUsername(username) // 使用连接信息进行连接 client := mqtt.NewClient(opts) if token := client.Connect(); token.Wait() && token.Error() != nil { panic(token.Error()) } time.Sleep(1 * time.Second) fmt.Println("connect success:" + username) // 读取json文件，json文件里的是默认参数 fileReader, _ := os.Open("test.json") var eiopJsonMap map[string]interface{} json.NewDecoder(fileReader).Decode(&eiopJsonMap) // 设置一个开始时间戳和结束时间戳 startTime := 1598167852000 endTime := 1598167852000 // 循环发送遥测，每次遥测间隔时间戳为15分钟 for ; startTime < endTime; startTime = startTime + 900000 { // 定义一个ep的初始值为1,每循环一次就+1 var ep int ep++ eiopJsonMap["ts"] = startTime eiopJsonMap["values"].(map[string]interface{})["ep"] = ep // 把修改过的json内容从map转换为json格式 eiopJsonText, _ := json.Marshal(eiopJsonMap) fmt.Println(string(eiopJsonText)) // 发送遥测，发完之后休眠1秒 result := client.Publish("topic", 0, true, eiopJsonText) result.Wait() time.Sleep(1 * time.Second) } // 发送完信息之后，退出连接 fmt.Println("disconnect:" + username) client.Disconnect(250) total++ end_number <- total } func main() { userNameList := readcsv("connect_info.csv") endNumber := make(chan int, len(userNameList)) // 变量所有的username，通过go关键字并发多个设备 for _, userName := range userNameList { go mqttDevice(userName, endNumber) } // 当所有的设备都发送完毕后，关闭程序 for i := range endNumber { fmt.Println("已经有" + strconv.Itoa(i) + "个设备发送完毕") if i == len(userNameList) { return } } }
test.json
{ "ts": 1603088274000, "values": { "ep": 12 } }
connect_info.csv
localhost,1883,XecUwSmMGiYJp2BspMK2 localhost,1883,GqVqoPP2wblDjS2P9pQ9
