谈谈代码：函数式编程

    private void getTopicsDiskSizeForSomeBroker(int brokerID, AdminClient admin, Map<String, Long> topicsSizeMap) throws ExecutionException, InterruptedException {
        DescribeLogDirsResult ret = admin.describeLogDirs(Collections.singletonList(brokerID));
        Map<Integer, Map<String, DescribeLogDirsResponse.LogDirInfo>> tmp = ret.all().get();
        for (Map.Entry<Integer, Map<String, DescribeLogDirsResponse.LogDirInfo>> entry : tmp.entrySet()) {
            Map<String, DescribeLogDirsResponse.LogDirInfo> tmp1 = entry.getValue();
            for (Map.Entry<String, DescribeLogDirsResponse.LogDirInfo> entry1 : tmp1.entrySet()) {
                DescribeLogDirsResponse.LogDirInfo info = entry1.getValue();
                Map<TopicPartition, DescribeLogDirsResponse.ReplicaInfo> replicaInfoMap = info.replicaInfos;
                for (Map.Entry<TopicPartition, DescribeLogDirsResponse.ReplicaInfo> replicas : replicaInfoMap.entrySet()) {
                    String topic = replicas.getKey().topic();
                    Long topicSize = topicsSizeMap.get(topic);
                    if (topicSize == null) {
                        topicsSizeMap.put(topic, replicas.getValue().size);
                    } else {
                        topicsSizeMap.put(topic, replicas.getValue().size + topicSize);
                    }
                }
            }
        }
    }

    public List<KafkaTopicInfoBO> getTopicDiskSize() {
        return getTopicPartitionReplicaInfo().entrySet().stream()
                .map(e -> new KafkaTopicInfoBO(e.getKey().topic(), e.getValue().size))
                .collect(Collectors.toList());

    }

    protected Map<TopicPartition, DescribeLogDirsResponse.ReplicaInfo> getTopicPartitionReplicaInfo() {
        Properties globalConf = zkConfigService.getProperties(ZkPathUtils.GLOBAL_CONFIG);
        Properties adminConfig = new Properties();
        adminConfig.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, globalConf.getProperty((ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)));
        AdminClient adminClient = AdminClient.create(adminConfig);
        List<String> brokerIds = zkConfigService.getChildByPath(kafkaIdsPath);
        return  brokerIds.stream()
                .map(Integer::valueOf)
                .map(Collections::singletonList)
                .map(adminClient::describeLogDirs)
                .map(DescribeLogDirsResult::all)
                .map(mapKafkaFuture -> {
                    try {
                        return mapKafkaFuture.get();
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                })
                .map(Map::values)
                .flatMap(Collection::stream)
                .map(Map::values)
                .flatMap(Collection::stream)
                .map(e -> e.replicaInfos)
                .map(Map::entrySet)
                .flatMap(Collection::stream)
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
    }

f(x)=5x^2+4x+3
g(x)=2f(x)+5=10x^2+8x+11
h(x)=f(x)+g(x)=15x^2+12x+14

public class Quint{
    public static void main (String args[]){
        for (int i=0; i<25; i++){
            System.out.println(i*i);
        }
    }
}

(println (take 25 (map (fn [x] (*x x) (range)))))

    protected fun getTopicPartitionReplicaInfo(): Map<TopicPartition, DescribeLogDirsResponse.ReplicaInfo> {
        val globalConf = zkConfigService.getProperties(ZkPathUtils.GLOBAL_CONFIG)
        val adminConfig = Properties()
        adminConfig.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, globalConf.getProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))
        val adminClient = AdminClient.create(adminConfig)
        val brokerIds = zkConfigService.getChildByPath(kafkaIdsPath)
        return brokerIds.stream()
                .mapToInt {
                    Integer.valueOf(it)
                }.let { intStream ->
                    adminClient.describeLogDirs(intStream.boxed().collect(Collectors.toList()))
                }.let { describeLogDirsResult ->
                    describeLogDirsResult.all()
                }.let { mapKafkaFutrue ->
                    mapKafkaFutrue.get()
                }.let { mapStream ->
                    mapStream.values
                }.let {
                    it.stream().map { e -> e.values }.flatMap { e -> e.stream() }.collect(Collectors.toList())
                }.flatMap {
                    it.replicaInfos.entries.toList()
                }.let { it ->
                    it.associateBy({ it.key }, { it.value })
                }
    }

     * Returns a stream consisting of the results of applying the given
     * function to the elements of this stream.
     *
     * <p>This is an <a href="package-summary.html#StreamOps">intermediate
     * operation</a>.
     *
     * @param <R> The element type of the new stream
     * @param mapper a <a href="package-summary.html#NonInterference">non-interfering</a>,
     *               <a href="package-summary.html#Statelessness">stateless</a>
     *               function to apply to each element
     * @return the new stream
     */
    <R> Stream<R> map(Function<? super T, ? extends R> mapper);

/**
 * Calls the specified function [block] with `this` value as its argument and returns its result.
 *
 * For detailed usage information see the documentation for [scope functions](https://kotlinlang.org/docs/reference/scope-functions.html#let).
 */
@kotlin.internal.InlineOnly
public inline fun <T, R> T.let(block: (T) -> R): R {
    contract {
        callsInPlace(block, InvocationKind.EXACTLY_ONCE)
    }
    return block(this)
}