标签: java-stream

在我的机器上,下面的程序打印:

OptionalLong[134043]
 PARALLEL took 127869 ms
OptionalLong[134043]
 SERIAL took 60594 ms

我不清楚为什么串行执行程序比并行执行程序要快.我把两个程序-Xms2g -Xmx2g放在一个8gb相对安静的盒子上.有人可以澄清最新情况吗？

import java.util.stream.LongStream;
import java.util.stream.LongStream.Builder;

public class Problem47 {

    public static void main(String[] args) {

        final long startTime = System.currentTimeMillis();
        System.out.println(LongStream.iterate(1, n -> n + 1).parallel().limit(1000000).filter(n -> fourConsecutives(n)).findFirst());
        final long endTime = System.currentTimeMillis();
        System.out.println(" PARALLEL took " +(endTime - startTime) + " ms");

        final long startTime2 = System.currentTimeMillis();
        System.out.println(LongStream.iterate(1, n -> n + 1).limit(1000000).filter(n -> fourConsecutives(n)).findFirst());
        final long endTime2 = System.currentTimeMillis();
        System.out.println(" SERIAL took …

我正在尝试收集丢弃很少使用的项目的流,如下例所示:

import java.util.*;
import java.util.function.Function;
import static java.util.stream.Collectors.*;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsInAnyOrder;
import org.junit.Test;

@Test
public void shouldFilterCommonlyUsedWords() {
    // given
    List<String> allWords = Arrays.asList(
       "call", "feel", "call", "very", "call", "very", "feel", "very", "any");

    // when
    Set<String> commonlyUsed = allWords.stream()
            .collect(groupingBy(Function.identity(), counting()))
            .entrySet().stream().filter(e -> e.getValue() > 2)
            .map(Map.Entry::getKey).collect(toSet());

    // then
    assertThat(commonlyUsed, containsInAnyOrder("call", "very"));
}

我觉得有可能做得更简单 - 我是对的吗？

我正在使用流分裂器直接用于我正在编写的库中的低级操作.最近,当我使用流分裂器和交错tryAdvance/trySplit调用时,我发现了非常奇怪的行为.这是一个简单的代码,演示了这个问题:

import java.util.Arrays;
import java.util.Spliterator;

public class SpliteratorBug {
    public static void main(String[] args) {
        Integer[][] input = { { 1 }, { 2, 3 }, { 4, 5, 6 }, { 7, 8 }, { 9 } };
        Spliterator<Integer> spliterator = Arrays.stream(input).parallel()
                .flatMap(Arrays::stream).spliterator();
        spliterator.trySplit();
        spliterator.tryAdvance(s -> {});
        spliterator.trySplit();
        spliterator.forEachRemaining(System.out::println);
    }
}

输出是

5
6
9

正如你所看到的,后平映射我应该得到连续编号的有序流1来9.我将分裂器分开一次,所以它应该跳到一些中间位置.接下来我从中消耗一个元素并将其拆分一次.之后我打印所有剩余的元素.我希望我将从流尾部有几个连续的元素(可能是零元素,它也会很好).然而我所得到的是5和6,然后突然跳出来9.

我知道目前在JDK分裂器中并没有这样使用:它们总是在遍历之前分裂.但是官方文档没有明确禁止调用trySplit后tryAdvance.

当我使用直接从集合,数组,生成的源等创建的spliterator时,从未观察到这个问题.只有当spliterator是从具有中间体的并行流创建时才会观察到flatMap.

所以问题是:我是否遇到了这个错误,或者明确禁止某个地方以这种方式使用分裂器？

我不明白以下方法的第三个参数的效用:

<R> R collect(Supplier<R> supplier,
              BiConsumer<R,? super T> accumulator,
              BiConsumer<R,R> combiner)

来自javaDoc:

这产生的结果相当于:

 R result = supplier.get();
 for (T element : this stream)
     accumulator.accept(result, element);
 return result;

Run Code Online (Sandbox Code Playgroud)

如您所见,combiner未使用该参数.例如,以下内容将字符串累积到ArrayList中:

 List<String> asList = stringStream.collect(ArrayList::new, ArrayList::add,
                                            ArrayList::addAll);

但我期待这个:

List<String> asList = stringStream.collect(ArrayList::new, ArrayList::add );

我几天来一直在努力解决这个问题.我正在尝试使用Java Streams创建Pivot功能.我只需要实现SUM,COUNT,MAX,MIN和AVERAGE.对于输入,我给出了一个Pivot Column Index,一个pivot行索引数组和要计算的值.

catch是数据在List <List <Object >>中,Where对象是String,Integer或Double.但直到运行时我才知道.我必须将结果作为List <List <Object >>返回.

我在使用MAX/MIN时遇到问题(我假设AVERAGE与MAX和MIN类似)

为了在多个表值上进行数据透视,我创建了一个类来使用我的第二个groupingBy

这不会编译,我不知道要比较什么,将对象转换为int或者如果我甚至需要.我想用一个流来做这一切,但我不确定它是否可行.我做错了什么,或者我可以做不同的事情.提前致谢.

package pivot.test;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

public class PivotTest {

    List<List<Object>> rows = new ArrayList<List<Object>>();

    public PivotTest() throws Exception {

        rows.add(Arrays.asList(new Object[]{ "East", "Boy", "Tee", 10, 12.00}));
        rows.add(Arrays.asList(new Object[]{ "East", "Boy", "Golf", 15, 20.00}));
        rows.add(Arrays.asList(new Object[]{ "East", "Girl", "Tee", 8, 14.00}));
        rows.add(Arrays.asList(new Object[]{ "East", "Girl", "Golf", 20, 24.00}));
        rows.add(Arrays.asList(new Object[]{ "West", "Boy", "Tee", 5, …

所以我有一个列表,我从中获得一个并行流来填写地图,如下所示:

Map<Integer, TreeNode> map = new HashMap<>();
List<NodeData> list = some_filled_list;

//Putting data from the list into the map
list.parallelStream().forEach(d -> {
                TreeNode node = new TreeNode(d);
                map.put(node.getId(), node);
            });

//print out map
map.entrySet().stream().forEach(entry -> {
     System.out.println("Processing node with ID = " + entry.getValue().getId());
                });

这段代码的问题在于,当"放置数据"过程仍在进行时(因为它是并行的),地图正在被打印出来,因此,地图尚未从列表中接收到所有元素.当然,在我的真实代码中,不仅仅是打印出地图; 我使用地图来利用O(1)查找时间.

我的问题是:

1. 如何使主线程等待,以便在打印出地图之前完成"放置数据"？我试图把"把数据"线程T内,做t.start()和t.join(),但是这并没有帮助.

2. 也许在这种情况下我不应该使用并行流？列表很长,我只想利用并行性来提高效率.

如果我有2个Streams,如下面所示的方法

public Stream<Transaction> getPendingTransaction(Stream<PendingTransaction> pendingTransactionStream,Stream<ProcessedTransaction> processedTransactionStream){ }

我想找到其中存在的所有对象中pendingTransactionStream同样存在于processedTransactionStream基于像一些标准

if transaction.getId()对于存在的Transaction对象是相同的, pendingTransactionStream并且processedTransactionStreamthen该对象是相同的,我们可以在列表中收集它们.

我尝试这样做,但它给出了错误

processedTransactionStream
        .filter( (processedTransaction)->
        {
            pendingTransactionStream.anyMatch(s->s.getTransactionId().equals(processedTransaction.getTransactionId()) );
        } 
        ).collect(Collectors.toList());

目标是在Java 8流的帮助下处理连续的元素流.因此,在处理该流时,将元素添加到并行流的数据源中.

Streams的Javadoc在"无干扰"部分中描述了以下属性:

对于大多数数据源,防止干扰意味着确保在流管道的执行期间根本不修改数据源.值得注意的例外是其源是并发集合的流,这些集合专门用于处理并发修改.并发流源是Spliterator报告CONCURRENT特性的源.

这就是在我们的尝试中使用ConcurrentLinkedQueue的原因,它返回true

new ConcurrentLinkedQueue<Integer>().spliterator().hasCharacteristics(Spliterator.CONCURRENT)

没有明确说明,在并行流中使用时不得修改数据源.

在我们的示例中,对于流中的每个元素,递增的计数器值被添加到队列中,该队列是流的数据源,直到计数器大于N.通过调用queue.stream(),一切正常,顺序执行:

import static org.junit.Assert.assertEquals;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

public class StreamTest {
    public static void main(String[] args) {
        final int N = 10000;
        assertEquals(N, testSequential(N));
    }

    public static int testSequential(int N) {
        final AtomicInteger counter = new AtomicInteger(0);
        final AtomicInteger check = new AtomicInteger(0);
        final Queue<Integer> queue = new ConcurrentLinkedQueue<Integer>();

        for (int i = 0; i < N / 10; ++i) {
            queue.add(counter.incrementAndGet());
        }

        Stream<Integer> …

我有一个班级列表说 ProductDto

public class ProductDto {
    private String Id;
    private String status;
    private Booker booker;
    private String category;
    private String type;
}

我想要一张地图如下: -

Map<String,Map<String,Map<String,Booker>>

属性将映射如下:

Map<status,Map<category,Map<type,Booker>

我知道可以轻松完成一个级别的分组而无需任何麻烦Collectors.groupingBy.

我试图将它用于嵌套级别,但是当我为关键字段开始出现相同的值时,它失败了.

我的代码如下: -

 list.stream()
                .collect(Collectors.groupingBy(
                        (FenergoProductDto productDto) -> 
                        productDto.getStatus()
                        ,
                        Collectors.toMap(k -> k.getProductCategory(), fProductDto -> {
                            Map<String, Booker> productTypeMap = new ProductTypes();
                            productTypeMap.put(fProductDto.getProductTypeName(),
                                    createBooker(fProductDto.getBookingEntityName()));
                            return productTypeMap;
                        })
                ));

如果有人知道通过使用流来做到这一点的好方法,请分享!

我有一个单声道A.对象A包含两个列表.我想直接创建两个Flux.这可能没有block()吗？

Mono<A> a = ... ;

Flux<AList1> a1 =  Flux.fromIterable(a.block().getList1());