Stream.distinctをフィールド・プロパティなどで行うには #Java

Stream.distinctをフィールド・プロパティ・計算結果などで行うには

java Stream APIのStream.distinctは引数にラムダ式をとれません。

class Item{
    String name,shop;
    int price;
    Item(String n,int p,String s){ name=n; price=p; shop=s; }
    public String toString(){ return name+", "+price+", "+shop; }
}
Item[] items = {
    new Item("item-1",1000,"shop-A"),
    new Item("item-2",1100,"shop-B"),
    new Item("item-3",1200,"shop-C"),
    new Item("item-4",2000,"shop-A"),
    new Item("item-5",2100,"shop-B"),
    new Item("item-6",2200,"shop-C"),
    new Item("item-7",3000,"shop-A"),
    new Item("item-8",3100,"shop-B"),
    new Item("item-9",3200,"shop-C"),
};

上の配列からshopごとに1つ商品を取り出すには、distinctが引数にラムダ式をとれるなら、次のようになるでしょう。

Stream.of(items)
    .distinct(item->item.shop)
    ...

でもこんな書き方はできないので、filterとSetを組み合わせて代用します。

Set<String> unique = new HashSet<>();
Stream.of(items)
    .filter(item->unique.add(item.shop))
    .forEach(System.out::println);

> item-1, 1000, shop-A
> item-2, 1100, shop-B
> item-3, 1200, shop-C

なぜこれでdistinctと同等なのかというと、Set.addは次のような動作をするからです。

指定された要素が集合に含まれない場合、集合に追加し、trueを返す。
指定された要素が集合にすでに含まれる場合、集合を変更せずにfalseを返す。

つまり、重複する要素の最初の1つ対してtrueを返す条件式として使えるということです。

並列ストリームでは

しかし、HashSetはスレッドセーフではないので並列ストリームでは危険です。

Set<String> unique = new HashSet<>(); //スレッドセーフではない！
Stream.of(items)
    .parallel() //マルチスレッドで処理される！
    .filter(item->unique.add(item.shop))
    .forEach(System.out::println);

> item-7,3000,shop-A //重複！
> item-5,2100,shop-B
> item-1,1000,shop-A //重複！
> item-6,2200,shop-C

1/100くらいの割合で、上のような結果になりました。
代わりに並行性をサポートするConcurrentHashMapを、Collections.newSetFromMapでSetに変換して使います。

Set<String> unique = Collections.newSetFromMap(new ConcurrentHashMap<>());
Stream.of(items)
    .parallel()
    .filter(item->unique.add(item.shop))
    .forEach(System.out::println);

> item-6, 2200, shop-C
> item-5, 2100, shop-B
> item-7, 3000, shop-A