Sample code for sending and receiving messages with the Apache RocketMQ Java SDK - ApsaraMQ for RocketMQ

Apache RocketMQ Java SDK kullanarak ApsaraMQ for RocketMQ ile mesaj gönderin ve alın. Bu sayfada hem gRPC protokolü SDK'sı (rocketmq-client-java) hem de Remoting protokolü SDK'sı (rocketmq-client) için örnekler yer almaktadır.

Ön koşullar

Başlamadan önce aşağıdakilere sahip olduğunuzdan emin olun:

ApsaraMQ for RocketMQ konsolunda oluşturulmuş konulara ve grup kimliklerine sahip bir ApsaraMQ for RocketMQ örneği
Örnek uç noktası (örneğin, rmq-cn-XXXX.rmq.aliyuncs.com:8080)
Projenize eklenmiş gerekli SDK bağımlılığı. Sürüm ayrıntıları için Sürüm kılavuzuna bakın

Önemli

Serverless örneği kullanıyorsanız, devam etmeden önce Genel ağ erişimi için SDK sürüm gereksinimlerini kontrol edin.

gRPC protokolü SDK'sı

rocketmq-client-java SDK'sı, gRPC protokolü üzerinden iletişim kurar. Aşağıdaki tablolarda, Apache RocketMQ istemcileri deposunda barındırılan her mesaj türü için örnek kodlar listelenmiştir.

Spring Boot entegrasyonu için rocketmq-v5-client-spring-boot-samples adresine bakın.

Önemli

gRPC protokolü SDK'sı ile işlemsel mesajlar gönderirken, üreticiyi başlatırken bir konu belirtin. Konu belirtilmezse işlem kontrolleri gecikir. Mesaj dört saat içinde gönderilmezse yarı işlemsel mesaj atılabilir.

Mesaj gönderme

Mesaj türü	Örnek kod
Normal mesaj (senkron)	ProducerNormalMessageExample.java
Normal mesaj (asenkron)	AsyncProducerExample.java
Sıralı mesaj	ProducerFifoMessageExample.java
Zamanlanmış veya gecikmeli mesaj	ProducerDelayMessageExample.java
İşlemsel mesaj	ProducerTransactionMessageExample.java
Hafif mesaj	LiteProducerExample.java

Mesaj tüketme

Tüketici türü	Örnek kod
Push tüketici	PushConsumerExample.java
Basit tüketici (senkron)	SimpleConsumerExample.java
Basit tüketici (asenkron)	AsyncSimpleConsumerExample.java
Hafif push tüketici	LitePushConsumerExample.java

Push tüketiciler ve basit tüketiciler hakkında daha fazla bilgi için Tüketici türleri bölümüne bakın.

Remoting protokolü SDK'sı

rocketmq-client SDK'sı, Remoting protokolü üzerinden iletişim kurar. Bu bölümde her mesaj türü için satır içi kodlar sağlanmıştır.

Spring Boot entegrasyonu için rocketmq-spring-boot-samples adresine bakın.

Ortak kurulum

Tüm Remoting protokolü örnekleri aynı kimlik doğrulama ve bağlantı kurulumunu paylaşır. Önce bu bölümü inceleyin, ardından aşağıdaki mesaj türüne özel kodlara bakın.

Kimlik doğrulama

Üreticiyi veya tüketiciyi nasıl başlatacağınız erişim yöntemine bağlıdır:

Genel uç nokta -- Örnek kullanıcı adı ve parolasıyla bir RPCHook iletin: Örnek kullanıcı adını ve parolasını Resource Access Management konsolunun Intelligent Identity Recognition sekmesinden alın. Alibaba Cloud hesabınızın AccessKey ID ve AccessKey secret bilgisini kullanmayın.
```
  private static RPCHook getAclRPCHook() {
      return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
  }

  // Pass the RPCHook when creating the producer
  DefaultMQProducer producer = new DefaultMQProducer(getAclRPCHook());
```
VPC uç noktası -- RPCHook gerekmez. Sunucu, VPC'ye göre kimlik doğrulaması yapar:
```
  DefaultMQProducer producer = new DefaultMQProducer();
```
Serverless örneği -- Genel ağ erişimi için bir RPCHook iletin. Dahili ağ üzerinden parolasız erişim etkinleştirildiyse RPCHook gerekmez.

Bağlantı

// Group ID created in the ApsaraMQ for RocketMQ console
producer.setProducerGroup("<your-group-id>");

// Endpoint from the ApsaraMQ for RocketMQ console
// Use the domain name and port as shown. Do not add http:// or https://.
// Do not use a resolved IP address.
producer.setNamesrvAddr("<your-access-point>");

Mesaj izleri (isteğe bağlı)

Cloud mesaj izlemeyi etkinleştirmek için erişim kanalını ayarlayın ve izleri etkinleştirin:

producer.setAccessChannel(AccessChannel.CLOUD);

// Required for SDK v5.3.0 and later, in addition to setAccessChannel
producer.setEnableTrace(true);

Yer tutucu referansı

Aşağıdaki yer tutucuları gerçek değerlerinizle değiştirin:

Yer tutucu	Açıklama	Örnek
`<instance-username>`	Konsoldan alınan örnek kullanıcı adı	--
`<instance-password>`	Konsoldan alınan örnek parolası	--
`<your-group-id>`	Konsolda oluşturulan grup kimliği	GID_example
`<your-access-point>`	Konsoldan alınan uç nokta	rmq-cn-XXXX.rmq.aliyuncs.com:8080
`<your-topic>`	Konsolda oluşturulan konu	topic_example
`<your-order-topic>`	Sıralı mesajlar için konu	order_topic_example
`<your-transaction-topic>`	İşlemsel mesajlar için konu	transaction_topic_example
`<your-transaction-group-id>`	İşlemsel mesajlar için özel grup kimliği	GID_transaction_example
`<your-message-tag>`	Filtreleme için mesaj etiketi	TagA

Normal mesajlar

Normal mesajlar, sıralama ve işlemsel garantilerin gerekmediği çoğu kullanım durumu için uygundur.

Normal mesaj gönderme (senkron)

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.Date;

public class RocketMQProducer {
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer(getAclRPCHook());
        producer.setProducerGroup("<your-group-id>");
        producer.setAccessChannel(AccessChannel.CLOUD);
        producer.setEnableTrace(true);
        producer.setNamesrvAddr("<your-access-point>");
        producer.start();

        for (int i = 0; i < 128; i++) {
            try {
                Message msg = new Message("<your-topic>",
                        "<your-message-tag>",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            } catch (Exception e) {
                System.out.println(new Date() + " Send mq message failed.");
                e.printStackTrace();
            }
        }

        // Shut down the producer before the application exits.
        // Note: Destroying the producer object saves system memory.
        // To send messages continuously, keep the producer running.
        producer.shutdown();
    }
}

Normal mesaj gönderme (asenkron)

Asenkron gönderim hemen döner ve aracı yanıt verdiğinde bir geri arama çağırır.

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.Date;
import java.util.concurrent.TimeUnit;

public class RocketMQAsyncProducer {
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer(getAclRPCHook());
        producer.setProducerGroup("<your-group-id>");
        producer.setAccessChannel(AccessChannel.CLOUD);
        producer.setEnableTrace(true);
        producer.setNamesrvAddr("<your-access-point>");
        producer.start();

        for (int i = 0; i < 128; i++) {
            try {
                Message msg = new Message("<your-topic>",
                        "<your-message-tag>",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult result) {
                        System.out.println("send message success. msgId= " + result.getMsgId());
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        System.out.println("send message failed.");
                        throwable.printStackTrace();
                    }
                });
            } catch (Exception e) {
                System.out.println(new Date() + " Send mq message failed.");
                e.printStackTrace();
            }
        }
        // Wait 3 seconds for the asynchronous callbacks to complete
        TimeUnit.SECONDS.sleep(3);

        producer.shutdown();
    }
}

Normal mesaj gönderme (tek yönlü)

Tek yönlü gönderimler, aracı yanıtını beklemeyen moddur. Günlük toplama gibi ara sıra mesaj kaybının kabul edilebilir olduğu yüksek aktarım hızı senaryolarında bu modu kullanın.

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.Date;

public class RocketMQOnewayProducer {
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer(getAclRPCHook());
        producer.setProducerGroup("<your-group-id>");
        producer.setAccessChannel(AccessChannel.CLOUD);
        producer.setEnableTrace(true);
        producer.setNamesrvAddr("<your-access-point>");
        producer.start();

        for (int i = 0; i < 128; i++) {
            try {
                Message msg = new Message("<your-topic>",
                        "<your-message-tag>",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                // sendOneway does not return a result or throw an exception on failure
                producer.sendOneway(msg);
            } catch (Exception e) {
                System.out.println(new Date() + " Send mq message failed.");
                e.printStackTrace();
            }
        }

        producer.shutdown();
    }
}

Normal mesajları tüketme

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;

import java.util.List;

public class RocketMQPushConsumer {
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(getAclRPCHook());
        consumer.setConsumerGroup("<your-group-id>");
        consumer.setAccessChannel(AccessChannel.CLOUD);
        consumer.setEnableTrace(true);
        consumer.setNamesrvAddr("<your-access-point>");

        // Subscribe to a topic. Use "*" to receive all tags, or specify a tag expression.
        consumer.subscribe("<your-topic>", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                System.out.printf("Receive New Messages: %s %n", msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }
}

Sıralı mesajlar

Sıralı mesajlar, aynı parçalama anahtarı içinde FIFO (ilk giren ilk çıkar) teslimatı garanti eder. Sıralı olay işleme veya gerçek zamanlı veri senkronizasyonu gibi senaryolar için kullanın.

Sıralı mesaj gönderme

Aynı parçalama anahtarına sahip mesajlar sırayla aynı kuyruğa teslim edilir. Aracının ilgili mesajları aynı kuyruğa yönlendirmesi için __SHARDINGKEY özelliğini ayarlayın.

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.List;

public class RocketMQOrderProducer {
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer(getAclRPCHook());
        producer.setProducerGroup("<your-group-id>");
        producer.setAccessChannel(AccessChannel.CLOUD);
        producer.setEnableTrace(true);
        producer.setNamesrvAddr("<your-access-point>");
        producer.start();

        for (int i = 0; i < 128; i++) {
            try {
                int orderId = i % 10;
                Message msg = new Message("<your-order-topic>",
                        "<your-message-tag>",
                        "OrderID188",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));

                // Set the sharding key to route related messages to the same queue.
                // In v5.x, you can replace the following line with:
                // msg.putUserProperty(MessageConst.PROPERTY_SHARDING_KEY, orderId + "");
                msg.putUserProperty("__SHARDINGKEY", orderId + "");

                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        // Select a queue based on the orderId
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        return mqs.get(index);
                    }
                }, orderId);

                System.out.printf("%s%n", sendResult);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        producer.shutdown();
    }
}

Sıralı mesajları tüketme

Her kuyruk içindeki tüketim sırasını korumak için MessageListenerConcurrently yerine MessageListenerOrderly kullanın.

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;

import java.util.List;

public class RocketMQOrderConsumer {
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(getAclRPCHook());
        consumer.setConsumerGroup("<your-group-id>");
        consumer.setAccessChannel(AccessChannel.CLOUD);
        consumer.setEnableTrace(true);
        consumer.setNamesrvAddr("<your-access-point>");
        consumer.subscribe("<your-order-topic>", "*");

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.registerMessageListener(new MessageListenerOrderly() {

            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                // Return SUSPEND_CURRENT_QUEUE_A_MOMENT on failure to retry
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

Zamanlanmış ve gecikmeli mesajlar

Zamanlanmış mesajlar belirli bir zamanda teslim edilir. Gecikmeli mesajlar yapılandırılabilir bir gecikmeden sonra teslim edilir. Her ikisi de __STARTDELIVERTIME kullanıcı özelliğini kullanır.

Zamanlanmış veya gecikmeli mesaj gönderme

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.Date;

public class RocketMQDelayProducer {
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer(getAclRPCHook());
        producer.setProducerGroup("<your-group-id>");
        producer.setAccessChannel(AccessChannel.CLOUD);
        producer.setEnableTrace(true);
        producer.setNamesrvAddr("<your-access-point>");
        producer.start();

        for (int i = 0; i < 128; i++) {
            try {
                Message msg = new Message("<your-topic>",
                        "<your-message-tag>",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));

                // Delayed message: deliver after 3 seconds
                long delayTime = System.currentTimeMillis() + 3000;
                msg.putUserProperty("__STARTDELIVERTIME", String.valueOf(delayTime));

                // Scheduled message: deliver at a specific time (format: yyyy-MM-dd HH:mm:ss)
                // If the specified time is earlier than the current time, the message is delivered immediately.
                // long timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2021-08-10 18:45:00").getTime();
                // msg.putUserProperty("__STARTDELIVERTIME", String.valueOf(timeStamp));

                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            } catch (Exception e) {
                System.out.println(new Date() + " Send mq message failed.");
                e.printStackTrace();
            }
        }

        producer.shutdown();
    }
}

Zamanlanmış ve gecikmeli mesajları tüketme

Zamanlanmış ve gecikmeli mesajları normal mesajlar ile aynı şekilde tüketin. Ek yapılandırma gerekmez.

İşlemsel mesajlar

İşlemsel mesajlar iki aşamalı taahhüt protokolü kullanır. Üretici yarım işlemsel bir mesaj gönderir, yerel bir işlem çalıştırır ve ardından taahhüt eder veya geri alır. Aracı, işlenmemiş işlemleri düzenli olarak checkLocalTransaction öğesini çağırarak kontrol eder.

Önemli

İşlemsel mesajlar için kullanılan grup kimliği diğer mesaj türleriyle paylaşılamaz. İşlemsel üreticiler için özel bir grup kimliği oluşturun.

İşlemsel mesaj gönderme

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class RocketMQTransactionProducer {

    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("<instance-username>", "<instance-password>"));
    }

    public static void main(String[] args) throws MQClientException {
        // Use a dedicated group ID for transactional messages
        TransactionMQProducer transactionMQProducer = new TransactionMQProducer("<your-transaction-group-id>", getAclRPCHook());
        transactionMQProducer.setAccessChannel(AccessChannel.CLOUD);
        transactionMQProducer.setEnableTrace(true);
        transactionMQProducer.setNamesrvAddr("<your-access-point>");

        transactionMQProducer.setTransactionListener(new TransactionListener() {

            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                // Run your local transaction logic here
                System.out.println("Start to execute the local transaction: " + msg);
                return LocalTransactionState.UNKNOW;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                // Return the local transaction status when the broker checks back
                System.out.println("Received a transaction check request, MsgId: " + msg.getMsgId());
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        transactionMQProducer.start();

        for (int i = 0; i < 10; i++) {
            try {
                Message message = new Message("<your-transaction-topic>",
                        "<your-message-tag>",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = transactionMQProducer.sendMessageInTransaction(message, null);
                assert sendResult != null;
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

İşlemsel mesajları tüketme

İşlemsel mesajları normal mesajlar ile aynı şekilde tüketin. Ek yapılandırma gerekmez.

Serverless örneği genel ağ erişimi

Bir Serverless örneğine genel ağ üzerinden erişmek için SDK'nızın minimum bir sürüm gereksinimini karşılaması gerekir. Aşağıda gösterilen ad alanı yapılandırmasını ekleyin.

Not

InstanceId ifadesini gerçek örnek kimliğinizle değiştirin.

Remoting protokolü SDK'sı (rocketmq-client >= 5.2.0)

Ad alanını üreticiye veya tüketiciye ekleyin:

// Producer
producer.setNamespaceV2("InstanceId");

// Consumer
consumer.setNamespaceV2("InstanceId");

gRPC protokolü SDK'sı (rocketmq-client-java >= 5.0.6)

ClientConfiguration içinde ad alanını ayarlayın:

ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
    .setEndpoints(endpoints)
    .setNamespace("InstanceId")
    .setCredentialProvider(sessionCredentialsProvider)
    .build();