ApsaraMQ for RocketMQ:Subscribe to messages

// Clustering is the default mode. To set it explicitly:
properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);

// Set the subscription mode to broadcasting.
properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);

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();
       // The consumer group ID created in the ApsaraMQ for RocketMQ console.
       properties.put(PropertyKeyConst.GROUP_ID, "<your-group-id>");
       // Read AccessKey credentials from environment variables.
       properties.put(PropertyKeyConst.AccessKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
       properties.put(PropertyKeyConst.SecretKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
       // The TCP endpoint from the Instance Details page in the console.
       properties.put(PropertyKeyConst.NAMESRV_ADDR, "<your-tcp-endpoint>");
       // Clustering mode (default). Uncomment the line below to switch to broadcasting.
       // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
       // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);

       Consumer consumer = ONSFactory.createConsumer(properties);
       // Subscribe to specific tags. Use "||" as a separator.
       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 with 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");
   }
}

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();
        // The consumer group ID created in the ApsaraMQ for RocketMQ console.
        consumerProperties.setProperty(PropertyKeyConst.GROUP_ID, MqConfig.GROUP_ID);
        // Read AccessKey credentials from environment variables.
        consumerProperties.put(PropertyKeyConst.AccessKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
        consumerProperties.put(PropertyKeyConst.SecretKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        // The TCP endpoint from the Instance Details page in the console.
        consumerProperties.setProperty(PropertyKeyConst.NAMESRV_ADDR, MqConfig.NAMESRV_ADDR);

        // Set the maximum batch size to 128 messages.
        consumerProperties.setProperty(PropertyKeyConst.ConsumeMessageBatchMaxSize, String.valueOf(128));
        // Set the maximum wait time between batches to 10 seconds.
        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 the batch consumer.
        batchConsumer.start();
        System.out.println("Consumer start success.");

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

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();
        // The consumer group ID created in the ApsaraMQ for RocketMQ console.
        properties.setProperty(PropertyKeyConst.GROUP_ID, "<your-group-id>");
        // Read AccessKey credentials from environment variables.
        properties.put(PropertyKeyConst.AccessKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
        properties.put(PropertyKeyConst.SecretKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        // The TCP endpoint from the Instance Details page in the console.
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "<your-tcp-endpoint>");
        PullConsumer consumer = ONSFactory.createPullConsumer(properties);
        // Start the pull consumer.
        consumer.start();
        // Get all partitions for the topic.
        Set<TopicPartition> topicPartitions = consumer.topicPartitions("topic-xxx");
        // Assign partitions to this consumer.
        consumer.assign(topicPartitions);

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