Apache RocketMQ中如果只是消息推送，消费，代码这一层是可以识别的，但是到了rocke？

Apache RocketMQ中如果只是消息推送，消费，代码这一层是可以识别的，但是到了rocketmq这块，就是状态改不过来，还有消息推送之后，消息重复的问题， 我rocketmq用controller集群部署，proxy换成CLUSTER模式后，消费的消息显示我消费者不在线是啥原因你知道嘛？import java.io.IOException; import java.util.Collections; import org.apache.rocketmq.client.apis.ClientConfiguration; import org.apache.rocketmq.client.apis.ClientException; import org.apache.rocketmq.client.apis.ClientServiceProvider; import org.apache.rocketmq.client.apis.consumer.ConsumeResult; import org.apache.rocketmq.client.apis.consumer.FilterExpression; import org.apache.rocketmq.client.apis.consumer.FilterExpressionType; import org.apache.rocketmq.client.apis.consumer.PushConsumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory;

public class PushConsumerExample02 { private static final Logger logger = LoggerFactory.getLogger(PushConsumerExample.class);

private PushConsumerExample02() {
}

public static void main(String[] args) throws ClientException, IOException, InterruptedException {
    final ClientServiceProvider provider = ClientServiceProvider.loadService();
    // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
    String endpoints = "192.168.21.170:28081;192.168.21.170:29081";
    ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoints)
            .build();
    // 订阅消息的过滤规则，表示订阅所有Tag的消息。
    String tag = "messageTag";
    FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
    // 为消费者指定所属的消费者分组，Group需要提前创建。
    String consumerGroup = "test";
    // 指定需要订阅哪个目标Topic，Topic需要提前创建。
    String topic = "TestTopic";
    // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
    PushConsumer pushConsumer = provider.newPushConsumerBuilder()
            .setClientConfiguration(clientConfiguration)
            // 设置消费者分组。
            .setConsumerGroup(consumerGroup)
            // 设置预绑定的订阅关系。
            .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
            // 设置消费监听器。
            .setMessageListener(messageView -> {
                // 处理消息并返回消费结果。
                logger.info("Consume message successfully, messageId={}", messageView.getMessageId());
                return ConsumeResult.SUCCESS;
            })
            .build();
    Thread.sleep(Long.MAX_VALUE);
    // 如果不需要再使用 PushConsumer，可关闭该实例。
    // pushConsumer.close();
}

} proxy服务换成LOCAL模式，消息状态就正常，proxy的CLUSTER模式，组这边识别不到，LOCAL模式这边能有个客户端信息