You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by GitBox <gi...@apache.org> on 2023/04/12 01:40:55 UTC

[GitHub] [rocketmq] jiakme created a discussion: single instance with multi DefaultLitePullConsumer(use same topic and different tag) don't poll message right

GitHub user jiakme created a discussion: single instance with multi DefaultLitePullConsumer(use same topic and different tag) don't poll message right

1. single java springboot server instance with multi DefaultLitePullConsumer(use the same topic and different tag)
2. invoke DefaultLitePullConsumer#poll() in loop
3. some messagequeues are not consumed



GitHub link: https://github.com/apache/rocketmq/discussions/6578

----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org

[GitHub] [rocketmq] jiakme added a comment to the discussion: single instance with multi DefaultLitePullConsumer(use same topic and different tag) don't poll message right

Posted by GitBox <gi...@apache.org>.
GitHub user jiakme added a comment to the discussion: single instance with multi DefaultLitePullConsumer(use same topic and different tag) don't poll message right

我测试了一下， 使用不同 consumer group 的情况, 消息消费还是有问题:
![image](https://user-images.githubusercontent.com/8815138/231365721-02987f71-e0b5-40ca-bf9d-5f758c058a52.png)


GitHub link: https://github.com/apache/rocketmq/discussions/6578#discussioncomment-5588427

----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org

[GitHub] [rocketmq] jiakme added a comment to the discussion: single instance with multi DefaultLitePullConsumer(use same topic and different tag) don't poll message right

Posted by GitBox <gi...@apache.org>.
GitHub user jiakme added a comment to the discussion: single instance with multi DefaultLitePullConsumer(use same topic and different tag) don't poll message right

rocketmq-dashboard:

![231166815-e57cf0e6-e79c-455d-a159-17e6894d2be1](https://user-images.githubusercontent.com/8815138/231326146-b4708423-68e4-4a9a-9536-3ce434006829.png)


GitHub link: https://github.com/apache/rocketmq/discussions/6578#discussioncomment-5587201

----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org

[GitHub] [rocketmq] RongtongJin added a comment to the discussion: single instance with multi DefaultLitePullConsumer(use same topic and different tag) don't poll message right

Posted by GitBox <gi...@apache.org>.
GitHub user RongtongJin added a comment to the discussion: single instance with multi DefaultLitePullConsumer(use same topic and different tag) don't poll message right

lite pull consumer和push consumer，需要保证订阅关系的一致

GitHub link: https://github.com/apache/rocketmq/discussions/6578#discussioncomment-5587290

----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org

[GitHub] [rocketmq] jiakme added a comment to the discussion: single instance with multi DefaultLitePullConsumer(use same topic and different tag) don't poll message right

Posted by GitBox <gi...@apache.org>.
GitHub user jiakme added a comment to the discussion: single instance with multi DefaultLitePullConsumer(use same topic and different tag) don't poll message right

`private void initRocketMQPushConsumer() throws MQClientException {

    List<DefaultLitePullConsumer> pullConsumers = new ArrayList<>();

    DefaultLitePullConsumer defaultLitePullConsumer = getDefaultLitePullConsumer("high");
    pullConsumers.add(defaultLitePullConsumer);

    DefaultLitePullConsumer defaultLitePullConsumer_1 = getDefaultLitePullConsumer("low");
    pullConsumers.add(defaultLitePullConsumer_1);

    PULL_CONSUMERS = Collections.unmodifiableList(pullConsumers);
}

private DefaultLitePullConsumer getDefaultLitePullConsumer(String tag) throws MQClientException {
    RocketMQProperties.Consumer consumerConfig = new RocketMQProperties.Consumer();
    consumerConfig.setGroup(MqConfig.CATEGORY_SMART_CAMPUS_APP_CENTER + "_priority");

    consumerConfig.setTopic(TOPIC);
    consumerConfig.setPullBatchSize(50);
    DefaultLitePullConsumer defaultLitePullConsumer;

    consumerConfig.setSelectorExpression(tag);
    defaultLitePullConsumer = buildPullConsumer(consumerConfig);
    return defaultLitePullConsumer;
}
private DefaultLitePullConsumer buildPullConsumer(RocketMQProperties.Consumer consumerConfig) throws MQClientException {
    /**
     * {@link org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration#defaultLitePullConsumer(RocketMQProperties)}
     */
    String nameServer = rocketMQProperties.getNameServer();
    String groupName = consumerConfig.getGroup();
    String topicName = consumerConfig.getTopic();
    Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
    Assert.hasText(groupName, "[rocketmq.pull-consumer.group] must not be null");
    Assert.hasText(topicName, "[rocketmq.pull-consumer.topic] must not be null");

    String accessChannel = rocketMQProperties.getAccessChannel();
    MessageModel messageModel = MessageModel.valueOf(consumerConfig.getMessageModel());
    SelectorType selectorType = SelectorType.valueOf(consumerConfig.getSelectorType());
    String selectorExpression = consumerConfig.getSelectorExpression();
    String ak = consumerConfig.getAccessKey();
    String sk = consumerConfig.getSecretKey();
    int pullBatchSize = consumerConfig.getPullBatchSize();
    boolean useTLS = consumerConfig.isTlsEnable();

    DefaultLitePullConsumer litePullConsumer = RocketMQUtil.createDefaultLitePullConsumer(nameServer, accessChannel,
            groupName, topicName, messageModel, selectorType, selectorExpression, ak, sk, pullBatchSize, useTLS);
    litePullConsumer.setEnableMsgTrace(consumerConfig.isEnableMsgTrace());
    litePullConsumer.setCustomizedTraceTopic(consumerConfig.getCustomizedTraceTopic());
    litePullConsumer.setNamespace(consumerConfig.getNamespace());
    litePullConsumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragely());
    litePullConsumer.setAutoCommit(true);
    return litePullConsumer;
}`

GitHub link: https://github.com/apache/rocketmq/discussions/6578#discussioncomment-5587162

----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org

[GitHub] [rocketmq] jiakme edited a comment on the discussion: single instance with multi DefaultLitePullConsumer(use same topic and different tag) don't poll message right

Posted by GitBox <gi...@apache.org>.
GitHub user jiakme edited a comment on the discussion: single instance with multi DefaultLitePullConsumer(use same topic and different tag) don't poll message right

generate DefaultLitePullConsumer instance

```java
private void initRocketMQPushConsumer() throws MQClientException {

    List<DefaultLitePullConsumer> pullConsumers = new ArrayList<>();

    DefaultLitePullConsumer defaultLitePullConsumer = getDefaultLitePullConsumer("high");
    pullConsumers.add(defaultLitePullConsumer);

    DefaultLitePullConsumer defaultLitePullConsumer_1 = getDefaultLitePullConsumer("low");
    pullConsumers.add(defaultLitePullConsumer_1);

    PULL_CONSUMERS = Collections.unmodifiableList(pullConsumers);
}

private DefaultLitePullConsumer getDefaultLitePullConsumer(String tag) throws MQClientException {
    RocketMQProperties.Consumer consumerConfig = new RocketMQProperties.Consumer();
    consumerConfig.setGroup(MqConfig.CATEGORY_SMART_CAMPUS_APP_CENTER + "_priority");

    consumerConfig.setTopic(TOPIC);
    consumerConfig.setPullBatchSize(50);
    DefaultLitePullConsumer defaultLitePullConsumer;

    consumerConfig.setSelectorExpression(tag);
    defaultLitePullConsumer = buildPullConsumer(consumerConfig);
    return defaultLitePullConsumer;
}
private DefaultLitePullConsumer buildPullConsumer(RocketMQProperties.Consumer consumerConfig) throws MQClientException {
    /**
     * {@link org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration#defaultLitePullConsumer(RocketMQProperties)}
     */
    String nameServer = rocketMQProperties.getNameServer();
    String groupName = consumerConfig.getGroup();
    String topicName = consumerConfig.getTopic();
    Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
    Assert.hasText(groupName, "[rocketmq.pull-consumer.group] must not be null");
    Assert.hasText(topicName, "[rocketmq.pull-consumer.topic] must not be null");

    String accessChannel = rocketMQProperties.getAccessChannel();
    MessageModel messageModel = MessageModel.valueOf(consumerConfig.getMessageModel());
    SelectorType selectorType = SelectorType.valueOf(consumerConfig.getSelectorType());
    String selectorExpression = consumerConfig.getSelectorExpression();
    String ak = consumerConfig.getAccessKey();
    String sk = consumerConfig.getSecretKey();
    int pullBatchSize = consumerConfig.getPullBatchSize();
    boolean useTLS = consumerConfig.isTlsEnable();

    DefaultLitePullConsumer litePullConsumer = RocketMQUtil.createDefaultLitePullConsumer(nameServer, accessChannel,
            groupName, topicName, messageModel, selectorType, selectorExpression, ak, sk, pullBatchSize, useTLS);
    litePullConsumer.setEnableMsgTrace(consumerConfig.isEnableMsgTrace());
    litePullConsumer.setCustomizedTraceTopic(consumerConfig.getCustomizedTraceTopic());
    litePullConsumer.setNamespace(consumerConfig.getNamespace());
    litePullConsumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragely());
    litePullConsumer.setAutoCommit(true);
    return litePullConsumer;
}

```

invoke poll method

```java
try {
            if (CollUtil.isEmpty(PULL_CONSUMERS)) {
                this.setRunning(true);
                return;
            }
            for (DefaultLitePullConsumer pullConsumer : PULL_CONSUMERS) {
                pullConsumer.start();
            }
        } catch (MQClientException e) {
            throw new IllegalStateException("start pull fail", e);
        }

```


GitHub link: https://github.com/apache/rocketmq/discussions/6578#discussioncomment-5587162

----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org

[GitHub] [rocketmq] jiakme edited a comment on the discussion: single instance with multi DefaultLitePullConsumer(use same topic and different tag) don't poll message right

Posted by GitBox <gi...@apache.org>.
GitHub user jiakme edited a comment on the discussion: single instance with multi DefaultLitePullConsumer(use same topic and different tag) don't poll message right

```java
private void initRocketMQPushConsumer() throws MQClientException {

    List<DefaultLitePullConsumer> pullConsumers = new ArrayList<>();

    DefaultLitePullConsumer defaultLitePullConsumer = getDefaultLitePullConsumer("high");
    pullConsumers.add(defaultLitePullConsumer);

    DefaultLitePullConsumer defaultLitePullConsumer_1 = getDefaultLitePullConsumer("low");
    pullConsumers.add(defaultLitePullConsumer_1);

    PULL_CONSUMERS = Collections.unmodifiableList(pullConsumers);
}

private DefaultLitePullConsumer getDefaultLitePullConsumer(String tag) throws MQClientException {
    RocketMQProperties.Consumer consumerConfig = new RocketMQProperties.Consumer();
    consumerConfig.setGroup(MqConfig.CATEGORY_SMART_CAMPUS_APP_CENTER + "_priority");

    consumerConfig.setTopic(TOPIC);
    consumerConfig.setPullBatchSize(50);
    DefaultLitePullConsumer defaultLitePullConsumer;

    consumerConfig.setSelectorExpression(tag);
    defaultLitePullConsumer = buildPullConsumer(consumerConfig);
    return defaultLitePullConsumer;
}
private DefaultLitePullConsumer buildPullConsumer(RocketMQProperties.Consumer consumerConfig) throws MQClientException {
    /**
     * {@link org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration#defaultLitePullConsumer(RocketMQProperties)}
     */
    String nameServer = rocketMQProperties.getNameServer();
    String groupName = consumerConfig.getGroup();
    String topicName = consumerConfig.getTopic();
    Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
    Assert.hasText(groupName, "[rocketmq.pull-consumer.group] must not be null");
    Assert.hasText(topicName, "[rocketmq.pull-consumer.topic] must not be null");

    String accessChannel = rocketMQProperties.getAccessChannel();
    MessageModel messageModel = MessageModel.valueOf(consumerConfig.getMessageModel());
    SelectorType selectorType = SelectorType.valueOf(consumerConfig.getSelectorType());
    String selectorExpression = consumerConfig.getSelectorExpression();
    String ak = consumerConfig.getAccessKey();
    String sk = consumerConfig.getSecretKey();
    int pullBatchSize = consumerConfig.getPullBatchSize();
    boolean useTLS = consumerConfig.isTlsEnable();

    DefaultLitePullConsumer litePullConsumer = RocketMQUtil.createDefaultLitePullConsumer(nameServer, accessChannel,
            groupName, topicName, messageModel, selectorType, selectorExpression, ak, sk, pullBatchSize, useTLS);
    litePullConsumer.setEnableMsgTrace(consumerConfig.isEnableMsgTrace());
    litePullConsumer.setCustomizedTraceTopic(consumerConfig.getCustomizedTraceTopic());
    litePullConsumer.setNamespace(consumerConfig.getNamespace());
    litePullConsumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragely());
    litePullConsumer.setAutoCommit(true);
    return litePullConsumer;
}

```

GitHub link: https://github.com/apache/rocketmq/discussions/6578#discussioncomment-5587162

----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org