aliyun-mns
aliyun-mns是对阿里云消息服务的封装
队列模型
具有以下特点：
- 动态创建队列
- 可以设置消费者数目
- 消息处理时长自适应
- 发送消息重试。目前基于网络错误、阿里云MNS错误码表InternalError重试
- 监控报警
- 优雅的关闭消费者
- 处理函数处理最大时间限制
- 队列消费者使用协程池，每一个消息队列独占自己的协程池
- 发送消息失败保存进入redis，尝试重新发送，提高发送成功率
- 业务需要自己做消息幂等，有可能出现同样消息内容发送多次，这种情况非常罕见
消费者
package main
import (
"context"
"github.com/go-redis/redis/v7"
"github.com/xiaojiaoyu100/aliyun-mns/v2"
)
type Builder struct {
}
func (b *Builder) Handle(ctx context.Context) error {
//return alimns.BackoffError{
// Err: err,
// N: 30,
//}
//return err
//return nil
}
func Before(m *alimns.M) (context.Context, error) {
return context.TODO(), nil
}
func After(ctx context.Context) {
}
func main() {
option := &redis.Options{
Addr: "127.0.0.1:6379",
DB: 0,
}
redisClient := redis.NewClient(option)
client, err := alimns.NewClient(alimns.Config{
Cmdable: redisClient,
Endpoint: "",
QueuePrefix: "", // 可以留空，表示拉取全部消息队列
AccessKeyID: "",
AccessKeySecret: "",
})
if err != nil {
return
}
client.SetBefore(Before)
client.SetAfter(After)
consumer := alimns.NewConsumer(client)
err = consumer.AddQueue(
&alimns.Queue{
Name: "QueueTest1",
Builder: &Builder{},
},
)
if err != nil {
return
}
consumer.Run()
}生产者
producer := alimns.NewProducer(client)
producer.SendBase64EncodedJSONMessage()主题模型
支持以下主题api:
- 支持主题的创建，删除
- 支持订阅主题，取消主题订阅
- 支持向主题发布消息
创建/订阅主题
endpoint := QueueEndPoint{
AccountID: "xxx",
Region: "xxx",
QueueName: "xxx",
}
// 创建
err := client.CreateTopic("topicName")
if err != nil {
return
}
// 订阅
err = client.Subscribe("topicName", "subscriptionName", endpoint)
if err != nil {
return
}发布消息
messageID, err := client.PublishMessage("topicName", "hello world")
if err != nil {
return
}