ApsaraMQ for RocketMQ:Send and receive normal messages

package main

import (
    "fmt"
    "os"
    "strconv"
    "time"

    "github.com/aliyunmq/mq-http-go-sdk"
)

func main() {
    endpoint := "<your-http-endpoint>"
    accessKey := os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    secretKey := os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    topic := "<your-topic>"
    instanceId := "<your-instance-id>"

    client := mq_http_sdk.NewAliyunMQClient(endpoint, accessKey, secretKey, "")
    producer := client.GetProducer(instanceId, topic)

    for i := 0; i < 4; i++ {
        msg := mq_http_sdk.PublishMessageRequest{
            MessageBody: "hello mq!",
            MessageTag:  "",                  // Optional: filter tag
            Properties:  map[string]string{}, // Optional: custom properties
        }
        msg.MessageKey = "MessageKey"
        msg.Properties["a"] = strconv.Itoa(i)

        ret, err := producer.PublishMessage(msg)
        if err != nil {
            fmt.Println(err)
            return
        }
        fmt.Printf("Publish ---->\n\tMessageId:%s, BodyMD5:%s, \n",
            ret.MessageId, ret.MessageBodyMD5)

        time.Sleep(100 * time.Millisecond)
    }
}

package main

import (
    "fmt"
    "os"
    "strings"
    "time"

    "github.com/aliyunmq/mq-http-go-sdk"
    "github.com/gogap/errors"
)

func main() {
    endpoint := "<your-http-endpoint>"
    accessKey := os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    secretKey := os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    topic := "<your-topic>"
    instanceId := "<your-instance-id>"
    groupId := "<your-group-id>"

    client := mq_http_sdk.NewAliyunMQClient(endpoint, accessKey, secretKey, "")
    consumer := client.GetConsumer(instanceId, topic, groupId, "")

    for {
        endChan := make(chan int)
        respChan := make(chan mq_http_sdk.ConsumeMessageResponse)
        errChan := make(chan error)

        go func() {
            select {
            case resp := <-respChan:
                // Process the batch
                var handles []string
                fmt.Printf("Consume %d messages---->\n", len(resp.Messages))
                for _, v := range resp.Messages {
                    handles = append(handles, v.ReceiptHandle)
                    fmt.Printf("\tMessageID: %s, PublishTime: %d, MessageTag: %s\n"+
                        "\tConsumedTimes: %d, FirstConsumeTime: %d, NextConsumeTime: %d\n"+
                        "\tBody: %s\n"+
                        "\tProps: %s\n",
                        v.MessageId, v.PublishTime, v.MessageTag, v.ConsumedTimes,
                        v.FirstConsumeTime, v.NextConsumeTime, v.MessageBody, v.Properties)
                }

                // Acknowledge processed messages
                ackerr := consumer.AckMessage(handles)
                if ackerr != nil {
                    fmt.Println(ackerr)
                    if errAckItems, ok := ackerr.(errors.ErrCode).Context()["Detail"].([]mq_http_sdk.ErrAckItem); ok {
                        for _, errAckItem := range errAckItems {
                            fmt.Printf("\tErrorHandle:%s, ErrorCode:%s, ErrorMsg:%s\n",
                                errAckItem.ErrorHandle, errAckItem.ErrorCode, errAckItem.ErrorMsg)
                        }
                    } else {
                        fmt.Println("ack err =", ackerr)
                    }
                    time.Sleep(3 * time.Second)
                } else {
                    fmt.Printf("Ack ---->\n\t%s\n", handles)
                }
                endChan <- 1

            case err := <-errChan:
                if strings.Contains(err.(errors.ErrCode).Error(), "MessageNotExist") {
                    fmt.Println("\nNo new message, continue!")
                } else {
                    fmt.Println(err)
                    time.Sleep(3 * time.Second)
                }
                endChan <- 1

            case <-time.After(35 * time.Second):
                fmt.Println("Timeout of consumer message ??")
                endChan <- 1
            }
        }()

        // Long polling: wait up to 3 seconds for new messages, fetch up to 3 per batch
        consumer.ConsumeMessage(respChan, errChan,
            3, // Max messages per batch (up to 16)
            3, // Long polling wait time in seconds (up to 30)
        )
        <-endChan
    }
}

Uzun oylama (Long Polling) çalışma mekanizması

Uzun oylama modunda, mevcut bir mesaj yoksa istek, belirttiğiniz süre boyunca (en fazla 30 saniye) broker üzerinde bekletilir. Bir mesaj geldiğinde broker anında yanıt verir. Bu yöntem, kısa oylamaya kıyasla boş yanıtları azaltır. Varsayılan ağ zaman aşımı 35 saniyedir.

Onaylama ve yeniden teslimat

Bir mesajı işledikten sonra, tüketimi onaylamak için makbuz tanıtıcısı ile birlikte AckMessage çağrısını yapın. Broker, NextConsumeTime son tarihinden önce bir onay almazsa mesajı yeniden teslim eder. Her yeniden teslimatta yeni bir makbuz tanıtıcısı atanır.