Java.util.function 接口集成

Question

将基线移至 Java 8 允许我们考虑与 Function、Predicate、Supplier 等关键 java.util.Function 接口固有的连接。我们需要评估用户能做什么，以及它将开放何种自动集成。

https://docs.oracle.com/javase/8/docs/api/java/util/function/package-summary.html

Comments

为什么到 2023 年我们还没有任何官方的回应关于何时实现或处理这件事？

---

我们已经对此问题进行了多次评估。有很多方面和几个方案供我们选择，其中一些我们已经创建了原型，但我们还没有决定要做什么。它在我们的 1.12 范围列表中。

Answer 1

这是用Java编写的Kafka Streams应用的示例

      sb.table("input", Consumed.with(sl, sl))
            .groupBy((k, v) -> KeyValue.pair(k / 10, v), Grouped.with(sl, sl))
            .aggregate(() -> 0L,
                    (k, v, acc) -> acc + v,
                    (k, v, acc) -> acc - v,
                    Materialized.with(sl, sl))
            .toStream()
            .to("output", Produced.with(sl, sl));

同样的应用用Clojure编写如下

    (-> sb
        (.table "input" (topic->consumed data-in))
        (.groupBy (key-value-mapper
                    (fn [k v] (KeyValue/pair (long (/ k 10)) v)))
                (serdes->grouped "groupie" data-in))
        (.aggregate (reify Initializer
                    (apply [_] 0))
                    (reify Aggregator
                    (apply [_ k v acc]
                        (+ acc v)))
                    (reify Aggregator
                    (apply [_ k v acc]
                        (- acc v)))
                    (serdes->materialised ...))
        (.toStream)
        (.to "output" (topic->produced data-out)))

如果我们能够在期望SAM类型的Lambda表达式处使用Lambda表达式，则可以写作这种形式

    (-> sb
        (.table "input" (topic->consumed data-in))
        (.groupBy (fn [k v] (KeyValue/pair (long (/ k 10)) v))
                (serdes->grouped "groupie" data-in))
        (.aggregate (constantly 0)
                    (fn [k v acc] (+ acc v))
                    (fn [k v acc] (- acc v))
                    (serdes->materialised ...))
        (.toStream)
        (.to "output" (topic->produced data-out)))

Comments

恭喜您在Clojure 1.12中实现这一功能，感谢！
https://clojure.atlassian.net/browse/CLJ-2799

Answer 2

评论者：jwhitlark

如果可以在需要java.util.function.*的任何地方使用IFn，那将非常棒！

Comments

该补丁似乎实现了这一点： https://clojure.atlassian.net/plugins/servlet/mobile?originPath=%2Fbrowse%2FCLJ-2637#issue/CLJ-2637

Answer 3

评论者：marctrem

将基线版本升级到Java 8可允许我们使用默认接口方法。

some-java-fns-interface.patch 补丁在IFn上实现了Consumer、Function、Predicate和Supplier。

如果您想走这条路，我将非常乐意在IFn上也实现java.util.function下的所有接口，以及相关的测试。我目前使用这段代码通过Java客户端使用FoundationDB，并且它对我来说效果很好。

https://github.com/marctrem/clojure/commit/97742493f674edd8f6c034ee94da84fa62a76bad

Answer 4

https://clojure.atlassian.net/plugins/servlet/mobile?originPath=/browse/CLJ-2637#issue/CLJ-2637

Comments

哎呀，我们最好为CLJ-2637分配一个“问题”，以免这里的评论在这两种方法之间纠缠不清。

Answer 5

刚刚遇到了这个问题，我想提供一些使用/上下文信息。

当与使用CompletableFuture和CompletionStage编写的异步Java代码交互时，需要提供实现Function、Consumer、BiFunction等的参数。

我正在使用这样的宏

(defmacro as-function [f]
  `(reify java.util.function.Function
     (apply [this arg#]
       (~f arg#))))

(defmacro as-consumer [f]
  `(reify java.util.function.Consumer
     (accept [this arg#]
       (~f arg#))))

但是这很快就变得繁琐，因为需要根据函数参数的不同进行变体。

java.util.function定义了大量的接口，但根据我的（有限的）观点，我最需要的是最常见的接口，特别是那些由CompletionStage和CompletableFuture需要的。

Comments

Answer 6

数据请求

抛开java.util.stream.Stream以及 existing JDK API 的 retrofit 如 new Thread(() -> doSomething(x))，以下哪些是 lambda/SAM 使用库，在 Clojure 中使用起来比较痛苦？

Answer 7

_由：jwhitlark_发表的评论


;; 我从一些关于 kafka streams 的实验代码中找到了这个。原始代码中所有的 reify 填充了 java8 的 lambda。

;; 我会发现使用 java.utils.funstion.* 的一些示例。

;; 以下内容部分来源于一个 franzy 示例或者别的什么？

;; 注意，例如： https://kafka.apache.org/0102/javadoc/org/apache/kafka/streams/kstream/Predicate.html
;; 与 https://docs.oracle.com/javase/8/docs/api/java/util/function/Predicate.html
;; 是不同的。
(ns utils

  (:import (org.apache.kafka.streams.kstream Reducer KeyValueMapper ValueMapper Predicate))
(defmacro reducer [kv & body

    `(reify Reducer
     (apply [_# ~(first kv) ~(second kv)]
       ~@body)))
  

(ns utils
(ocurrency) (:import (org.apache.kafka.streams.kstream Reducer KeyValueMapper ValueMapper Predicate))
;; public interface KeyValueMapper)
;; apply(K key, V value)
       ~@body)))
  

(defmacro kv-mapper [kv & body)
;; public interface ValueMapper)
;; apply(V1 value)
(defmacro v-mapper [v & body)
  `(reify ValueMapper
  

     (apply [_# ~v
(defmacro pred [kv & body)
  `(reify Predicate
  

     (test [_# ~(first kv) ~(second kv)]

;; 我像这样使用了它
(ns our-service.kafka-streams
  (:require
   [our-service.util :as k]
   [clojure.string :as str]
  (:import
           (org.apache.kafka.streams StreamsConfig KafkaStreams KeyValue)

           (org.apache.kafka.streams.kstream KStreamBuilder ValueMapper)))
(defn create-word-count-topology []
  (let [builder (KStreamBuilder.)
        init-stream (.stream builder (into-array ["streams-str-input"]))
        wc (-> init-stream
            (.flatMapValues (k/v-mapper [& value]
                              `(str/split (apply str value) #"\s")))
            (.map (k/kv-mapper [k v]
                     `(KeyValue/pair v v)))
            (.filter (k/pred [k v]
                                          （not= v "the"）)
            （.groupByKey）
            （.count "CountStore"）
            显示项目
            ;; 这需要是mapValues
            （.mapValues（reify ValueMapper
                          （apply [_ v]
                            ；（println v）
                            ；（str v））））
            （.toStream）
            （.to "wordcount-output")】
【构建器 wc】

Answer 8

评论者：gshayban

根据 JLS，通过搜索匹配的功能接口，即“单个抽象方法”类（链接：1）来推断 lambda 类型（无论是接口还是抽象类）。我们可以有一个类似的辅助方法来检测这些类（链接：2）。您需要指定目标类。我们实际上不需要既是 IFn 又是 j.u.f.Predicate 等的东西。

`
（导入 '[java.util.function Predicate Consumer]）

（let [orig [1 2 3]

  st (atom [])]

（.forEach orig (jfn Consumer [x] (swap! st conj x)))
（= @st orig）
`

（链接：1） https://docs.oracle.com/javase/specs/jls/se8/html/jls-9.html#jls-9.8
（链接：2） spike https://gist.github.com/ghadishayban/0ac41e81d4df02ff176c22d16ee8b972

Answer 9

评论者：jwhitlark

好吧，这将是一种改进。我遇到的实际问题是，我经常深入到流畅的接口中，并不一定知道确切的类。话虽如此，通常只有几个地方。有一个登记处有意义吗？或许像这样的东西

（auto-infer-lambda（链接：java.util.function，org.apache.kafka.streams.kstream））

Answer 10

评论者：gshayban

您是否使用过抽象类而不是接口的 SAM 类？

Answer 11

评论者：ajoberstar

以下是我（链接：[这里是链接](https://github.com/ajoberstar/ike.cljj/blob/master/src/main/clojure/ike/cljj/function.clj) 文档：ike.cljj）库中的一个替代方法。这个方法使用 MethodHandles（即 java.lang.invoke 包）而不是常规反射。我不确定是否已经在抽象类上测试过这个方法。

用法看起来与 Ghadi 发布的类似

`
(defsam my-sam
java.util.function.Predicate
[x]
(= x "it matched"))

(-> (Stream/of "not a match" "it matched")

(.filter my-sam)
(.collect Collectors/toList)

(-> (IntStream/range 0 10)

(.filter (sam* java.util.function.IntPredicate odd?))
(.collect Collectors/toList)

`

它使用（链接：[这里是链接](https://docs.oracle.com/javase/8/docs/api/java/lang/invoke/MethodHandleProxies.html#asInterfaceInstance-java.lang.Class-java.lang.invoke.MethodHandle-)）的 MethodHandleProxies.asInterfaceInstance 方法来创建一个代理实例，该实例调用一个调用 Clojure 函数的方法句柄。它不会尝试验证参数数量，它只是将 varargs 代理到 ISeq 的 applyTo。不确定这是否是最有效的方法，但对我来说很有效。

我认为（链接：[这里是链接](https://docs.oracle.com/javase/8/docs/api/index.html?java/lang/invoke/MethodHandles.html)）可能是满足此类用例的首选方法。不过，对我来说，要确切了解如何使用它有点困难，所以我并没有深入研究。

我对我的方法（以及 Ghadi 的方法）的主要功能性问题是，你必须明确提供你想要代理的接口。Java lambdas 和 Groovy Closures 可以用于期望 SAM 的方法，它只是根据方法期望的内容进行强制转换。理想情况下，Clojure 也应该支持这一点。

而不是不得不这样做

`
(-> (IntStream/range 0 10)

(.filter (sam* java.util.function.IntPredicate odd?))
(.collect Collectors/toList)

`

我想这样做

`
(-> (IntStream/range 0 10)

(.filter odd?)
(.collect Collectors/toList)

`

Answer 12

评论者：gshayban

另一个可能的方法是将 java.util.function.Supplier 扩展到 Clojure 函数中，这是一个明确的 0 参数。该接口在现实世界中变得越来越常见；这可能是一个值得特殊处理的案例。（我们不应该也不可能对 defrecords 做同样的事情，因为他们已经有一个名叫 get 的方法，这与 Supplier 的唯一方法相冲突。）

Answer 13

参考：[这里是链接](https://clojure.atlassian.net/browse/CLJ-2365)（由 alexmiller 报告）