ApsaraMQ for RocketMQ 5.x örnekleri, RocketMQ 3.x ve 4.x SDK'ları ile oluşturulan istemcileri destekler. Bu sayfa, söz konusu eski SDK'ları kullanarak normal, sıralı, zamanlanmış, gecikmeli ve işlemli mesajların gönderilmesine ve alınmasına ilişkin Java örnekleri sunar.
En yeni RocketMQ 5.x SDK'ları, ApsaraMQ for RocketMQ 5.x broker'larla tam uyumludur ve ek özellikler sunar. Yeni projeler için 5.x SDK'larını kullanın. Ayrıntılar için bkz. Sürüm notları. 3.x, 4.x ve TCP istemci SDK'ları yalnızca mevcut iş yüklerini desteklemek amacıyla korunmaktadır.
Ön koşullar
Başlamadan önce aşağıdaki öğelere sahip olduğunuzdan emin olun:
Bir ApsaraMQ for RocketMQ 5.x örneği
ApsaraMQ for RocketMQ konsolunda oluşturulmuş bir konu ve tüketici grubu
Projenize eklenmiş Java için RocketMQ 3.x veya 4.x SDK'sı
Konsolda bulunan
rmq-cn-XXXX.rmq.aliyuncs.com:8080biçimindeki örnek erişim noktası
Tam kurulum talimatları için bkz. Hazırlıklar.
Paylaşılan yapılandırma
Bu sayfadaki tüm örnekler aynı kimlik doğrulama ve bağlantı kurulumunu kullanır. Aşağıdaki bloklar, her mesaj türü örneğinin gönderme veya alma mantığına odaklanabilmesi amacıyla paylaşılan yapılandırmayı tanımlar.
Kimlik doğrulama
ApsaraMQ for RocketMQ iki erişim yöntemini destekler. Ağ ortamınıza göre seçim yapın:
Genel uç nokta -- Örnek kullanıcı adı ve parolasıyla
RPCHooköğesini yapılandırın. Bu kimlik bilgilerini ApsaraMQ for RocketMQ konsolundaki Intelligent Authentication sekmesindeki Access Control sayfasından alın. Alibaba Cloud hesabınızın AccessKey çiftini kullanmayın.VPC uç noktası -- Kimlik bilgisi gerekmez. İstemci bir Virtual Private Cloud (VPC) içindeki bir Elastic Compute Service (ECS) örneğinde çalıştığında broker, kimlik bilgilerini VPC yapılandırmasından otomatik olarak alır.
Sunucusuz örnekler için genel uç nokta erişiminde kimlik bilgileri gereklidir. VPC'lerde kimlik doğrulamasız özelliğini etkinleştirirseniz VPC erişimi için kimlik bilgisi gerekmez.
RPCHook yardımcı programı
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.remoting.RPCHook;
/**
* Returns an RPCHook configured with instance credentials.
* Required for public endpoint access and serverless instances over the Internet.
* Not required for VPC access on standard instances.
*/
private static RPCHook getAclRPCHook() {
return new AclClientRPCHook(new SessionCredentials(
"<instance-username>", // From the Intelligent Authentication tab
"<instance-password>" // From the Intelligent Authentication tab
));
}Üretici başlatma
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
// --- Public endpoint: pass RPCHook ---
DefaultMQProducer producer = new DefaultMQProducer(getAclRPCHook());
// --- VPC endpoint: omit RPCHook ---
// DefaultMQProducer producer = new DefaultMQProducer();
// Group ID created in the console
producer.setProducerGroup("<your-group-id>");
// Required for message trace. Omit to disable tracing.
producer.setAccessChannel(AccessChannel.CLOUD);
// Access point from the console. Use the domain and port only -- no http:// prefix, no resolved IP.
producer.setNamesrvAddr("<your-access-point>");
producer.start();Tüketici başlatma
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
// --- Public endpoint: pass RPCHook ---
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(getAclRPCHook());
// --- VPC endpoint: omit RPCHook ---
// DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
consumer.setConsumerGroup("<your-group-id>");
consumer.setAccessChannel(AccessChannel.CLOUD);
consumer.setNamesrvAddr("<your-access-point>");
consumer.subscribe("<your-topic>", "*");Yer tutucu referansı
Aşağıdaki yer tutucuları gerçek değerlerinizle değiştirin:
| Yer Tutucu | Açıklama | Örnek |
|---|---|---|
<instance-username> |
Intelligent Authentication sekmesinden alınan örnek kullanıcı adı | MjY1NTI5MD... |
<instance-password> |
Intelligent Authentication sekmesinden alınan örnek parolası | OTk2QjFBMz... |
<your-group-id> |
Konsolda oluşturulan tüketici grubu kimliği | GID_test_group |
<your-access-point> |
Konsoldan alınan örnek erişim noktası (etkialanı:bağlantı_noktası) | rmq-cn-hangzhou.rmq.aliyuncs.com:8080 |
<your-topic> |
Konsolda oluşturulan konu | test_topic |
Normal mesajlar
Normal mesajlar eşzamanlı, eşzamansız ve tek yönlü olmak üzere üç gönderme modunu destekler. Kavramsal ayrıntılar için bkz. Normal mesajlar.
Normal mesajları eşzamanlı gönderme
Eşzamanlı gönderme, broker her mesajı onaylayana kadar işlemi engeller. Teslimat onayı gerektiğinde bu modu kullanın.
import java.util.Date;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
// After producer initialization (see Shared configuration)
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) {
// Handle send failure: retry or persist the message
System.out.println(new Date() + " Send mq message failed.");
e.printStackTrace();
}
}
// Shut down the producer before exiting.
// Skip shutdown if your application sends messages continuously.
producer.shutdown();Normal mesajları eşzamansız gönderme
Eşzamansız gönderme hemen yanıt döner ve sonuçları bir geri arama aracılığıyla iletir. Gecikme süresine duyarlı üreticiler için bu modu kullanın.
import java.util.Date;
import java.util.concurrent.TimeUnit;
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.common.RemotingHelper;
// After producer initialization (see Shared configuration)
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) {
// Handle send failure: retry or persist the message
System.out.println("send message failed.");
throwable.printStackTrace();
}
});
} catch (Exception e) {
System.out.println(new Date() + " Send mq message failed.");
e.printStackTrace();
}
}
// Wait for async callbacks to complete before shutting down
TimeUnit.SECONDS.sleep(3);
producer.shutdown();Normal mesajları tek yönlü modda gönderme
Tek yönlü gönderme, broker yanıtını beklemeden mesajı gönderir. Günlük toplama gibi ara sıra mesaj kaybının kabul edilebilir olduğu yüksek aktarım hızı gerektiren iş yükleri için bu modu kullanın.
import java.util.Date;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
// After producer initialization (see Shared configuration)
for (int i = 0; i < 128; i++) {
try {
Message msg = new Message(
"<your-topic>",
"<your-message-tag>",
"Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
producer.sendOneway(msg);
} catch (Exception e) {
System.out.println(new Date() + " Send mq message failed.");
e.printStackTrace();
}
}
producer.shutdown();Normal mesajları alma
Normal mesajlara eşzamanlı bir dinleyici ile abone olun. Her mesaj bağımsız olarak işlenir ve sıralama garanti edilmez.
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.common.message.MessageExt;
import java.util.List;
// After consumer initialization (see Shared configuration)
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ı mesaj kuyruğu içinde ilk giren ilk çıkar (FIFO) teslimatı garanti eder. Aynı parçalama anahtarına sahip mesajlar aynı kuyruğa yönlendirilir. Kavramsal ayrıntılar için bkz. Sıralı mesajlar.
Sıralı mesajları gönderme
Normal mesajlardan farklı olarak sıralı mesajlar, aynı parçalama anahtarına sahip mesajları aynı kuyruğa yönlendirmek için bir MessageQueueSelector gerektirir.
import java.util.List;
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.common.RemotingHelper;
// After producer initialization (see Shared configuration)
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 ensure ordered delivery.
// For ApsaraMQ for RocketMQ 5.x instances, an alternative is:
// 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) {
// Route messages with the same orderId to the same queue
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ı alma
MessageListenerConcurrently yerine MessageListenerOrderly kullanın. Broker, her kuyruk içindeki mesajları sırayla teslim eder.
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.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
// After consumer initialization (see Shared configuration)
// Subscribe to the ordered topic instead of a normal topic:
// 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 belirtilen bir zamanda teslim edilir. Gecikmeli mesajlar belirtilen bir gecikmeden sonra teslim edilir. Her ikisi de __STARTDELIVERTIME özelliğini kullanır. Kavramsal ayrıntılar için bkz. Zamanlanmış ve gecikmeli mesajlar.
Zamanlanmış veya gecikmeli mesajları gönderme
import java.util.Date;
import java.text.SimpleDateFormat;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
// After producer initialization (see Shared configuration)
for (int i = 0; i < 128; i++) {
try {
Message msg = new Message(
"<your-topic>",
"<your-message-tag>",
"Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
// Option 1: Delayed delivery -- deliver 3 seconds from now
long delayTime = System.currentTimeMillis() + 3000;
msg.putUserProperty("__STARTDELIVERTIME", String.valueOf(delayTime));
// Option 2: Scheduled delivery -- deliver at a specific time
// Format: yyyy-MM-dd HH:mm:ss
// If the time is in the past, 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ış veya gecikmeli mesajları alma
Tüketici kodu, Normal mesajları alma bölümündekiyle aynıdır. Tüketici tarafında özel bir yapılandırma gerekmez.
İşlemli mesajlar
İşlemli mesajlar, iki aşamalı taahhüt yoluyla dağıtık işlem desteği sağlar: üretici yarı mesaj gönderir, yerel bir işlem çalıştırır ve ardından işlem sonucuna göre mesajı taahhüt eder veya geri alır. Kavramsal ayrıntılar için bkz. İşlemli mesajlar.
İşlemli mesajlar özel bir tüketici grubu gerektirir. Grubu diğer mesaj türleriyle paylaşmayın.
İşlemli mesajları gönderme
Bu örnek bir yarı mesaj gönderir ve yerel bir işlem çalıştırır. Broker, üretici mesajı taahhüt edene veya geri alana kadar yarı mesajı tutar.
import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
// --- Public endpoint: pass RPCHook and a dedicated transaction group ID ---
TransactionMQProducer transactionMQProducer = new TransactionMQProducer(
"<your-transaction-group-id>", getAclRPCHook());
// --- VPC endpoint: omit RPCHook ---
// TransactionMQProducer transactionMQProducer = new TransactionMQProducer(
// "<your-transaction-group-id>");
transactionMQProducer.setAccessChannel(AccessChannel.CLOUD);
transactionMQProducer.setNamesrvAddr("<your-access-point>");
// Register the transaction status checker (see next section)
transactionMQProducer.setTransactionCheckListener(new LocalTransactionCheckerImpl());
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,
new LocalTransactionExecuter() {
@Override
public LocalTransactionState executeLocalTransactionBranch(
Message msg, Object arg) {
System.out.println("Start executing the local transaction: " + msg);
// Run your local transaction logic here.
// Return COMMIT_MESSAGE, ROLLBACK_MESSAGE, or UNKNOW.
return LocalTransactionState.UNKNOW;
}
},
null);
assert sendResult != null;
} catch (Exception e) {
e.printStackTrace();
}
}İşlem durumunu kontrol etme
Üretici UNKNOW döndürdüğünde broker, işlemi çözmek için düzenli aralıklarla checkLocalTransactionState öğesini çağırır. Yerel işlem durumunuzu sorgulamak ve nihai sonucu döndürmek için TransactionCheckListener arayüzünü uygulayın.
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
import org.apache.rocketmq.common.message.MessageExt;
public class LocalTransactionCheckerImpl implements TransactionCheckListener {
@Override
public LocalTransactionState checkLocalTransactionState(MessageExt msg) {
System.out.println("Transaction status check received. MsgId: " + msg.getMsgId());
// Query your local data store to determine the transaction outcome.
// Return COMMIT_MESSAGE, ROLLBACK_MESSAGE, or UNKNOW.
return LocalTransactionState.COMMIT_MESSAGE;
}
}İşlemli mesajları alma
Tüketici kodu, Normal mesajları alma bölümündekiyle aynıdır. Tüketici tarafında özel bir yapılandırma gerekmez.