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

##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

Ok