rocketmq使用DefaultMQPushConsumer创建消费者客户端

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("organization");
consumer.setNamesrvAddr("172.22.0.64:9876"); // NAME_SERVER地址
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); // 从哪个位置开始消费消息
consumer.subscribe("my-topic", "*"); // 订阅主题
consumer.registerMessageListener(new MessageListenerOrderly() { // 注册消息监听(顺序方式)
    @Override
    public ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) {
        for (MessageExt msg : msgs) {
            System.out.println(new String(msg.getBody()));
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
});
consumer.start();

• ConsumeFromWhere 控制新的消费者组从哪个位置开始消费

注：如果使用CONSUME_FROM_TIMESTAMP ,需设置参数 DefaultMQPushConsumer.setConsumeTimestamp(“20131223171201”) 时间戳字符串格式为yyyyMMddHHmmss

• DefaultMQPushConsumer.subscribe(String topic, String subExpression) subExpression参数为tag选择表达式 语法：

1. 不过滤tag："*" 或者null
2. 根据多个tag过滤：“tag1 || tag2 || tag3”

• DefaultMQPushConsumer.registerMessageListener注册消费监听器

1. 顺序消费监听器MessageListenerOrderly 返回ConsumeOrderlyStatus

1. 并行消费监听器MessageListenerConcurrently 返回ConsumeConcurrentlyStatus

• org.apache.rocketmq.client.impl.consumer.RebalancePushImpl

switch (consumeFromWhere) {
    case CONSUME_FROM_LAST_OFFSET_AND_FROM_MIN_WHEN_BOOT_FIRST:
    case CONSUME_FROM_MIN_OFFSET:
    case CONSUME_FROM_MAX_OFFSET:
    case CONSUME_FROM_LAST_OFFSET: {
        long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);
        if (lastOffset >= 0) {
            result = lastOffset;
        }
        // First start,no offset
        else if (-1 == lastOffset) {
            if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                result = 0L;
            } else {
                try {
                    result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq);
                } catch (MQClientException e) {
                    result = -1;
                }
            }
        } else {
            result = -1;
        }
        break;
    }
    case CONSUME_FROM_FIRST_OFFSET: {
        long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);
        if (lastOffset >= 0) {
            result = lastOffset;
        } else if (-1 == lastOffset) {
            result = 0L;
        } else {
            result = -1;
        }
        break;
    }
    case CONSUME_FROM_TIMESTAMP: {
        long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);
        if (lastOffset >= 0) {
            result = lastOffset;
        } else if (-1 == lastOffset) {
            if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                try {
                    result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq);
                } catch (MQClientException e) {
                    result = -1;
                }
            } else {
                try {
                    long timestamp = UtilAll.parseDate(this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeTimestamp(),
                        UtilAll.YYYYMMDDHHMMSS).getTime();
                    result = this.mQClientFactory.getMQAdminImpl().searchOffset(mq, timestamp);
                } catch (MQClientException e) {
                    result = -1;
                }
            }
        } else {
            result = -1;
        }
        break;
    }

    default:
        break;
}