Sample code used to connect RocketMQ 1.x TCP client SDK for Java to an ApsaraMQ for RocketMQ 5.x instance to send and receive messages - ApsaraMQ for RocketMQ

ApsaraMQ for RocketMQ 5.x örnekleri, Java için RocketMQ 1.x TCP istemci SDK'sını destekler. Aşağıdaki örnek kodlar dört mesaj türünü kapsar: normal, sıralı, zamanlanmış ve gecikmeli ile işlemeli mesajlar.

Önemli

En son RocketMQ 5.x SDK'ları 5.x aracılarla tamamen uyumludur ve daha fazla özellik sunar. Yeni projelerinizde bunları kullanın. Daha fazla bilgi için bkz. Sürüm notları. Alibaba Cloud yalnızca RocketMQ 3.x, 4.x ve TCP istemci SDK'larını sürdürmektedir. Bunları yalnızca mevcut iş yükleri için kullanın.

Kod dizini

İhtiyacınız olan koda hızlıca erişmek için aşağıdaki tabloyu kullanın.

Mesaj türü	Gönder	Al
Normal	Eşzamanlı	Push
Normal	Asenkron	Toplu push
Normal	Tek yönlü	Pull
Normal	Çok iş parçacıklı
Sıralı	Gönder	Abone ol
Zamanlanmış	Gönder	Normal ile aynı
Gecikmeli	Gönder	Normal ile aynı
İşlemeli	Gönder	Normal ile aynı

Önkoşullar

Başlamadan önce aşağıdakileri yaptığınızdan emin olun:

Konsolda uç nokta, konu ve tüketici gruplarının oluşturulduğu bir ApsaraMQ for RocketMQ 5.x örneği
Projenize eklenmiş Java için RocketMQ 1.x TCP istemci SDK'sı. Daha fazla bilgi için bkz. Hazırlıklar

Ortak yapılandırma

Tüm örnekler aşağıda gösterilen aynı bağlantı ve kimlik doğrulama kurulumunu paylaşır. Sonraki her kod bloğu bu standart kodu atlar ve yalnızca mesaja özel mantığa odaklanır.

import com.aliyun.openservices.ons.api.PropertyKeyConst;
import java.util.Properties;

Properties properties = new Properties();

// Instance credentials
// Obtain the username and password on the Intelligent Authentication tab
// of the Access Control page for your instance in the ApsaraMQ for RocketMQ console.
// Do not use your Alibaba Cloud account AccessKey pair.
properties.put(PropertyKeyConst.AccessKey, "<instance-username>");
properties.put(PropertyKeyConst.SecretKey, "<instance-password>");

// Endpoint
// Enter the domain name and port from the console (e.g., rmq-cn-XXXX.rmq.aliyuncs.com:8080).
// Do not add an http:// or https:// prefix. Do not use a resolved IP address.
properties.put(PropertyKeyConst.NAMESRV_ADDR, "<endpoint>");

// Send timeout in milliseconds
properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");

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

Yer tutucu	Açıklama	Örnek
`<instance-username>`	Konsoldan alınan örnek kullanıcı adı	LTAI5tXxx
`<instance-password>`	Konsoldan alınan örnek parolası	xXxXxXx
`<endpoint>`	Konsoldan alınan örnek uç noktası	rmq-cn-XXXX.rmq.aliyuncs.com:8080

Kimlik doğrulama kuralları

Erişim yöntemi	Kimlik bilgisi gerekir?
Genel uç nokta	Evet. `AccessKey` ve `SecretKey` ayarlayın
Elastic Compute Service (ECS) örneğinden Sanal Özel Bulut (VPC)	Hayır. Aracı, kimlik bilgilerini VPC bilgilerinden otomatik olarak alır
İnternet üzerinden sunucusuz örnek	Evet
VPC içindeki sunucusuz örnek (kimlik doğrulamasız etkinleştirilmiş)	Hayır

5.x örneğine bağlanmak için TCP istemci SDK'sı kullanırken örnek kimliğini ayarlamayın. Ayarlanması bağlantı hatalarına neden olur.

İnternet üzerinden sunucusuz örnekler

Bir sunucusuz örneğe internet üzerinden erişiyorsanız SDK sürümü 1.9.0.Final veya üstünü kullanın ve aşağıdaki özelliği ekleyin:

properties.setProperty(PropertyKeyConst.Namespace, "<instance-id>");

<instance-id> ifadesini ApsaraMQ for RocketMQ örnek kimliğinizle değiştirin.

Normal mesajları gönderme ve alma

Normal mesajları eşzamanlı olarak gönderme

Eşzamanlı gönderim, aracı mesajı onaylayana kadar engeller. Teslimat onayının önemli olduğu durumlarda bu modu kullanın.

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

import java.util.Date;
import java.util.Properties;

public class ProducerTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // ... common configuration (see above) ...

        Producer producer = ONSFactory.createProducer(properties);
        // Call start() once before sending any messages.
        producer.start();

        for (int i = 0; i < 100; i++) {
            Message msg = new Message(
                    "TopicTestMQ",           // Topic created in the console.
                                             // A normal message topic cannot send or receive other message types.
                    "TagA",                  // Tag for consumer-side filtering.
                    "Hello MQ".getBytes());  // Message body (binary). Producer and consumer must
                                             // agree on serialization and deserialization methods.

            // Business key. Should be globally unique when possible.
            // Use it to look up messages in the console if delivery issues occur.
            msg.setKey("ORDERID_" + i);

            try {
                SendResult sendResult = producer.send(msg);
                // No exception means the message was sent successfully.
                if (sendResult != null) {
                    System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
                }
            } catch (Exception e) {
                // Implement retry or persistence logic here.
                System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                e.printStackTrace();
            }
        }

        // Shut down the producer before exiting the application.
        // Skip this if the application sends messages continuously.
        producer.shutdown();
    }
}

Normal mesajları asenkron olarak gönderme

Asenkron gönderim hemen döner ve sonucu bir geri arama aracılığıyla iletir. Çağrı iş parçacığının engellenmemesi gereken ve daha yüksek aktarım hızı gerektiren durumlarda bu modu kullanın.

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.OnExceptionContext;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.SendCallback;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class ProducerTest {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();
        // ... common configuration (see above) ...

        Producer producer = ONSFactory.createProducer(properties);
        producer.start();

        Message msg = new Message(
                "TopicTestMQ",
                "TagA",
                "Hello MQ".getBytes());

        msg.setKey("ORDERID_100");

        // sendAsync returns immediately. The result is delivered through the callback.
        producer.sendAsync(msg, new SendCallback() {
            @Override
            public void onSuccess(final SendResult sendResult) {
                System.out.println("send message success. topic=" + sendResult.getTopic() + ", msgId=" + sendResult.getMessageId());
            }

            @Override
            public void onException(OnExceptionContext context) {
                // Implement retry or persistence logic here.
                System.out.println("send message failed. topic=" + context.getTopic() + ", msgId=" + context.getMessageId());
            }
        });

        // Wait for the async callback before shutting down.
        TimeUnit.SECONDS.sleep(3);

        producer.shutdown();
    }
}

Normal mesajları tek yönlü modda gönderme

Tek yönlü gönderim, aracı yanıtını beklemeden mesajı ateşler. Bu mod en yüksek aktarım hızını sağlar ancak teslimat garantisi sunmaz. Yalnızca günlük toplama gibi ara sıra mesaj kaybına tolerans gösteren senaryolarda kullanın.

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

import java.util.Properties;

public class ProducerTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // ... common configuration (see above) ...

        Producer producer = ONSFactory.createProducer(properties);
        producer.start();

        for (int i = 0; i < 100; i++) {
            Message msg = new Message(
                    "TopicTestMQ",
                    "TagA",
                    "Hello MQ".getBytes());

            msg.setKey("ORDERID_" + i);

            // sendOneway does not wait for a broker response.
            // If delivery confirmation matters, use synchronous or asynchronous sending instead.
            producer.sendOneway(msg);
        }

        producer.shutdown();
    }
}

Normal mesajları birden çok iş parçacığından gönderme

Üretici iş parçacığı açısından güvenlidir. Her iş parçacığı için yeni bir tane oluşturmak yerine tek bir üretici örneğini iş parçacıkları arasında paylaşın.

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;

import java.util.Date;
import java.util.Properties;

public class SharedProducer {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, "XXX");
        // ... common configuration (see above) ...

        Producer producer = ONSFactory.createProducer(properties);
        producer.start();

        // Both threads share the same producer instance.
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                Message msg = new Message(
                        "TopicTestMQ",
                        "TagA",
                        "Hello MQ".getBytes());
                try {
                    SendResult sendResult = producer.send(msg);
                    if (sendResult != null) {
                        System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
                    }
                } catch (Exception e) {
                    System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                    e.printStackTrace();
                }
            }
        });
        thread.start();

        Thread anotherThread = new Thread(new Runnable() {
            @Override
            public void run() {
                Message msg = new Message("TopicTestMQ", "TagA", "Hello MQ".getBytes());
                try {
                    SendResult sendResult = producer.send(msg);
                    if (sendResult != null) {
                        System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
                    }
                } catch (Exception e) {
                    System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                    e.printStackTrace();
                }
            }
        });
        anotherThread.start();

        // (Optional) Shut down the producer when it is no longer needed.
        // producer.shutdown();
    }
}

Push modunda normal mesajlara abone olma

Push modunda aracı, mesajlar geldikçe bunları tüketicilere iletir. Bu en yaygın tüketim modelidir.

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

import java.util.Properties;

public class ConsumerTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, "XXX");
        // ... common configuration (see above) ...

        // Consumption mode (optional):
        // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);    // Default
        // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);

        Consumer consumer = ONSFactory.createConsumer(properties);

        // Subscribe to multiple tags with "||".
        consumer.subscribe("TopicTestMQ", "TagA||TagB", new MessageListener() {
            public Action consume(Message message, ConsumeContext context) {
                System.out.println("Receive: " + message);
                return Action.CommitMessage;
            }
        });

        // Subscribe to another topic. Use "*" to receive all tags.
        // To unsubscribe, remove the subscription code and restart the consumer.
        consumer.subscribe("TopicTestMQ-Other", "*", new MessageListener() {
            public Action consume(Message message, ConsumeContext context) {
                System.out.println("Receive: " + message);
                return Action.CommitMessage;
            }
        });

        consumer.start();
        System.out.println("Consumer Started");
    }
}

Toplu push modunda normal mesajlara abone olma

Toplu push modu, her geri arama çağrısında birden çok mesaj iletir ve mesaj başına ek yükü azaltır.

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.batch.BatchConsumer;
import com.aliyun.openservices.ons.api.batch.BatchMessageListener;
import java.util.List;
import java.util.Properties;

import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

public class SimpleBatchConsumer {

    public static void main(String[] args) {
        Properties consumerProperties = new Properties();
        consumerProperties.put(PropertyKeyConst.GROUP_ID, "XXX");
        // ... common configuration (see above) ...

        // Maximum number of messages per batch. Valid values: 1 to 1024. Default: 32.
        consumerProperties.setProperty(PropertyKeyConst.ConsumeMessageBatchMaxSize, String.valueOf(128));
        // Maximum wait time between batches in seconds. Valid values: 0 to 450. Default: 0.
        consumerProperties.setProperty(PropertyKeyConst.BatchConsumeMaxAwaitDurationInSeconds, String.valueOf(10));

        BatchConsumer batchConsumer = ONSFactory.createBatchConsumer(consumerProperties);
        batchConsumer.subscribe("TopicTestMQ", "TagA", new BatchMessageListener() {

             @Override
            public Action consume(final List<Message> messages, ConsumeContext context) {
                System.out.printf("Batch-size: %d\n", messages.size());
                // Process multiple messages at a time.
                return Action.CommitMessage;
            }
        });

        batchConsumer.start();
        System.out.println("Consumer start success.");

        // Keep the process alive.
        try {
            Thread.sleep(200000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Pull modunda normal mesajlara abone olma

Pull modunda tüketici, mesajları ne zaman alacağını kontrol eder. Tüketim hızı üzerinde hassas kontrole ihtiyaç duyduğunuz durumlarda bu modu kullanın.

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.PullConsumer;
import com.aliyun.openservices.ons.api.TopicPartition;
import java.util.List;
import java.util.Properties;
import java.util.Set;

public class PullConsumerClient {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-xxxxx");
        // ... common configuration (see above) ...

        PullConsumer consumer = ONSFactory.createPullConsumer(properties);
        consumer.start();

        // Get all partitions for the topic.
        Set<TopicPartition> topicPartitions = consumer.topicPartitions("topic-xxx");
        // Assign partitions to pull from.
        consumer.assign(topicPartitions);

        while (true) {
            // Poll with a 3000 ms timeout.
            List<Message> messages = consumer.poll(3000);
            System.out.printf("Received message: %s %n", messages);
        }
    }
}

Sıralı mesajları gönderme ve alma

Sıralı mesajlar, aynı parçalama anahtarına sahip mesajların gönderildikleri sırayla tüketilmesini garanti eder. Sipariş durumu güncellemeleri gibi katı sıralama gerektiren senaryolarda kullanın.

Sıralı mesajları gönderme

İlgili mesajları aynı parçaya yönlendirmek için OrderProducer kullanın ve bir parçalama anahtarı iletin.

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.order.OrderProducer;

import java.util.Date;
import java.util.Properties;

public class ProducerClient {

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, "XXX");
        // ... common configuration (see above) ...

        OrderProducer producer = ONSFactory.createOrderProducer(properties);
        producer.start();

        for (int i = 0; i < 1000; i++) {
            String orderId = "biz_" + i % 10;
            Message msg = new Message(
                    "Order_global_topic",              // Topic for ordered messages.
                    "TagA",
                    "send order global msg".getBytes()
            );
            msg.setKey(orderId);

            // Sharding key determines partition routing.
            // Messages with the same sharding key are delivered to the same partition
            // and consumed in order.
            String shardingKey = String.valueOf(orderId);
            try {
                SendResult sendResult = producer.send(msg, shardingKey);
                if (sendResult != null) {
                    System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
                }
            } catch (Exception e) {
                System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                e.printStackTrace();
            }
        }

        producer.shutdown();
    }
}

Sıralı mesajlara abone olma

OrderConsumer ile MessageOrderListener kullanın. Tüketim başarısız olursa aracı mesajı sırayı bozmadan yeniden denemesi için OrderAction.Suspend döndürün.

package com.aliyun.openservices.ons.example.order;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.order.ConsumeOrderContext;
import com.aliyun.openservices.ons.api.order.MessageOrderListener;
import com.aliyun.openservices.ons.api.order.OrderAction;
import com.aliyun.openservices.ons.api.order.OrderConsumer;

import java.util.Properties;

public class ConsumerClient {

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, "XXX");
        // ... common configuration (see above) ...

        // Retry interval (ms) if consumption fails. Valid values: 10 to 30000.
        properties.put(PropertyKeyConst.SuspendTimeMillis, "100");
        // Maximum retry count for failed messages.
        properties.put(PropertyKeyConst.MaxReconsumeTimes, "20");

        OrderConsumer consumer = ONSFactory.createOrderedConsumer(properties);

        consumer.subscribe(
                "Order_global_topic",
                // Tag filter:
                // "*"           -> subscribe to all tags
                // "TagA||TagB"  -> subscribe to TagA or TagB
                "*",
                new MessageOrderListener() {
                    @Override
                    public OrderAction consume(Message message, ConsumeOrderContext context) {
                        System.out.println(message);
                        // Return OrderAction.Suspend if consumption fails or an exception occurs.
                        return OrderAction.Success;
                    }
                });

        consumer.start();
    }
}

Zamanlanmış mesajları gönderme ve alma

Zamanlanmış mesajlar hemen gönderilmek yerine mutlak bir teslimat zamanı belirtir. Kullanıcı kaydından 24 saat sonra hatırlatma gönderme gibi belirli bir anda gerçekleşmesi gereken görevler için kullanın.

Zamanlanmış mesajları gönderme

startDeliverTime değerini milisaniye cinsinden bir Unix zaman damgasına ayarlayın. Aracı mesajı bu zamanda iletir. Zaman damgası geçmişte ise aracı mesajı hemen iletir.

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;

public class ProducerDelayTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // ... common configuration (see above) ...

        Producer producer = ONSFactory.createProducer(properties);
        producer.start();

        Message msg = new Message(
                "Topic",
                "tag",
                "Hello MQ".getBytes());
        msg.setKey("ORDERID_100");

        try {
            // Deliver at a specific time. Replace with your target timestamp.
            long timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2016-03-07 16:21:00").getTime();
            msg.setStartDeliverTime(timeStamp);

            SendResult sendResult = producer.send(msg);
            System.out.println("Message Id:" + sendResult.getMessageId());
        } catch (Exception e) {
            System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
            e.printStackTrace();
        }

        producer.shutdown();
    }
}

Zamanlanmış mesajlara abone olma

Zamanlanmış mesajlara normal mesajlarla aynı şekilde abone olun. Bkz. Push modunda normal mesajlara abone olma.

Gecikmeli mesajları gönderme ve alma

Gecikmeli mesajlar mutlak bir teslimat zamanı yerine mevcut zamana göre göreceli bir gecikme belirtir. Başarısız bir işlemi bekleme süresi sonrasında yeniden deneme gibi senaryolar için kullanın.

Gecikmeli mesajları gönderme

startDeliverTime değerini System.currentTimeMillis() artı istenen gecikme süresi (milisaniye) olarak ayarlayın. Maksimum gecikme 40 gündür (3.456.000.000 ms).

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;

import java.util.Date;
import java.util.Properties;

public class ProducerDelayTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // ... common configuration (see above) ...

        Producer producer = ONSFactory.createProducer(properties);
        producer.start();

        Message msg = new Message(
                "Topic",
                "tag",
                "Hello MQ".getBytes());
        msg.setKey("ORDERID_100");

        try {
            // Deliver 3 seconds from now.
            long delayTime = System.currentTimeMillis() + 3000;
            msg.setStartDeliverTime(delayTime);

            SendResult sendResult = producer.send(msg);
            if (sendResult != null) {
                System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
            }
        } catch (Exception e) {
            System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
            e.printStackTrace();
        }

        producer.shutdown();
    }
}

Gecikmeli mesajlara abone olma

Gecikmeli mesajlara normal mesajlarla aynı şekilde abone olun. Bkz. Push modunda normal mesajlara abone olma.

İşlemeli mesajları gönderme ve alma

İşlemeli mesajlar, yerel bir işlemi ve mesaj teslimatını birbirine bağlar; böylece ikisi de bir bütün olarak başarılı olur veya başarısız olur. İş akışı şu şekildedir:

Aracı yarım mesaj gönderir.
Uygulamanız yerel işlemi çalıştırır.
İşlem sonucuna göre aracı mesajı commit eder veya geri alır.

Aracı zamanında bir commit veya geri alma almazsa işlem durumunu sorgulamak için bir denetleyici geri araması çağırır.

İşlemeli mesajlar özel bir tüketici grubu gerektirir. Grubu diğer mesaj türleriyle paylaşmayın.

İşlemeli mesajları gönderme

İki arabirim uygulayın:

LocalTransactionExecuter: Aracı yarım mesajı kabul ettikten sonra yerel işleminizi çalıştırır.
LocalTransactionChecker: İşlem sonucu bilinmediğinde aracı durum denetimlerine yanıt verir.

package com.aliyun.openservices.tcp.example.producer;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import com.aliyun.openservices.ons.api.transaction.LocalTransactionChecker;
import com.aliyun.openservices.ons.api.transaction.LocalTransactionExecuter;
import com.aliyun.openservices.ons.api.transaction.TransactionProducer;
import com.aliyun.openservices.ons.api.transaction.TransactionStatus;

import java.util.Date;
import java.util.Properties;

public class SimpleTransactionProducer {

    public static void main(String[] args) {

        Properties properties = new Properties();
        // Transactional messages require a dedicated group ID.
        properties.put(PropertyKeyConst.GROUP_ID, "XXX");
        // ... common configuration (see above) ...

        // Register the transaction status checker before creating the producer.
        LocalTransactionCheckerImpl localTransactionChecker = new LocalTransactionCheckerImpl();
        TransactionProducer transactionProducer = ONSFactory.createTransactionProducer(properties, localTransactionChecker);
        transactionProducer.start();

        Message msg = new Message("XXX", "TagA", "Hello MQ transaction===".getBytes());

        for (int i = 0; i < 3; i++) {
            try {
                SendResult sendResult = transactionProducer.send(msg, new LocalTransactionExecuter() {
                    @Override
                    public TransactionStatus execute(Message msg, Object arg) {
                        // Run the local transaction here.
                        System.out.println("Execute the local transaction and commit the transaction status.");
                        return TransactionStatus.CommitTransaction;
                    }
                }, null);
                assert sendResult != null;
            } catch (ONSClientException e) {
                System.out.println(new Date() + " Send mq message failed! Topic is:" + msg.getTopic());
                e.printStackTrace();
            }
        }

        System.out.println("Send transaction message success.");
    }
}

// Transaction status checker.
// The broker calls check() when it does not receive a commit or rollback.
class LocalTransactionCheckerImpl implements LocalTransactionChecker {

    @Override
    public TransactionStatus check(Message msg) {
        System.out.println("The request to check the transaction status of the message is received. MsgId: " + msg.getMsgID());
        return TransactionStatus.CommitTransaction;
    }
}

İşlemeli mesajlara abone olma

İşlemeli mesajlara normal mesajlarla aynı şekilde abone olun. Bkz. Push modunda normal mesajlara abone olma.

Sonraki adımlar

Mesaj teslimatı sorunlarını gidermeye yardımcı olması için istemci tarafı günlüğünü ayarlayın. Bkz. Günlük yapılandırmaları.
Daha fazla özellik ve daha iyi uyumluluk için en son RocketMQ 5.x SDK'larına geçiş yapın. Bkz. Sürüm notları.