这个数据是存储在kafka中的，统计的功能尽量用kafka自带的功能实现

明白了，你希望使用Kafka自带的功能来实现消费数据的统计。在这种情况下，你可以使用Kafka Streams来进行实时数据处理和统计。

下面是一个简单的案例，使用Kafka Streams和Java来从Kafka中消费数据并实时统计消费总额：

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.state.KeyValueStore;

import java.util.Arrays;
import java.util.Properties;

public class KafkaExpenseTracker {

    public static void main(String[] args) {
        // Kafka配置
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "expense-tracker");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        // 构建流处理拓扑
        StreamsBuilder builder = new StreamsBuilder();

        // 从Kafka主题中消费数据
        KStream<String, String> source = builder.stream("expense-topic");

        // 对消费数据进行处理和转换
        KTable<String, Double> expenses = source.flatMapValues(value -> Arrays.asList(value.split(",")))
                .groupBy((key, value) -> value)
                .aggregate(() -> 0.0, (key, newValue, aggValue) -> aggValue + Double.parseDouble(newValue),
                        Materialized.<String, Double, KeyValueStore<org.apache.kafka.common.utils.Bytes, byte[]>>as("expense-aggregates")
                                .withKeySerde(Serdes.String())
                                .withValueSerde(Serdes.Double()))
                .mapValues((readOnlyKey, value) -> String.format("%.2f", value));

        // 将统计结果发送到另一个Kafka主题
        expenses.toStream().to("expense-total-topic", Produced.with(Serdes.String(), Serdes.String()));

        // 创建Kafka Streams应用程序
        KafkaStreams streams = new KafkaStreams(builder.build(), props);

        // 启动应用程序
        streams.start();
    }
}

在这个案例中，我们使用`StreamsBuilder`构建了一个流处理拓扑。首先，我们从名为`expense-topic`的Kafka主题中消费数据流，并将每条消息的值按逗号分隔为多个消费金额。然后，我们根据消费金额进行分组，并使用`aggregate`操作对每个金额进行累加计算，最终得到每个消费金额对应的总额。最后，我们将统计结果发送到名为`expense-total-topic`的Kafka主题。

请注意，上述代码中的Kafka配置信息需要根据实际情况进行修改，包括`bootstrap.servers`和主题名称等。

这个案例演示了如何使用Kafka Streams进行实时的消费数据统计。你可以根据实际需求进行相应的扩展和修改，例如添加更多的处理逻辑或使用不同的聚合函数来计算消费总额。

另外，为了保证代码的正确性和性能，你需要考虑一些方面，例如Kafka主题的分区和副本配置、数据序列化和反序列化方式的选择以及流处理应用程序的容错和恢复机制等。