func (v *value) Add(val float64) {
	for {
		oldBits := atomic.LoadUint64(&v.valBits)
		newBits := math.Float64bits(math.Float64frombits(oldBits) + val)
		if atomic.CompareAndSwapUint64(&v.valBits, oldBits, newBits) {
			return
		}
	}
}

// Collector is the interface implemented by anything that can be used by
// Prometheus to collect metrics. A Collector has to be registered for
// collection. See Registerer.Register.
//
// The stock metrics provided by this package (Gauge, Counter, Summary,
// Histogram, Untyped) are also Collectors (which only ever collect one metric,
// namely itself). An implementer of Collector may, however, collect multiple
// metrics in a coordinated fashion and/or create metrics on the fly. Examples
// for collectors already implemented in this library are the metric vectors
// (i.e. collection of multiple instances of the same Metric but with different
// label values) like GaugeVec or SummaryVec, and the ExpvarCollector.
type Collector interface {
	// Describe sends the super-set of all possible descriptors of metrics
	// collected by this Collector to the provided channel and returns once
	// the last descriptor has been sent. The sent descriptors fulfill the
	// consistency and uniqueness requirements described in the Desc
	// documentation. (It is valid if one and the same Collector sends
	// duplicate descriptors. Those duplicates are simply ignored. However,
	// two different Collectors must not send duplicate descriptors.) This
	// method idempotently sends the same descriptors throughout the
	// lifetime of the Collector. If a Collector encounters an error while
	// executing this method, it must send an invalid descriptor (created
	// with NewInvalidDesc) to signal the error to the registry.
	Describe(chan<- *Desc)
	// Collect is called by the Prometheus registry when collecting
	// metrics. The implementation sends each collected metric via the
	// provided channel and returns once the last metric has been sent. The
	// descriptor of each sent metric is one of those returned by
	// Describe. Returned metrics that share the same descriptor must differ
	// in their variable label values. This method may be called
	// concurrently and must therefore be implemented in a concurrency safe
	// way. Blocking occurs at the expense of total performance of rendering
	// all registered metrics. Ideally, Collector implementations support
	// concurrent readers.
	Collect(chan<- Metric)
}

var (
	sentBytes = prometheus.NewCounter(prometheus.CounterOpts{
		Namespace: "etcd",
		Subsystem: "network",
		Name:      "client_grpc_sent_bytes_total",
		Help:      "The total number of bytes sent to grpc clients.",
	})

	receivedBytes = prometheus.NewCounter(prometheus.CounterOpts{
		Namespace: "etcd",
		Subsystem: "network",
		Name:      "client_grpc_received_bytes_total",
		Help:      "The total number of bytes received from grpc clients.",
	})
)

func init() {
	prometheus.MustRegister(sentBytes)
	prometheus.MustRegister(receivedBytes)
}

type gaugeVecFuncCollector struct {
	desc                        *prometheus.Desc
	gaugeVecFuncWithLabelValues []gaugeVecFuncWithLabelValues
	labelsDeduplicatedMap       map[string]bool
}

// NewGaugeVecFunc
func NewGaugeVecFunc(opts GaugeOpts, labelNames []string) *gaugeVecFuncCollector {
	return &gaugeVecFuncCollector{
		desc: prometheus.NewDesc(
			prometheus.BuildFQName(opts.Namespace, opts.Subsystem, opts.Name),
			opts.Help,
			labelNames,
			opts.ConstLabels,
		),
		labelsDeduplicatedMap: make(map[string]bool),
	}
}

// Describe
func (dc *gaugeVecFuncCollector) Describe(ch chan<- *prometheus.Desc) {
	ch <- dc.desc
}

// Collect
func (dc *gaugeVecFuncCollector) Collect(ch chan<- prometheus.Metric) {
	for _, v := range dc.gaugeVecFuncWithLabelValues {
		ch <- prometheus.MustNewConstMetric(dc.desc, prometheus.GaugeValue, v.gaugeVecFunc(), v.labelValues...)
	}
}

// RegisterGaugeVecFunc 
// 同一组labelValues只能注册一次
func (dc *gaugeVecFuncCollector) RegisterGaugeVecFunc(labelValues []string, gaugeVecFunc func() float64) (err error) {
	// prometheus每次允许收集一次labelValues相同的metric
	deduplicateKey := strings.Join(labelValues, "")
	if dc.labelsDeduplicatedMap[deduplicateKey] {
		return fmt.Errorf("labelValues func already registered, labelValues:%v", labelValues)
	}
	dc.labelsDeduplicatedMap[deduplicateKey] = true
	handlePanicGaugeVecFunc := func() float64 {
		if rec := recover(); rec != nil {
			const size = 10 * 1024
			buf := make([]byte, size)
			buf = buf[:runtime.Stack(buf, false)]
			logger.Errorf("gaugeVecFunc panic:%v\nstack:%s", rec, buf)
		}
		return gaugeVecFunc()
	}
	dc.gaugeVecFuncWithLabelValues = append(dc.gaugeVecFuncWithLabelValues, gaugeVecFuncWithLabelValues{
		gaugeVecFunc: handlePanicGaugeVecFunc,
		labelValues:  labelValues,
	})
	return nil
}

http_requests_total{job="prometheus",group="canary"}

http_requests_total{job="prometheus"}[5m]

package main

import (
	"context"
	"github.com/go-stomp/stomp"
	"github.com/hanjm/log"
	"os"
	"os/signal"
	"strconv"
	"sync"
	"syscall"
	"time"
)

func main() {
	var wg sync.WaitGroup
	ctx, cancel := context.WithCancel(context.Background())

	wg.Add(1)
	go func() {
		defer wg.Done()
		publisher(ctx, "/topic/stomp")
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		Subscriber(ctx, "channel1", "Consumer.channel1.stomp")
	}()
	//
	wg.Add(1)
	go func() {
		defer wg.Done()
		Subscriber(ctx, "channel2", "Consumer.channel2.stomp")
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		Subscriber(ctx, "channel3", "/topic/stomp")
	}()

	defer func() {
		cancel()
		wg.Wait()
	}()
	SignalsListen()
}

func publisher(ctx context.Context, destination string) {
	conn, err := stomp.Dial("tcp", "127.0.0.1:61613")
	if err != nil {
		log.Fatal(err)
		return
	}
	defer conn.Disconnect()
	for i := 0; ; i++ {
		select {
		case <-ctx.Done():
			return
		case <-time.After(time.Second):
			err = conn.Send(
				destination,  // destination
				"text/plain", // content-type
				[]byte("Test message #"+strconv.Itoa(i)), stomp.SendOpt.Header("persistent", "true")) // body
			if err != nil {
				log.Error(err)
				return
			}
		}
	}
}

func Subscriber(ctx context.Context, clientID string, destination string) {
	conn, err := stomp.Dial("tcp", "127.0.0.1:61613")
	if err != nil {
		log.Fatal(err)
		return
	}
	defer conn.Disconnect()
	sub, err := conn.Subscribe(destination, stomp.AckClientIndividual, stomp.SubscribeOpt.Id(clientID), stomp.SubscribeOpt.Header("persistent", "true"))
	if err != nil {
		log.Fatal(err)
		return
	}
	go func() {
		select {
		case <-ctx.Done():
			err := sub.Unsubscribe()
			if err != nil {
				log.Fatal(clientID, err)
				return
			}
			return
		}
	}()
	for m := range sub.C {
		if m.Err != nil {
			log.Fatal(err)
			return
		}
		log.Infof("%s msg body:%s", clientID, m.Body)
		//log.Infof("%s msg header:%s", clientID, *m.Header)
		//log.Infof("%s msg content-type:%s", clientID, m.ContentType)
		//log.Infof("%s msg destination:%s", clientID, m.Destination)
		m.Conn.Ack(m)
	}
	log.Info("close sub")
}

func SignalsListen() {
	sigs := make(chan os.Signal, 1)
	signal.Notify(sigs, syscall.SIGQUIT,
		syscall.SIGTERM,
		syscall.SIGINT,
		syscall.SIGUSR1,
		syscall.SIGUSR2)

	switch <-sigs {
	case syscall.SIGTERM, syscall.SIGINT, syscall.SIGQUIT:
		log.Info("service close")
	}
	return
}

package all_bench

import (
	"bytes"
	"context"
	"github.com/eclipse/paho.mqtt.golang"
	"github.com/go-stomp/stomp"
	"github.com/hanjm/log"
	"pack.ag/amqp"
	"sync/atomic"
	"testing"
	"time"
)

var msgData = bytes.Repeat([]byte("1"), 1024)

var (
	stompDestination = "bench-stomp"
	amqpDestination  = "bench-amqp"
	mqttDestination  = "bench-mqtt"
	pubMsgCount      = 20000
	subMsgCount      = 100
)

func TestMain(m *testing.M) {
	m.Run()
}

// go test -bench Publish  -benchmem
// go test -bench Sub  -benchmem
func BenchmarkStompPublish(b *testing.B) {
	conn, err := stomp.Dial("tcp", "127.0.0.1:61613")
	if err != nil {
		log.Fatal(err)
		return
	}
	defer conn.Disconnect()

	b.N = pubMsgCount
	b.ReportAllocs()
	b.ResetTimer()

	for i := 0; i < b.N; i++ {
		err = conn.Send(
			stompDestination, // destination
			"text/plain",     // content-type
			msgData) // body
		if err != nil {
			log.Error(err)
			return
		}
	}
}

func BenchmarkAmqpPublish(b *testing.B) {
	// Create client
	client, err := amqp.Dial("amqp://127.0.0.1",
		amqp.ConnSASLPlain("system", "manager"),
	)
	if err != nil {
		log.Fatal("Dialing AMQP server:", err)
	}
	defer client.Close()

	// Open a session
	session, err := client.NewSession()
	if err != nil {
		log.Fatal("Creating AMQP session:", err)
	}
	defer func() {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		err = session.Close(ctx)
		if err != nil {
			log.Errorf("failed to close session:%s", err)
			return
		}
		//log.Info("session close")
	}()

	// Create a sender
	sender, err := session.NewSender(
		amqp.LinkTargetAddress(amqpDestination),
		amqp.LinkSourceDurability(amqp.DurabilityUnsettledState),
		amqp.LinkSourceExpiryPolicy(amqp.ExpiryNever),
	)
	if err != nil {
		log.Fatal("Creating sender link:", err)
	}

	defer func() {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		err := sender.Close(ctx)
		if err != nil {
			log.Errorf("failed to close sender:%s", err)
			return
		}
		//log.Infof("sender close")
	}()

	ctx := context.Background()
	msg := amqp.NewMessage(msgData)

	b.N = pubMsgCount
	b.ReportAllocs()
	b.ResetTimer()

	for i := 0; i < b.N; i++ {
		// Send message
		err = sender.Send(ctx, msg)
		if err != nil {
			log.Fatal("Sending message:", err)
		}
		if err != nil {
			log.Fatal(err)
			return
		}
	}
}

func BenchmarkMqttPublish(b *testing.B) {
	opt := mqtt.NewClientOptions().SetClientID("pubClient").SetCleanSession(false)
	opt.AddBroker("tcp://127.0.0.1:1883")
	client := mqtt.NewClient(opt)
	t := client.Connect()
	err := t.Error()
	if err != nil {
		log.Fatal(err)
		return
	}
	if t.Wait() {
		err := t.Error()
		if err != nil {
			log.Fatal(err)
			return
		}
	}
	defer func() {
		client.Disconnect(10000)
	}()

	b.N = pubMsgCount
	b.ReportAllocs()
	b.ResetTimer()

	for i := 0; i < b.N; i++ {
		token := client.Publish(mqttDestination, 2, true, msgData)
		err := token.Error()
		if err != nil {
			log.Fatal(err)
			return
		}
	}
}

func BenchmarkStompSubscriber(b *testing.B) {
	conn, err := stomp.Dial("tcp", "127.0.0.1:61613")
	if err != nil {
		log.Fatal(err)
		return
	}
	clientID := "1"
	//defer conn.Disconnect()
	sub, err := conn.Subscribe(stompDestination, stomp.AckClientIndividual, stomp.SubscribeOpt.Id(clientID))
	if err != nil {
		log.Fatal(err)
		return
	}
	//defer func() {
	//	err := sub.Unsubscribe()
	//	if err != nil {
	//		log.Fatal(clientID, err)
	//		return
	//	}
	//	return
	//}()
	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Second)
	defer cancel()

	b.N = subMsgCount
	b.ReportAllocs()
	b.ResetTimer()
	defer b.StopTimer()
	var i int64 = 0

	go func() {
		for range time.Tick(time.Second) {
			if atomic.LoadInt64(&i) >= int64(b.N) {
				cancel()
			}
		}
	}()

	defer func() {
		//log.Info("close")
	}()
	for {
		select {
		case m := <-sub.C:
			if m.Err != nil {
				log.Fatal(m.Err)
				return
			}
			m.Conn.Ack(m)
			i++
			if atomic.LoadInt64(&i) > int64(b.N) {
				return
			}
		case <-ctx.Done():
			return
		}
	}
}

func BenchmarkAmqpSubscriber(b *testing.B) {
	// Create client
	client, err := amqp.Dial("amqp://127.0.0.1",
		amqp.ConnSASLPlain("system", "manager"),
	)
	if err != nil {
		log.Fatal("Dialing AMQP server:", err)
	}
	//defer client.Close()

	// Open a session
	session, err := client.NewSession()
	if err != nil {
		log.Fatal("Creating AMQP session:", err)
	}

	clientID := "1"
	defer func() {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		err := session.Close(ctx)
		if err != nil {
			log.Errorf("%s failed to close session:%s", clientID, err)
			return
		}
		//log.Errorf("%s session close", clientID)
	}()

	// Continuously read messages
	// Create a receiver
	receiver, err := session.NewReceiver(
		amqp.LinkSourceAddress(amqpDestination),
		amqp.LinkCredit(10),
	)
	if err != nil {
		log.Fatal("Creating receiver link:", err)
	}
	defer func() {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		err := receiver.Close(ctx)
		if err != nil {
			log.Errorf("%s failed to close receiver:%s", clientID, err)
			return
		}
		//log.Errorf("%s receiver close", clientID)
	}()
	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Second)
	defer cancel()

	b.N = subMsgCount
	b.ReportAllocs()
	b.ResetTimer()
	defer b.StopTimer()
	var i int64 = 0

	go func() {
		for range time.Tick(time.Second) {
			if atomic.LoadInt64(&i) >= int64(b.N) {
				cancel()
			}
		}
	}()
	for {
		// Receive next message
		msg, err := receiver.Receive(ctx)
		if err != nil {
			if err == context.Canceled {
				log.Infof("Reading message from AMQP:%s", err)
				break
			}
			log.Errorf("Reading message from AMQP:%s", err)
			break
		}
		//log.Infof("%s msg body:%s value:%T %s", clientID, msg.GetData(), msg.Value, msg.Value)
		// Accept message
		msg.Accept()
		atomic.AddInt64(&i, 1)
		if atomic.LoadInt64(&i) > int64(b.N) {
			//log.Info("return")
			return
		}
	}
}

func BenchmarkMqttSubscriber(b *testing.B) {
	opt := mqtt.NewClientOptions().SetClientID("subClient").SetCleanSession(false)
	opt.AddBroker("tcp://127.0.0.1:1883")
	client := mqtt.NewClient(opt)
	t := client.Connect()
	if t.Wait() {
		err := t.Error()
		if err != nil {
			log.Fatal(err)
			return
		}
	}
	defer func() {
		client.Disconnect(1000)
	}()

	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Second)
	defer cancel()

	b.N = subMsgCount
	b.ReportAllocs()
	b.ResetTimer()
	defer b.StopTimer()
	var i int64 = 0

	go func() {
		for range time.Tick(time.Second) {
			if atomic.LoadInt64(&i) >= int64(b.N) {
				cancel()
			}
		}
	}()
	client.Subscribe(mqttDestination, 2, func(c mqtt.Client, m mqtt.Message) {
		//log.Infof("%s msg body:%s", "1", m.Payload())
		m.Ack()
		atomic.AddInt64(&i, 1)
		if atomic.LoadInt64(&i) > int64(b.N) {
			//log.Info("return")
			return
		}
	})
	select {
	case <-ctx.Done():
		break
	}
	log.Info("close sub")
}