消息队列

版权声明 本站原创文章 由 萌叔 发表 转载请注明 萌叔 | http://vearne.cc 本文基于rocketmq-all-4.8.0 1. client使用Filter RocketMQ消费者订阅消息时，可以指定以某种表达式进行过滤，目前支持2种类型–Tag和SQL92。 1.1 Tag 按照Tag进行过滤. 首先需要给消息打上tag，注意：1条消息只能包含最多1个tag 生产者代码示例1 package main import ( "context" "fmt" "os" "strconv" "github.com/apache/rocketmq-client-go/v2" "github.com/apache/rocketmq-client-go/v2/primitive" "github.com/apache/rocketmq-client-go/v2/producer" ) // Package main implements a simple producer to send message. func main() { p, _ := rocketmq.NewProducer( producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.12.100:9876"})), producer.WithRetry(2), ) err := p.Start() if err != nil { fmt.Printf("start producer error: %s", err.Error()) os.Exit(1) } topic := "test" msg := &primitive.Message{ Topic: topic, Body: []byte("Hello RocketMQ Go Client! " + strconv.Itoa(i)), } // 打tag msg.WithTag("tagA") // broker会针对key，建立索引存储在indexFile中 msg.WithKeys([]string{"key1", "key2"}) msg.WithProperty("age", "10") msg.WithProperty("color", "red") res, err := p.SendSync(context.Background(), msg) ... } Message中一些重要的属性 ...

版权声明 本站原创文章 由 萌叔 发表 转载请注明 萌叔 | http://vearne.cc 1.引言 nsq的架构简单，代码清晰。对于自主造轮子，实现消息队列、消息推送系统、IM都是非常好的参考。 本文将以图表的形式来说明消息在nsdd中的流转。 2.消息流转 2.1 重要的数据结构 Topic/ Channel/ Client都是nsqd中的数据结构。 数据会从Topic复制到Channel1和Channel2 Client是 consumer在nsqd内部的表征 每个Client最多只能订阅一个Channel 它们内部都有一组queue memoryMsgChan chan *Message // 位于内存 backend BackendQueue //当memoryMsgChan写满，则默认写入磁盘 每个producer都有1个读协程，负责把producer发送的消息写入Topic 多个Client可以订阅同一个Channel。即一个Channel,多个消费者，谁抢到算谁的。 每一个Topic都对应1个协程Topic.messagePump负责从Topic复制数据到Channel NSQD.queueScanLoop会控制一组协程NSQD.queueScanWorker(动态大小的协程池) 从Channel(所有topic的Channels)中复制数据到Client中, 之所以是协程池，我觉得可能跟Channel中的消息有延迟发送，且有重入队列的有关操作 每一个Client对应着一个协程protocolV2.messagePump，负责通过TCP连接把数据发送给consumer 3. 其它有趣的小细节 比较有意思的是，nsq官方推荐，nsqd随着发布者一起部署。 发布者不必去发现其他的nsqd节点，他们总是可以向本地实例发布消息。 实际上解放了producer，而甩锅给了consumer，如果某个Topic 假定叫topic1。如果topic1位于多个nsqd，consumer需要通过nsqlookupd获知所有拥topic1的nsqd的地址，然后需要在多个nsqd上订阅topic1 这里的nsqlookupd相当于是注册中心。 如果某个nsqd宕机，由于nsqd没有副本，消息可能会丢失 打赏作者

版权声明 本站原创文章 由 萌叔 发表 转载请注明 萌叔 | http://vearne.cc 环境 软件版本 NSQ v1.2.0 golang 1.12 go-nsq github.com/nsqio/go-nsq v1.0.7 1.引言 NSQ 是一个基于 Go 语言的分布式实时消息平台，它具有分布式、去中心化的拓扑结构，该结构具有无单点故障、故障容错、高可用性以及能够保证消息的可靠传递的特征。它的吞吐能力非常强，在单实例4核CPU可以达到19wop/s。足以应付日常使用 (见参考资料1) 。 不过它没有内建的数据副本机制（如果某个nsqd宕机，则该nsqd上的消息会丢失），且默认数据也不持久化（需要将mem_queue_size修改为0），使用NSQ仍然需要谨慎。 当我们考查一个消息队列，除了吞吐能力和可靠性，安全性也是重要的考察点。NSQ默认提供了鉴权机制(见查考资料2) 2. 使用鉴权 2.1 启用鉴权&对鉴权服务的要求 启动鉴权 nsqd --auth-http-address=nsqauth.example.com:4181 --http-address=127.0.0.1:4151 NSQ把权限的管理委托给了外部的鉴权服务，通过auth-http-address指定 官方给的鉴权服务的参考代码是 jehiah/nsqauth-contrib 笔者跟据它的思路重新实现了Golang的版本 vearne/go-nsqauthd NSQ要求鉴权服务能够对如下的HTTP进行响应 /auth?secret=xxxx&remote_ip=39.156.69.79&tls=true nsqd会拿着客户端的信息到鉴权服务进行鉴权 secret是客户端提供的秘钥(相当于redis中的密码) remote_ip是客户端的IP地址 tls表示客户端与nsqd之间的通讯是否有TLS保护 鉴权返回形如如下的信息，表明此用户可以操作的topic { "identity": "xxxx", "ttl": 3600, "Authorizations": [ { "channels": [ ".*" ], "topic": "topic1", "permissions": [ "subscribe", // 有topic1上的订阅权限 "publish" // 有topic1上的发布权限 ] } ] } vearne/go-nsqauthd为了简单期间，将秘钥直接存储在了csv文件中。 login ip tls_required topic channel subscribe publish 4BFE467B-FCBA-4519-BAC8-E9A3C57EDEB6 false test .* subscribe publish test_local 127.0.0.1 true local .* subscribe publish test_publish 127.0.0.1/24 false test_publish .* publish 其中login对应secret是秘钥信息, ip可以是某个子网, 如果不写表示没有ip限制，channel和topic都支持通配符的表达式。 ...