ApsaraMQ for RocketMQ:Batch consumption

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;
import com.aliyun.openservices.tcp.example.MqConfig;

public class SimpleBatchConsumer {

    public static void main(String[] args) {
        Properties consumerProperties = new Properties();
        consumerProperties.setProperty(PropertyKeyConst.GROUP_ID, MqConfig.GROUP_ID);
        consumerProperties.setProperty(PropertyKeyConst.AccessKey, MqConfig.ACCESS_KEY);
        consumerProperties.setProperty(PropertyKeyConst.SecretKey, MqConfig.SECRET_KEY);
        consumerProperties.setProperty(PropertyKeyConst.NAMESRV_ADDR, MqConfig.NAMESRV_ADDR);

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

        BatchConsumer batchConsumer = ONSFactory.createBatchConsumer(consumerProperties);
        batchConsumer.subscribe(MqConfig.TOPIC, MqConfig.TAG, new BatchMessageListener() {

             @Override
            public Action consume(final List<Message> messages, ConsumeContext context) {
                System.out.printf("Batch-size: %d\n", messages.size());
                // Process messages in batches.
                return Action.CommitMessage;
            }
        });
        // Start BatchConsumer.
        batchConsumer.start();
        System.out.println("Consumer start success.") ;

        // Wait for a fixed period to prevent the process from exiting.
        try {
            Thread.sleep(200000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Toplu iş boyutunu ve bekleme süresini birlikte ayarlayın

Dağıtım tetikleyicisi, toplu iş boyutu veya bekleme süresi eşiklerinden herhangi biri aşıldığında devreye girer. İş yükünüzle uyumlu olması için her iki parametreyi de uygun şekilde ayarlayın:

• Yüksek aktarım hızı gerektiren senaryolar -- ConsumeMessageBatchMaxSize değerini büyük bir sayıya (örneğin, 128 veya 256) ve BatchConsumeMaxAwaitDurationInSeconds değerini kısa bir aralığa (örneğin, 1--5 saniye) ayarlayın. Bu yapı, tam bir toplu işin oluşmasını beklemeden sık sık dağıtım yapılmasını sağlar.

• Düşük aktarım hızı gerektiren senaryolar -- Çok küçük toplu işlerin dağıtılmasını önlemek için orta düzeyde bir toplu iş boyutu (örneğin, 32) ve daha uzun bir bekleme süresi (örneğin, 10--30 saniye) belirleyin.

Örnek: ConsumeMessageBatchMaxSize değeri 128 ve BatchConsumeMaxAwaitDurationInSeconds değeri 1 olarak ayarlandığında, 128'den az mesaj birikmiş olsa bile 1 saniye sonra bir toplu iş dağıtımı gerçekleşir. Bu durumda, geri arama yöntemindeki messages.size() değeri 128'den küçük döner.

Tüketim idempotensini uygulayın

Daha iyi bir toplu tüketim sağlamak için, bir mesajın yalnızca bir kez işlendiğinden emin olmak amacıyla tüketici istemcinizde mesaj idempotensi uygulayın. Daha fazla bilgi için bkz. Tüketim idempotensi.