kpt-pasture/service/mqtt/consumer.go

package mqtt

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

var (
	bufferPool = sync.Pool{
		New: func() interface{} {
			return make([]byte, 1024) // 根据实际情况调整缓冲区大小
		},
	}
	readMsgChan  = make(chan []byte, 2)
	writeMsgChan = make(chan []byte, 2)
)

// Consumer 处理收到的消息
func (s *IMqttClient) Consumer(handle func([]byte)) {
	ac := time.NewTicker(2 * time.Minute)
	var allCnt int32 = 0
	go func() {
		for {
			select {
			case <-ac.C:
				fmt.Println("allCnt:", allCnt)
				atomic.StoreInt32(&allCnt, 0)
			}
		}
	}()
	for {
		select {
		case msg := <-writeMsgChan:
			handle(msg)
		case rsg := <-readMsgChan:
			writeMsgChan <- rsg
			atomic.AddInt32(&allCnt, 1)
		}
	}
}