分布式消息服务ROCKETMQ版-收发普通消息:订阅普通消息

时间：2024-10-23 17:18:48

订阅普通消息

参考如下示例代码（以下加粗内容需要替换为实例自有信息，请根据实际情况替换）。

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/apache/rocketmq-clients/golang"
    "github.com/apache/rocketmq-clients/golang/credentials"
)

const (
    Topic     = "topic01"
    GroupName = "groupname"
    Endpoint  = "192.168.xx.xx:8080"
    AccessKey = os.Getenv("ROCKETMQ_AK")  //用户名和密钥直接硬编码到代码中或者明文存储都存在很大的风险，建议在配置文件或者环境变量中密文存放,使用时解密。
    SecretKey = os.Getenv("ROCKETMQ_SK")
)

var (
    // 接收消息请求的最大等待时间
    awaitDuration = time.Second * 5
    // 每次能接收的最大消息数
    maxMessageNum int32 = 16
    // 消息不可见时间，在消息被接收后对其他消费者不可见，直到超时。
    invisibleDuration = time.Second * 20 
)

func main() {
    os.Setenv("mq.consoleAppender.enabled", "true")
    golang.ResetLogger()
    simpleConsumer, err := golang.NewSimpleConsumer(&golang.Config{
        Endpoint: Endpoint,
        Group:    GroupName,
        Credentials: &credentials.SessionCredentials{
            AccessKey:    AccessKey,
            AccessSecret: SecretKey,
        },
    },
        golang.WithAwaitDuration(awaitDuration),
        golang.WithSubscriptionExpressions(map[string]*golang.FilterExpression{
            Topic: golang.SUB_ALL,
        }),
    )
    if err != nil {
        log.Fatal(err)
    }
    err = simpleConsumer.Start()
    if err != nil {
        log.Fatal(err)
    }
    defer simpleConsumer.GracefulStop()

    go func() {
        for {
            fmt.Println("start recevie message")
            mvs, err := simpleConsumer.Receive(context.TODO(), maxMessageNum, invisibleDuration)
            if err != nil {
                fmt.Println(err)
            }
            for _, mv := range mvs {
                simpleConsumer.Ack(context.TODO(), mv)
                fmt.Println(mv)
            }
            fmt.Println("wait a moment")
            fmt.Println()
            time.Sleep(time.Second * 3)
        }
    }()

    time.Sleep(time.Minute)
}

示例代码中的参数说明如下，请参考收集连接信息获取参数值。