使用golang开发mqtt服务压力测试工具 清明

package main

import (
    "flag"
    "fmt"
    "sync"
    "time"
    //导入mqtt包
    MQTT "github.com/eclipse/paho.mqtt.golang"
)

var f MQTT.MessageHandler = func(client MQTT.Client, msg MQTT.Message) {
    fmt.Printf("TOPIC: %s\n", msg.Topic())
    fmt.Printf("MSG: %s\n", msg.Payload())
}
var fail_nums int = 0

func main() {
    //生成连接的客户端数
    c := flag.Uint64("n", 3000, "client nums")
    flag.Parse()
    nums := int(*c)
    wg := sync.WaitGroup{}
    for i := 0; i < nums; i++ {
        wg.Add(1)
        time.Sleep(5 * time.Millisecond)
        go createTask(i, &wg)
    }
    wg.Wait()
}

func createTask(taskId int, wg *sync.WaitGroup) {
    defer wg.Done()
    opts := MQTT.NewClientOptions().AddBroker("tcp://192.168.1.110:1883").SetUsername("test").SetPassword("test")
    opts.SetClientID(fmt.Sprintf("go-simple-client:%d-%d", taskId, time.Now().Unix()))
    opts.SetDefaultPublishHandler(f)
    opts.SetConnectTimeout(time.Duration(60) * time.Second)

    //创建连接
    c := MQTT.NewClient(opts)

    if token := c.Connect(); token.WaitTimeout(time.Duration(60)*time.Second) && token.Wait() && token.Error() != nil {
        fail_nums++
        fmt.Printf("taskId:%d,fail_nums:%d,error:%s \n", taskId, fail_nums, token.Error())
        return
    }

    //每隔5秒向topic发送一条消息
    i := 0
    for {
        i++
        time.Sleep(time.Duration(5) * time.Second)
        text := fmt.Sprintf("this is msg #%d! from task:%d", i, taskId)
        token := c.Publish("test", 1, false, text)
        token.Wait()
    }

    c.Disconnect(250)
    fmt.Println("task ok!!")
}

　　

package main

import (
	"bytes"
	"crypto/rand"
	"crypto/tls"
	"encoding/json"
	"flag"
	"fmt"
	"math/big"
	"sync"
	"time"

	MQTT "github.com/eclipse/paho.mqtt.golang"
)

type Device struct {
	ID                  int     `json:"id"`
	Mac                 string  `json:"mac"`
	Name                string  `json:"name"`
	ProductType         string  `json:"product_type"`
	Lang                string  `json:"lang"`
	IP                  string  `json:"ip"`
	IP2                 string  `json:"ip2"`
	Gateway             string  `json:"gateway"`
	Gateway2            string  `json:"gateway2"`
	Netmask             string  `json:"netmask"`
	Netmask2            string  `json:"netmask2"`
	Inet                string  `json:"inet"`
	AreaID              int     `json:"area_id"`
	State               int     `json:"state"`
	RestartTime         string  `json:"restart_time"`
	LiveDetection       int     `json:"live_detection"`
	FeverHigh           float32 `json:"fever_high"`
	FeverLow            float32 `json:"fever_low"`
	TempAdd             float32 `json:"temp_add"`
	FaceRecThred        float32 `json:"face_rec_thred"`
	FaceMinThred        float32 `json:"half_face_rec_thred"`
	GateIsOpen          int     `json:"gate_is_open"`
	GateOpenType        int     `json:"gate_open_type"`
	GateOpenDur         int     `json:"gate_open_dur"`
	GateOpenCondition   int     `json:"gate_open_condition"`
	IoType              int     `json:"io_type"`
	IDThred             float32 `json:"id_thred"`
	VisitorGateOpenType int     `json:"visitor_gate_open_type"`
	VisitorCardNo       string  `json:"visitor_card_no"`
	Volume              int     `json:"volume"`
	ScreenOff           int     `json:"screen_off"`
	ScreenFillLight     int     `json:"screen_fill_light"`
	SuccessMsg          string  `json:"success_msg"`
	FailMsg             string  `json:"fail_msg"`
	Version             string  `json:"version"`
	LogSnap             int     `json:"log_snap"`
	SyncPeriod          int     `json:"sync_period"`
	FlowKeep            int     `json:"flow_keep"`
	LastConnect         string  `json:"last_connect"`
	LastModify          string  `json:"last_modify"`
	CreateTime          string  `json:"create_time"`
	Online              bool    `json:"online"`
	GateVoltage         int     `json:"gate_voltage"`
	GateAlarm           int     `json:"gate_alarm"`
	GateAlarmSec        int     `json:"gate_alarm_sec"`
	Brightness          int     `json:"brightness"`
	AutoBacklightCom    int     `json:"auto_backlight_com"`
	AudioUpload         string  `json:"audio_upload"`
	OSVersion           string  `json:"os_version"`
	APKversion          string  `json:"apk_version"`
	MyVersion           string  `json:"funpacs_version"`
}

type DeviceState struct {
	CPU        float32 `json:"CPU"`
	CPUTemp    float32 `json:"CPUTemp"`
	Memory     int64   `json:"Memory"`
	Runtime    int64   `json:"runtime"`
	APK        int     `json:"APK"`
	APKRuntime int64   `json:"APKRuntime"`
	NPU        bool    `json:"NPU"`
}

type MqttMsg struct {
	Mac         string      `json:"mac"`
	Cate        string      `json:"cate"`
	SubCate     int         `json:"subCate"`
	Cmd         string      `json:"cmd"`
	SN          uint64      `json:"SN"`
	Time        int64       `json:"time"`
	DeviceData  Device      `json:"deviceData"`
	DeviceState DeviceState `json:"deviceState"`
}

var f MQTT.MessageHandler = func(client MQTT.Client, msg MQTT.Message) {
	fmt.Printf("TOPIC: %s\n", msg.Topic())
	fmt.Printf("MSG: %s\n", msg.Payload())
}

func CreateRandomString(len int) string {
	var container string
	var str = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890"
	b := bytes.NewBufferString(str)
	length := b.Len()
	bigInt := big.NewInt(int64(length))
	for i := 0; i < len; i++ {
		randomInt, _ := rand.Int(rand.Reader, bigInt)
		container += string(str[randomInt.Int64()])
	}
	return container
}

var fail_nums int = 0
var gMyDevice Device

func main() {
	fmt.Println("PACS test start")

	//生成连接的客户端数
	c := flag.Uint64("n", 1, "client nums")
	flag.Parse()
	nums := int(*c)

	fmt.Println("Client number: ", nums)

	wg := sync.WaitGroup{}
	for i := 0; i < nums; i++ {
		wg.Add(1)
		time.Sleep(5 * time.Millisecond)
		go createTask(i, &wg)
	}
	wg.Wait()
}

func createTask(taskId int, wg *sync.WaitGroup) {
	defer wg.Done()
	opts := MQTT.NewClientOptions().AddBroker("tls://192.168.6.123:8883").SetUsername("superpacsmqtt").SetPassword("mqttpacssuper")
	opts.SetClientID(fmt.Sprintf("pacs-test-client:%d-%d", taskId, time.Now().Unix()))
	opts.SetTLSConfig(&tls.Config{InsecureSkipVerify: true})
	opts.SetDefaultPublishHandler(f)
	opts.SetConnectTimeout(time.Duration(60) * time.Second)

	//创建连接
	c := MQTT.NewClient(opts)

	if token := c.Connect(); token.WaitTimeout(time.Duration(60)*time.Second) && token.Wait() && token.Error() != nil {
		fail_nums++
		fmt.Printf("taskId:%d,fail_nums:%d,error:%s \n", taskId, fail_nums, token.Error())
		return
	}

	gServerRunTime := time.Now().Unix()
	//mac := "TEST" + strconv.Itoa(taskId)
	mac := "TEST" + CreateRandomString(8)

	topic := "hub/funpacs/u/" + mac

	gMyDevice.ID = taskId
	gMyDevice.Mac = mac
	gMyDevice.Name = mac
	gMyDevice.ProductType = "MS-5738"
	gMyDevice.Lang = "en"
	gMyDevice.State = 1
	gMyDevice.MyVersion = "1.0.68"

	data := MqttMsg{SN: 1, Time: time.Now().Unix(), Mac: mac, Cate: "status", Cmd: "deviceData", DeviceData: gMyDevice}
	jsDat, _ := json.Marshal(&data)
	msg := string(jsDat)
	fmt.Println(msg)
	c.Publish(topic, 1, false, msg)

	//每隔5秒向topic发送一条消息
	i := 0
	for {
		i++
		time.Sleep(time.Duration(5) * time.Second)

		runtime := time.Now().Unix() - gServerRunTime
		ds := DeviceState{Runtime: runtime, APK: i}

		sdata := MqttMsg{SN: 1, Time: time.Now().Unix(), Mac: mac, Cate: "status", Cmd: "deviceStat", DeviceState: ds}
		jsDat, _ := json.Marshal(&sdata)
		msg := string(jsDat)
		fmt.Println(msg)
		token := c.Publish(topic, 1, false, msg)
		token.Wait()
	}

}