代码之家 › 专栏 › 技术社区 › Sebastiaan van den Broek

如何提取函数,在Java 8中,我可以将这些函数作为链式lambda的参数传递?

method-chaining apache-kafka-streams lambda java

-1

Sebastiaan van den Broek · 技术社区 · 6 年前

我在一段代码中有一个代码模式,使用卡夫卡流,它不断重复,我做一个映射,然后按键分组,然后减少。看起来是这样的:

KTable<ProjectKey, EventConfigurationIdsWithDeletedState> eventConfigurationsByProjectTable = eventConfigurationStream
        .map((key, value) -> {
            Map<String, Boolean> eventConfigurationUpdates = new HashMap<>();
            eventConfigurationUpdates.put(key.getEventConfigurationId(), value != null);
            ProjectKey projectKey = ProjectKey.newBuilder().setId(key.getProjectId()).build();
            EventConfigurationIdsWithDeletedState eventConfigurationIdsWithDeletedState = EventConfigurationIdsWithDeletedState.newBuilder().setEventConfigurations(eventConfigurationUpdates).build();
            return KeyValue.pair(projectKey, eventConfigurationIdsWithDeletedState);
        })
        .groupByKey()
        .reduce((aggValue, newValue) -> {
            Map<String, Boolean> newEventConfigurations = newValue.getEventConfigurations();
            Map<String, Boolean> aggEventConfigurations = aggValue.getEventConfigurations();
            Map.Entry<String, Boolean> newEntry = newEventConfigurations.entrySet().iterator().next();
            if (newEntry.getValue())
                aggEventConfigurations.putAll(newEventConfigurations);
            else
                aggEventConfigurations.remove(newEntry.getKey());
            if (aggEventConfigurations.size() == 0)
                return null;
            return aggValue;
        });

(事件配置流的类型为 KStream<EventConfigurationKey, EventConfiguration> )

遵循此模式的另一个示例。注意这里也有一个过滤器,但情况并非总是这样:

KTable<ProjectKey, NotificationSettingsTransition> globalNotificationSettingsPerProjectTable = notificationSettingTable.toStream()
        .filter((key, value) -> {
            return key.getEventConfigurationId() == null;
        })
        .map((key, value) -> {
            ProjectKey projectKey = ProjectKey.newBuilder().setId(key.getProjectId()).build();
            Map<String, NotificationSetting> notificationSettingsMap = new HashMap<>();
            notificationSettingsMap.put(getAsCompoundKeyString(key), value);
            NotificationSettingsTransition notificationSettingTransition = NotificationSettingsTransition
                    .newBuilder()
                    .setNotificationSettingCompoundKeyLastUpdate(getAsCompoundKey(key))
                    .setNotificationSettingLastUpdate(value)
                    .setEventConfigurationIds(new ArrayList<>())
                    .setNotificationSettingsMap(notificationSettingsMap)
                    .build();

            return KeyValue.pair(projectKey, notificationSettingTransition);
        })
        .groupByKey()
        .reduce((aggValue, newValue) -> {
            Map<String, NotificationSetting> notificationSettingMap = aggValue.getNotificationSettingsMap();
            String compoundKeyAsString = getAsString(newValue.getNotificationSettingCompoundKeyLastUpdate());
            if (newValue.getNotificationSettingLastUpdate() != null)
                notificationSettingMap.put(compoundKeyAsString, newValue.getNotificationSettingLastUpdate());
            else
                notificationSettingMap.remove(compoundKeyAsString);
            aggValue.setNotificationSettingCompoundKeyLastUpdate(newValue.getNotificationSettingCompoundKeyLastUpdate());
            aggValue.setNotificationSettingLastUpdate(newValue.getNotificationSettingLastUpdate());
            aggValue.setNotificationSettingsMap(notificationSettingMap);
            return aggValue;
        });

(通知设置表类型为 KTable<NotificationSettingKey, NotificationSetting> notificationSettingTable 但马上也会转变成一种主流。)

如何将其提取到一个函数中,在该函数中我传递一个映射代码和reduce代码的函数,但不必重复 .map().groupByKey().reduce() 是吗?同时考虑到返回类型是不同的,并且依赖于map函数中的代码,应该保持类型化。理想的是在Java 8中,但更高的版本是可能的。我想我很清楚当 KeyValuePair 地图中的代码不会改变,但现在不确定如何做。

1 回复 | 直到 6 年前

Billy Brown 6 年前

您可以参数化您的函数以接受两个泛型函数,在调用函数时,类型将被推断出来(如果不可能,可以显式设置)。

输入到 map 你想要一个 BiFunction<K, V, T> 为了 reduce 你想要一个 BiFunction<U, U, U> ,其中:

K 是什么类型 key 在里面 地图 的功能。
V 是什么类型 value 在里面 地图 的功能。
T 返回类型是 地图 的功能。
U 是聚合器的类型、值和返回类型 减少 的功能。

看着 KStream 和 KGroupedStream ,您可以获得更详细的类型信息来进一步约束函数。

这将使您的自定义功能如下所示:

<K, V, T, U> U mapGroupReduce(final KStream<K, V> stream, final BiFunction<K, V, T> mapper, final BiFunction<U, U, U> reducer) {
    return stream.map(mapper).groupByKey().reduce(reducer);
}

然后你可以这样称呼它:

mapGroupReduce(yourStream,
    (key, value) -> new KeyValue(k, v)),
    (acc, value) -> acc);

在您的情况下,而不是使用 BiFunction S,您需要使用:

KeyValueMapper<K, V, KeyValue<T, U>> 对于映射器
Reducer<U> 对于减速器。

然而,这一切真的比仅仅写作要好得多吗? stream.map(M).groupByKey().reduce(R) 每一次?更详细的版本更为明确,考虑到mapper和reducer的相对大小,您并没有真正节省那么多。