Use scenarios and usage notes of normal messages in ApsaraMQ for RocketMQ - ApsaraMQ for RocketMQ

Normal mesajlar, ApsaraMQ for RocketMQ'deki varsayılan mesaj türüdür. Sıralı, zamanlanmış, geciktirilmiş ve işlem tabanlı mesajların aksine normal mesajlarda özel özellikler bulunmaz. Normal mesajlar, üreticiler ile tüketiciler arasında eşzamansız ve gevşek bağlı iletişim kurmayı destekler. Uygulamanız güvenilir teslimat gerektiriyor ancak sıkı bir sıralama veya zamanlamalı teslimata ihtiyaç duymuyorsa normal mesajları kullanın.

Normal mesajların ne zaman kullanılacağı

Normal mesajlar, işleme sırasından ziyade aktarım güvenilirliğine öncelik veren tüm senaryolar için uygundur. Yaygın örnekler arasında mikroservislerin birbirinden ayrılması, veri entegrasyonu ve olay güdümlü mimariler yer alır:

Mikroservislerin birbirinden ayrılması: Yukarı akıştaki bir hizmet, sipariş oluşturma ve ödeme işlemlerini bağımsız bir normal mesaj olarak kapsüller ve bu mesajı ApsaraMQ for RocketMQ broker'ına gönderir. Aşağı akıştaki hizmetler bağımsız şekilde abone olur ve mesajı kendi mantıklarına göre işler. Her mesaj kendi içinde bütündür; mesajlar arası bağımlılık bulunmaz.
Veri entegrasyonu: Bir izleme bileşeni, ön uç uygulamalarındaki işlem günlüklerini toplar ve bunları ApsaraMQ for RocketMQ'ye iletir. Her mesaj ayrı bir günlük verisi kaydıdır. Broker, bu kayıtları dönüştürmeden aşağı akıştaki depolama sistemlerine saklar ve teslim eder. Sonraki görevler arka uç uygulamaları tarafından işlenir.

Mesaj yaşam döngüsü

Bir normal mesaj, oluşturulması ile silinmesi arasında beş aşamadan geçer:

Lifecycle of a normal message

Aşama	Açıklama
Başlatıldı	Üretici mesajı oluşturur ve gönderim için hazırlar.
Hazır	Broker mesajı kabul eder. Mesaj artık görünürdür ve tüketiciler tarafından kullanılabilir durumdadır.
İşlemde	Bir tüketici mesajı alır ve işlemeye başlar. Broker belirli bir süre içinde onay almazsa teslimatı yeniden dener.
Onaylandı	Tüketici bir tüketim sonucu bildirir. Broker, mevcut mesajın tüketilip tüketilmediğini işaretler ancak mesajı hemen silmez.
Silindi	Mesaj, saklama süresi dolduğunda veya depolama alanı azaldığında kaldırılır. Silme işlemi, fiziksel dosyadan en eski mesajları önce kaldıran döngüsel bir stratejiyi izler. Ayrıntılar için bkz. Mesaj depolama ve temizleme.

Varsayılan olarak, ApsaraMQ for RocketMQ onaylanan mesajlar dahil tüm mesajları saklamaya devam eder. Tüketiciler, henüz silinmemiş herhangi bir mesajı yeniden tüketebilir.

Kod örnekleri (Java)

Aşağıdaki örneklerde normal mesajları göndermek ve tüketmek için Apache RocketMQ 5.x SDK kullanılır. Her örnekte bir istemci bağlantısı yapılandırılır ve tek bir işlem gerçekleştirilir. Kodu çalıştırmadan önce yer tutucuları gerçek değerlerinizle değiştirin.

Tam SDK başvurusu ve diğer dillerdeki örnekler için bkz. Apache RocketMQ 5.x SDK'leri.

Başlamadan önce

ApsaraMQ for RocketMQ konsolunda aşağıdaki kaynakları hazırlayın:

Kaynak	Nerede bulunur
Uç nokta	Instance Details sayfası > Endpoints sekmesi. Aynı sanal özel cloud (VPC) içindeki Elastic Compute Service (ECS) örneklerinden erişim için VPC uç noktasını kullanın. İnternet erişimi için genel uç noktayı kullanın (internet erişimi özelliğinin etkinleştirilmiş olması gerekir).
Konu	`MessageType` değeri `Normal` olarak ayarlanmış bir konu oluşturun. Normal mesajlar yalnızca bu türdeki konulara gönderilebilir.
Tüketici grubu	Tüketici örnekleri için gereklidir. Konsolda bir tane oluşturun.
Kimlik bilgileri (yalnızca genel uç nokta)	Kullanıcı adı ve parolayı Access Control sayfası > Intelligent Authentication sekmesinden edinin. Broker'ın otomatik olarak kimlik doğrulaması yaptığı VPC erişimi için gerekli değildir. VPC'lerde kimlik doğrulamasız erişim özelliği etkinleştirilmiş bir sunucusuz örneğe VPC üzerinden erişiliyorsa kimlik bilgileri yine gerekli değildir.

Yer tutucu referansı

Yer tutucu	Açıklama	Örnek
`<your-endpoint>`	Örnek uç noktası	`rmq-cn-xxx.cn-hangzhou.rmq.aliyuncs.com:8080`
`<your-topic>`	Konu adı	`normal-topic-01`
`<your-consumer-group>`	Tüketici grubu adı	`GID_normal_consumer`
`<your-username>`	Örnek kullanıcı adı (yalnızca genel uç nokta)	--
`<your-password>`	Örnek parolası (yalnızca genel uç nokta)	--

Örnek kod

Normal mesaj gönderme

package doc;

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;

public class ProducerExample {
    public static void main(String[] args) throws ClientException {
        // Specify the instance endpoint.
        String endpoints = "<your-endpoint>";
        String topic = "<your-topic>";

        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        // Uncomment the following line for public endpoint access:
        // builder.setCredentialProvider(new StaticSessionCredentialsProvider("<your-username>", "<your-password>"));
        ClientConfiguration configuration = builder.build();

        Producer producer = provider.newProducerBuilder()
                .setTopics(topic)
                .setClientConfiguration(configuration)
                .build();

        // Build a normal message with a key and tag for filtering and tracing.
        Message message = provider.newMessageBuilder()
                .setTopic(topic)
                .setKeys("messageKey")
                .setTag("messageTag")
                .setBody("messageBody".getBytes())
                .build();

        try {
            SendReceipt sendReceipt = producer.send(message);
            System.out.println("Send success. Message ID: " + sendReceipt.getMessageId());
        } catch (ClientException e) {
            e.printStackTrace();
        }
    }
}

Beklenen çıktı:

Send success. Message ID: 01BE0A3600F5762D0455025C2D1A0000

Push tüketici ile tüketme

Push tüketici, bir MessageListener geri arama aracılığıyla mesajları alır. Broker, mesajları tüketiciye otomatik olarak iter.

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collections;

public class PushConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(PushConsumerExample.class);

    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        String endpoints = "<your-endpoint>";
        String topic = "<your-topic>";
        String consumerGroup = "<your-consumer-group>";

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        // Uncomment the following line for public endpoint access:
        // builder.setCredentialProvider(new StaticSessionCredentialsProvider("<your-username>", "<your-password>"));
        ClientConfiguration clientConfiguration = builder.build();

        // Subscribe to all messages in the topic (tag = "*").
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .setMessageListener(messageView -> {
                    System.out.println("Received message: " + messageView);
                    return ConsumeResult.SUCCESS;
                })
                .build();

        // Keep the process running to continue receiving messages.
        Thread.sleep(Long.MAX_VALUE);

        // Close the consumer when it is no longer needed.
        // pushConsumer.close();
    }
}

Beklenen çıktı:

Received message: MessageView{messageId=01BE0A3600F5762D0455025C2D1A0000, topic=<your-topic>, ...}

Basit tüketici ile tüketme

Basit bir tüketici, mesajları açıkça receive() yöntemiyle çeker ve her mesajı ack() ile onaylar. Bu mod, tüketim hızı üzerinde tam kontrol sağlar.

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
import org.apache.rocketmq.client.apis.message.MessageId;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.List;

public class SimpleConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(SimpleConsumerExample.class);

    public static void main(String[] args) throws ClientException, IOException {
        String endpoints = "<your-endpoint>";
        String topic = "<your-topic>";
        String consumerGroup = "<your-consumer-group>";

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        // Uncomment the following line for public endpoint access:
        // builder.setCredentialProvider(new StaticSessionCredentialsProvider("<your-username>", "<your-password>"));
        ClientConfiguration clientConfiguration = builder.build();

        Duration awaitDuration = Duration.ofSeconds(10);
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

        SimpleConsumer consumer = provider.newSimpleConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setAwaitDuration(awaitDuration)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .build();

        int maxMessageNum = 16;
        Duration invisibleDuration = Duration.ofSeconds(10);

        // Pull and acknowledge messages in a loop.
        // For real-time consumption, use multiple threads to pull concurrently.
        while (true) {
            final List<MessageView> messages = consumer.receive(maxMessageNum, invisibleDuration);
            messages.forEach(messageView -> {
                System.out.println("Received message: " + messageView);
            });
            for (MessageView message : messages) {
                final MessageId messageId = message.getMessageId();
                try {
                    consumer.ack(message);
                    System.out.println("Acked message: " + messageId);
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        }

        // Close the consumer when it is no longer needed.
        // consumer.close();
    }
}

Beklenen çıktı:

Received message: MessageView{messageId=01BE0A3600F5762D0455025C2D1A0000, topic=<your-topic>, ...}
Acked message: 01BE0A3600F5762D0455025C2D1A0000

En iyi uygulamalar

Her mesaj için genel olarak benzersiz bir mesaj anahtarı ayarlayın

Sipariş kimliği veya kullanıcı kimliği gibi işe özgü bir tanımlayıcıyı mesaj anahtarı olarak atayın. Bu sayede ApsaraMQ for RocketMQ konsolunda bireysel mesajları ve izlerini sorgulayabilirsiniz.

Message message = provider.newMessageBuilder()
        .setTopic(topic)
        .setKeys("order-12345")   // Use a unique business identifier.
        .setTag("payment")
        .setBody(payload)
        .build();

Doğru tüketici modunu seçin

Mod	Çalışma biçimi	En uygun kullanım alanı
Push tüketici	Broker, mesajları bir `MessageListener` geri aramasına iter.	Düşük gecikme süresi gerektiren, olay güdümlü işleme; tüketici her mesajı geldiğinde işler.
Basit tüketici	Uygulama, bir toplu işi çekmek için `receive()` yöntemini çağırır ve her mesajı işledikten sonra `ack()` ile onaylar.	El ile akış denetimi veya toplu işlem gerektiren iş yükleri.