li-xiao-shuang commented on code in PR #4353:
URL: https://github.com/apache/rocketmq/pull/4353#discussion_r878710088
##########
example/src/main/java/org/apache/rocketmq/example/tracemessage/OpenTracingPushConsumer.java:
##########
@@ -22,49 +22,49 @@
import io.opentracing.Tracer;
import io.opentracing.util.GlobalTracer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
-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.client.exception.MQClientException;
import org.apache.rocketmq.client.trace.hook.ConsumeMessageOpenTracingHookImpl;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
-import org.apache.rocketmq.common.message.MessageExt;
-
-import java.util.List;
public class OpenTracingPushConsumer {
+
+ public static final String CONSUMER_GROUP = "CID_JODIE_1";
+ public static final String DEFAULT_NAMESRVADDR = "127.0.0.1:9876";
+ public static final String TOPIC = "TopicTest";
+
public static void main(String[] args) throws InterruptedException,
MQClientException {
Tracer tracer = initTracer();
- DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer("CID_JODIE_1");
+ DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer(CONSUMER_GROUP);
+
+ // If the debugging source code can open comments, you need to set the
namesrvAddr to the address of the local namesrvAddr
+// consumer.setNamesrvAddr(DEFAULT_NAMESRVADDR);
consumer.getDefaultMQPushConsumerImpl().registerConsumeMessageHook(new
ConsumeMessageOpenTracingHookImpl(tracer));
- consumer.subscribe("TopicTest", "*");
+ consumer.subscribe(TOPIC, "*");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.setConsumeTimestamp("20181109221800");
- consumer.registerMessageListener(new MessageListenerConcurrently() {
-
- @Override
- public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt>
msgs, ConsumeConcurrentlyContext context) {
- System.out.printf("%s Receive New Messages: %s %n",
Thread.currentThread().getName(), msgs);
- return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
- }
+ consumer.registerMessageListener((MessageListenerConcurrently) (msgs,
context) -> {
+ System.out.printf("%s Receive New Messages: %s %n",
Thread.currentThread().getName(), msgs);
+ return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();
System.out.printf("Consumer Started.%n");
}
private static Tracer initTracer() {
Configuration.SamplerConfiguration samplerConfig =
Configuration.SamplerConfiguration.fromEnv()
- .withType(ConstSampler.TYPE)
Review Comment:
> Please do not change irrelevant code unless necessary
This PR submission includes code formatting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
