逍遥阁

go语言操作阿里云RocketMQ

关键词: golang RocketMQ

go语言操作阿里云RocketMQ

```
package core

import (
    "github.com/apache/rocketmq-client-go/core"
    "sync"
)

type RocketMQService struct {
    producerConfig *rocketmq.ProducerConfig
    consumerConfig *rocketmq.PushConsumerConfig
}

type RocketMQTransactionLocalListener struct {
}

var rocketMQService *RocketMQService

var once sync.Once

func NewRocketMQService() *RocketMQService {
    once.Do(func() {
        rocketMQService = &RocketMQService{}
    })
    return rocketMQService
}

func (that *RocketMQService) SetProducerConfig(accessKey, secretKey, channel, groupId, nameServer string, producerModel rocketmq.ProducerModel) {
    that.producerConfig = &rocketmq.ProducerConfig{
        ClientConfig: rocketmq.ClientConfig{
            GroupID:    groupId,
            NameServer: nameServer,
            Credentials: &rocketmq.SessionCredentials{
                AccessKey: accessKey,
                SecretKey: secretKey,
                Channel:   channel,
            },
        },
        ProducerModel: producerModel,
    }
}

func (that *RocketMQService) SetConsumerConfig(accessKey, secretKey, channel, groupId, nameServer string) {
    that.consumerConfig = &rocketmq.PushConsumerConfig{
        ClientConfig: rocketmq.ClientConfig{
            GroupID:    groupId,
            NameServer: nameServer,
            Credentials: &rocketmq.SessionCredentials{
                AccessKey: accessKey,
                SecretKey: secretKey,
                Channel:   channel,
            },
        },
        Model:         rocketmq.Clustering,
        ConsumerModel: rocketmq.CoCurrently,
    }
}

func (that *RocketMQService) GetProducerConfig() *rocketmq.ProducerConfig {
    return that.producerConfig
}

func (that *RocketMQService) GetConsumerConfig() *rocketmq.PushConsumerConfig {
    return that.consumerConfig
}

func (that *RocketMQService) Send(topic, body string) (result *rocketmq.SendResult, err error) {
    listener := &RocketMQTransactionLocalListener{}
    producer, err := rocketmq.NewTransactionProducer(that.GetProducerConfig(), listener, nil)
    if err != nil {
        return
    }
    err = producer.Start()
    if err != nil {
        return
    }
    defer producer.Shutdown()
    result, err = producer.SendMessageTransaction(&rocketmq.Message{Topic: topic, Body: body}, nil)
    if err != nil {
        return
    }
    return
}

func (that *RocketMQService) Receive(topic string, cb func(message *rocketmq.MessageExt) rocketmq.ConsumeStatus) (err error) {
    consumer, err := rocketmq.NewPushConsumer(that.GetConsumerConfig())
    if err != nil {
        return
    }
    consumer.Subscribe(topic, "*", func(msg *rocketmq.MessageExt) rocketmq.ConsumeStatus {
        return cb(msg)
    })
    err = consumer.Start()
    if err != nil {
        return
    }
    var wg sync.WaitGroup
    wg.Add(1)
    wg.Wait()
    err = consumer.Shutdown()
    if err != nil {
        return
    }
    return
}

func (*RocketMQTransactionLocalListener) Execute(m *rocketmq.Message, arg interface{}) rocketmq.TransactionStatus {
    return rocketmq.CommitTransaction
}

func (*RocketMQTransactionLocalListener) Check(m *rocketmq.MessageExt, arg interface{}) rocketmq.TransactionStatus {
    return rocketmq.CommitTransaction
}
```

admin
2020-06-12 17:24:28
go

文章分类

文章标签

文章归档

友情链接

go语言操作阿里云RocketMQ