深入理解ActiveMQ消息队列协议STMOP AMQP MQTT

hanjm's blog

深入理解Prometheus(GO SDK及Grafana基本面板) 深入理解Prometheus(GO SDK及Grafana基本面板) Macos Docker container连接宿主机172.17.0.1的办法 Nginx With gRPC编译安装 Go sql.Driver的mysql Driver 中的一个有意思的行为学习Influxdb GRPC文档阅读心得 Go如何优雅地错误处理(Error Handling and Go 1) Go如何优雅地错误处理(Error Handling and Go 1) 深入理解NATS & NATS Streaming (踩坑记) 深入理解GO时间处理(time.Time) 深入理解GO时间处理(time.Time) Go如何精确计算小数-Decimal研究-Tidb MyDecimal问题 DockerContainer下gdb无法正常工作的解决办法 Go sync.Pool Slice Benchmark Go最佳实践 GO Logger 日志实践 Linux Cli下酷工具收集(持续) MacOS下酷工具收集(持续) Linux Cli下酷工具收集(持续) MacOS下酷工具收集(持续) 知名公司架构资料整理(持续) 知名公司架构资料整理(持续) Mysql 连接池问题 Go strings.TrimLeft() strings.TrimPrefix().md

本文作者： hanjm · 2019-02-07 · via hanjm's blog

AWS MQ是完全托管的 ActiveMQ 服务, 最近需要使用, 于是学习其文档, 实践其特性, 由于 ActiveMQ 支持非常丰富的协议, OpenWire amqp stomp mqtt, 所以也学习了各大协议的特性及其SDK.

需要注意的是, 首先要根据其docker hub镜像文档上的几步操作, 将镜像中的默认配置文件复制到自定义的本机conf目录下 /usr/local/activemq/conf, 然后就快速地启动了一个默认配置的 ActiveMQ server

这个特性对于监控MQ非常有用, 默认配置时关闭的, 需要在配置文件activemq.xml中打开.

通配符可以用在配置文件中表名作用范围, 也可以用于订阅时的destination名字, 这个功能很不错.

xxx对应类似NSQ中的Channel概念.

需要在activemq.xml中配置virtualDestinationInterceptor的范围 prefix及其他选项.

ActiveMQ支持延时消息及定时消息, 在message header中带上如下字段即可, 其中AMQ_SCHEDULED_PERIOD的最大值是long的最大值, 所以可以设置延时很长时间.

如果broker投递给消费者消息, 没有ACK或NACK, 则会触发重新投递, 投递超过一定次数则会进入死信队列, 默认只有一个公共的死信队列ActiveMQ.DLQ, 如果需要给topic分别设置死信队列, 则要在修改activemq.xml.

STOMP是Simple (or Streaming) Text Orientated Messaging Protocol 的缩写, 设计思路借鉴了HTTP, 有content-type, header, body, frame based, text based等类似HTTP的相关概念, 设计文档 < https://stomp.github.io/stomp-specification-1.2.html>, 非常得简洁, 一页就讲完了.

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
}

AMQP相比 stomp mqtt 就复杂得多, 毕竟名字就是高级消息队列(Advanced Message Queuing Protocol ).

github.com/vcabbage/amqp 76star 13issue 5contributors
github.com/go-stomp/stomp 132star 3issue 14contributors
github.com/eclipse/paho.mqtt.golang 650star 20issue 34contributors

作为SDK, 分别测试了下pub sub 1KB大小的消息普通场景.

publish性能上, amqp=stomp>mqtt, amqp和stomp差不多, 是mqtt的两倍多.
subscribe性能上, amqp比stomp快一点, mqtt则慢很多.

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")
}

如果使用了virtualTopic, 那么默认配置下, virtualTopic对应的Queue越多, 发送越慢, 因为默认virtualTopic转发到queue是串行的, 需要调整concurrentSend=true启用并发发送到queue.

https://activemq.apache.org/virtual-destinations
https://issues.jboss.org/browse/ENTMQ-1093
https://github.com/apache/activemq/blob/9abe2c6f97c92fc99c5a2ef02846f62002a671cf/activemq-broker/src/main/java/org/apache/activemq/broker/region/virtual/VirtualTopicInterceptor.java#L87

concurrentStoreAndDispatchQueues设置为false. 默认配置下, 这个值是true, 根据文档所说在快速消费者情况下, 此值设置为true可以加快持久化消息的性能, 因为被快速消费了消息可以不用落盘, 但实测发现此值为true则10个producer并发发送和1个producer并发发送的性能是一样的没有提高. 设置为false之后提高producer并发则可获得性能倍速提高, 并且单个producer的发送性能并没有下降.

启用mKahaDB, ActiveMQ为了减少打开的文件描述符数量, 默认是用一个KahaDB实例来持久化消息, 但是在磁盘性能比较好的情况下, 一个kahaDB实例发挥不出磁盘的潜力, 启用多个kahaDB后性能可以获得倍速增长. 可以按queue名字的pattern来设置多个kahaDB实例, 也可以使用perDestination="true"设置每个queue一个kahaDB实例, 但这个参数也有坑, 如果destination名字超过了42个字符串, 则会被截断, 发送会报不可恢复的错. 可解决的办法是手动分好destination使用的kahadb, 但是这个配置后续不能动态改了, 只能新开Broker然后迁移. 否则会重启后如果分配规则改变导致分配到了不同的kahadb, 则之前的数据不会被消费.

http://sigreen.github.io/2016/02/10/amq-tuning.html

https://activemq.apache.org/kahadb#multim-kahadb-persistence-adapter

此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。原文来自 — 版权归原作者所有。

推荐订阅源

hanjm's blog