diff --git a/example/src/main/java/org/apache/rocketmq/example/quickstart/Consumer.java b/example/src/main/java/org/apache/rocketmq/example/quickstart/Consumer.java index 00e8fda9199a800bf86da4838f6c31b06f3f42e5..58910abc9f272f22b37a5cfcd85668cf91275d0a 100644 --- a/example/src/main/java/org/apache/rocketmq/example/quickstart/Consumer.java +++ b/example/src/main/java/org/apache/rocketmq/example/quickstart/Consumer.java @@ -25,15 +25,39 @@ import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.common.consumer.ConsumeFromWhere; import org.apache.rocketmq.common.message.MessageExt; +/** + * This example shows how to subscribe and consume messages using providing {@link DefaultMQPushConsumer}. + */ public class Consumer { public static void main(String[] args) throws InterruptedException, MQClientException { + + /* + * Instantiate with specified consumer group name. + */ DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4"); + /* + * Specify name server addresses. + *
+ * + * Alternatively, you may specify name server addresses via exporting environmental variable: NAMESRV_ADDR + */ + consumer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876"); + + /* + * Specify where to start in case the specified consumer group is a brand new one. + */ consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); + /* + * Subscribe one more more topics to consume. + */ consumer.subscribe("TopicTest", "*"); + /* + * Register callback to execute on arrival of messages fetched from brokers. + */ consumer.registerMessageListener(new MessageListenerConcurrently() { @Override @@ -44,7 +68,11 @@ public class Consumer { } }); + /* + * Launch the consumer instance. + */ consumer.start(); + System.out.printf("Consumer Started.%n"); } } diff --git a/example/src/main/java/org/apache/rocketmq/example/quickstart/Producer.java b/example/src/main/java/org/apache/rocketmq/example/quickstart/Producer.java index 27be8150a66b459b2601290cb3adffb06bcd95ed..06e1665e5b6f6008a83518033ba5dbd956884610 100644 --- a/example/src/main/java/org/apache/rocketmq/example/quickstart/Producer.java +++ b/example/src/main/java/org/apache/rocketmq/example/quickstart/Producer.java @@ -18,36 +18,60 @@ package org.apache.rocketmq.example.quickstart; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.producer.DefaultMQProducer; -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.common.message.Message; import org.apache.rocketmq.remoting.common.RemotingHelper; +/** + * This class demonstrates how to send messages to brokers using provided {@link DefaultMQProducer}. + */ public class Producer { public static void main(String[] args) throws MQClientException, InterruptedException { + + /* + * Instantiate with a producer group name. + */ DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name"); + + /* + * Specify name server addresses. + * + * + * Alternatively, you may specify name server addresses via exporting environmental variable: NAMESRV_ADDR + */ + producer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876"); + + /* + * Launch the instance. + */ producer.start(); for (int i = 0; i < 1000; i++) { try { - Message msg = new Message("TopicTest", - "TagA", - ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) + + /* + * Create a message instance, specifying topic, tag and message body. + */ + Message msg = new Message("TopicTest" /* Topic */, + "TagA" /* Tag */, + ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */ ); + + /* + * Call send message to deliver message to one of brokers. + */ SendResult sendResult = producer.send(msg); - LocalTransactionExecuter tranExecuter = new LocalTransactionExecuter() { - @Override - public LocalTransactionState executeLocalTransactionBranch(Message msg, Object arg) { - return null; - } - }; + System.out.printf("%s%n", sendResult); } catch (Exception e) { e.printStackTrace(); Thread.sleep(1000); } } + + /* + * Shut down once the producer instance is not longer in use. + */ producer.shutdown(); } }