class RocketmqConsumerHandler implements MessageListenerConcurrently {
    static final Logger log = LoggerFactory.getLogger(RocketmqConsumerHandler.class);

    private final CloudEventObserverManger observerManger;
    private final RocketmqConfig  config;

    public RocketmqConsumerHandler(RocketmqConfig config, CloudEventObserverManger observerManger) {
        this.observerManger = observerManger;
        this.config =config;
    }


    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        boolean isOk = true;

        try {
            for (MessageExt message : list) {
                String topicNew = message.getTopic();
                String group = null;
                String topic = null;
                if(topicNew.contains(RocketmqProps.GROUP_SPLIT_MARK)){
                    group = topicNew.split(RocketmqProps.GROUP_SPLIT_MARK)[0];
                    topic = topicNew.split(RocketmqProps.GROUP_SPLIT_MARK)[1];
                }else{
                    topic = topicNew;
                }

                Event event = new Event(topic, new String(message.getBody()));
                event.tags(message.getTags());
                event.key(message.getKeys());
                event.times(message.getReconsumeTimes());
                event.channel(config.getChannelName());
                if(Utils.isNotEmpty(group)){
                    event.group(group);
                }

                isOk = isOk && onReceive(event, topicNew); //可以不吃异常
            }
        } catch (Throwable e) {
            isOk = false;
            log.warn(e.getMessage(), e);
        }

        if (isOk) {
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } else {
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }

    /**
     * 处理接收事件
     */
    protected boolean onReceive(Event event, String topicNew) throws Throwable {<FILL_FUNCTION_BODY>}
}

        boolean isOk = true;
        CloudEventHandler handler = null;

        if (Utils.isEmpty(event.tags())) {
            handler = observerManger.getByTopicAndTag(topicNew, "*");
        } else {
            handler = observerManger.getByTopicAndTag(topicNew, event.tags());

            if (handler == null) {
                handler = observerManger.getByTopicAndTag(topicNew, "*");
            }
        }

        if (handler != null) {
            isOk = handler.handle(event);
        }else{
            //只需要记录一下
            log.warn("There is no observer for this event topic[{}]", topicNew);
        }

        return isOk;
    
<no_super_class>