class RocketmqConsumer {
    static Logger log = LoggerFactory.getLogger(RocketmqConsumer.class);

    private RocketmqConfig config;

    private DefaultMQPushConsumer consumer;

    public RocketmqConsumer(RocketmqConfig config) {
        this.config = config;
    }

    public void init(CloudProps cloudProps, CloudEventObserverManger observerManger) throws MQClientException {<FILL_FUNCTION_BODY>}
}

        if (consumer != null) {
            return;
        }

        Utils.locker().lock();

        try {
            if (consumer != null) {
                return;
            }

            RocketmqConsumerHandler handler = new RocketmqConsumerHandler(config, observerManger);

            if (Utils.isEmpty(config.getAccessKey())) {
                consumer = new DefaultMQPushConsumer();
            } else {
                RPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(config.getAccessKey(), config.getSecretKey()));
                consumer = new DefaultMQPushConsumer(rpcHook);
            }


            //服务地址
            consumer.setNamesrvAddr(config.getServer());
            //消费组
            consumer.setConsumerGroup(config.getConsumerGroup());
            //命名空间
            if (Utils.isNotEmpty(config.getNamespace())) {
                consumer.setNamespace(config.getNamespace());
            }

            //消息线程数
            if (config.getConsumeThreadNums() > 0) {
                consumer.setConsumeThreadMax(config.getConsumeThreadNums());
            }

            //最大尝试次数（之后进死信）
            if (config.getMaxReconsumeTimes() > 0) {
                consumer.setMaxReconsumeTimes(config.getMaxReconsumeTimes());
            }

            //一次最大消费的条数
            consumer.setConsumeMessageBatchMaxSize(1); //1是默认值
            //一次最大拉取的条数
            consumer.setPullBatchSize(32); //32是默认值
            //无消息时，最大阻塞时间。默认5000 单位ms

            //绑定定制属性
            Properties props = cloudProps.getEventConsumerProps();
            if (props.size() > 0) {
                Utils.injectProperties(consumer, props);
            }


            //要消费的topic，可使用tag进行简单过滤
            for (Map.Entry<String, Set<String>> kv : observerManger.topicTags().entrySet()) {
                String topic = kv.getKey();
                Set<String> tags = kv.getValue();
                String tagsExpr = String.join("||", tags);

                //支持 tag 过滤
                if (tags.contains("*")) {
                    consumer.subscribe(topic, "*");
                } else {
                    consumer.subscribe(topic, tagsExpr);
                }

                log.trace("Rocketmq consumer subscribe [" + topic + "(" + tagsExpr + ")] ok!");
            }

            consumer.registerMessageListener(handler);
            consumer.start();

            log.trace("Rocketmq consumer started!");
        } finally {
            Utils.locker().unlock();
        }
    
<no_super_class>