[GitHub] SpriderMan opened a new issue #452: why consumer can't get messages?

[GitBox <gi...@apache.org>]

SpriderMan opened a new issue #452: why consumer can't get messages?
URL: https://github.com/apache/rocketmq/issues/452
 
 
   
   package com.weimai.rocket.filter;
   
   import java.io.IOException;
   
   import org.apache.rocketmq.client.exception.MQBrokerException;
   import org.apache.rocketmq.client.exception.MQClientException;
   import org.apache.rocketmq.client.producer.DefaultMQProducer;
   import org.apache.rocketmq.client.producer.SendResult;
   import org.apache.rocketmq.common.message.Message;
   import org.apache.rocketmq.remoting.exception.RemotingException;
   
   public class Producer {
       
       public static void main(String[] args) throws IOException {
           try {
               DefaultMQProducer producer = new DefaultMQProducer("sequence_producer");
    
               producer.setNamesrvAddr("127.0.0.1:9876");
    
               producer.start();
               // 订单列表
               for (int i = 0; i < 20; i++) {
                   // 加个时间后缀
   
                   Message msg = new Message("SequenceTopicTest", // topic
                                             "TagA", // tag
                                             ("Hello RocketMQ " + i).getBytes("utf-8")// body
                   );
                   msg.putUserProperty("SequenceId", String.valueOf(i));// 设置过了参数
                   // 调用producer的send()方法发送消息
                   // 这里调用的是同步的方式，所以会有返回结果
                   SendResult sendResult = producer.send(msg);
                   System.out.println(i + "" + sendResult.getSendStatus());
               }
   
               producer.shutdown();
   
           } catch (MQClientException e) {
               e.printStackTrace();
           } catch (RemotingException e) {
               e.printStackTrace();
           } catch (MQBrokerException e) {
               e.printStackTrace();
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
       }
   }
   
   
   
   package com.weimai.rocket.filter;
   
   import java.io.IOException;
   import java.util.List;
   
   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.common.MixAll;
   import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
   import org.apache.rocketmq.common.message.MessageExt;
   public class Consumer {
   
       public static void main(String[] args) throws MQClientException, IOException {
           DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
           consumer.setNamesrvAddr("127.0.0.1:9876");
           /**
            * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
            * 如果非第一次启动，那么按照上次消费的位置继续消费
            */
           consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
   
   
           // String filterCode =
           // MixAll.file2String("E:\\workspace2\\rocketmq\\src\\main\\java\\com\\gwd\\rocketmq\\MyMessageFilter.java");
           String filterCode = MixAll.file2String("C:\\a_zf_secret\\MyMessageFilterImpl.java");
   
           consumer.subscribe("SequenceTopicTest", "com.weimai.rocket.filter.MyMessageFilterImpl", filterCode);
   
           consumer.registerMessageListener(new MessageListenerConcurrently() {
   
               @Override
               public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                   System.out.print(Thread.currentThread().getName() + " Receive New Messages: ");
                   for (MessageExt msg : msgs) {
                       System.out.println("content:" + new String(msg.getBody()));
                   }
   
                   return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
               }
           });
           consumer.start();
           System.out.println("Consumer Started.");
       }
   
   }
   
   
   
   package com.weimai.rocket.filter;
   
   import org.apache.rocketmq.common.filter.FilterContext;
   import org.apache.rocketmq.common.filter.MessageFilter;
   import org.apache.rocketmq.common.message.MessageExt;
   
   public class MyMessageFilterImpl implements MessageFilter {
   
       @Override
       public boolean match(MessageExt msg, FilterContext context) {
           String property = msg.getProperty("SequenceId");
           if (property != null) {
               int id = Integer.parseInt(property);
               if (((id % 2) == 0)) {
                   return true;
               }
           }
   
           return false;
       }
   }
   

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services