标签: java-stream

或者,有没有更好的方法来做到这一点,我错过了？我想基于供应商创建一个流(通常是无限的),但是当供应商返回null时,流终止.我自己做了这个,但是完成一个非常简单的概念似乎是相当多的工作.

public class NullTerminatedStreamFactory {

    static int characteristics = Spliterator.ORDERED | Spliterator.DISTINCT;

    public static<T> Stream<T> makeNullTerminatedStream(Supplier<T> supplier) {
        return StreamSupport.stream(new NullTerminatedSpliteratorFromSupplier<>(supplier, Long.MAX_VALUE, characteristics), false);
    }

    static class NullTerminatedSpliteratorFromSupplier<T> extends Spliterators.AbstractSpliterator<T> {

        public NullTerminatedSpliteratorFromSupplier(Supplier<T> supplier, long est, int additionalCharacteristics) {
            super(est, additionalCharacteristics);
            this.supplier = supplier;
        }

        public Supplier<T> supplier;

        @Override
        public boolean tryAdvance(Consumer<? super T> action) {
            T next = supplier.get();
            if (next != null) {
                action.accept(next);
                return true;
            }
            return false;
        }
    }
}

为了记录,我正在使用它,基本上从BlockingQueue创建一个Stream:

NullTerminatedStreamFactory.makeNullTerminatedStream(() -> {
            try { …

我知道这是一个菜鸟问题，但我在其他任何地方都找不到简单的答案。问题是：我需要编写一个返回 a 的方法SortedMap，因此树图应该可以正常工作。我有一个HashMap< String, Skill>，Skill该类具有方法getName和getNumApplicants我需要返回一个SortedMap<String, Long>，以技能名称为键，以申请人数为值。这是我的立场：

private Map<String,Skill> skillMap = new HashMap<>();

public SortedMap<String, Long> skill_nApplicants() {

    return skillMap.values().stream().collect(...);
}

这是技能课

public class Skill {

    private String name;
    private List <Position> reqPosition = new ArrayList<>();
    private Long numApplicants;

    public void plusOneApplicant() {
        this.numApplicants++;
    }

    public Long getNumApplicants() {
        return numApplicants;
    }
    public Skill(String name) {
        super();
        this.name = name;
        this.numApplicants = 0L;
    }
    public String getName() {
        return …

我有一张地图Map<K, V>，我的目标是删除重复的值并Map<K, V>再次输出相同的结构。如果重复的值被发现，必须有一个选择键（k从两个键（）k1和k1）持有这些值，因为这个原因，假设BinaryOperator<K>给k从k1和k2可用。

示例输入和输出：

// Input
Map<Integer, String> map = new HashMap<>();
map.put(1, "apple");
map.put(5, "apple");
map.put(4, "orange");
map.put(3, "apple");
map.put(2, "orange");

// Output: {5=apple, 4=orange} // the key is the largest possible

用我的尝试Stream::collect(Supplier, BiConsumer, BiConsumer)是位非常笨拙，包含可变操作，比如Map::put和Map::remove我想避免：

// // the key is the largest integer possible (following the example above)
final BinaryOperator<K> reducingKeysBinaryOperator = (k1, …

AtomicInteger recordNumber = new AtomicInteger();
Files.lines(inputFile.toPath(), StandardCharsets.UTF_8)
     .map(record -> new Record(recordNumber.incrementAndGet(), record)) 
     .parallel()           
     .filter(record -> doSomeOperation())
     .findFirst()

当我写这篇文章时，我假设线程将只产生 map 调用，因为 parallel 放在 map 之后。但是文件中的某些行在每次执行时都获得了不同的记录号。

我阅读了官方Java 流文档和一些网站，以了解流在幕后是如何工作的。

几个问题：

• Java 并行流基于SplitIterator 工作，它由每个集合（如 ArrayList、LinkedList 等）实现。当我们从这些集合中构造并行流时，将使用相应的拆分迭代器来拆分和迭代集合。这解释了为什么并行发生在原始输入源（文件行）级别而不是映射的结果（即记录 pojo）。我的理解正确吗？

• 就我而言，输入是文件 IO 流。将使用哪个拆分迭代器？

• 我们parallel()在管道中的位置并不重要。原始输入源将始终被拆分，并将应用剩余的中间操作。

  在这种情况下，Java 不应该允许用户在管道中的任何地方放置并行操作，除了原始源。因为，对于那些不知道 java 流内部如何工作的人来说，这是错误的理解。我知道parallel()已经为 Stream 对象类型定义了操作，因此它以这种方式工作。但是，最好提供一些替代解决方案。

• 在上面的代码片段中，我试图为输入文件中的每条记录添加一个行号，因此应该对其进行排序。但是，我想doSomeOperation()并行应用，因为它是重量级逻辑。实现的一种方法是编写我自己的自定义拆分迭代器。有没有其他办法？

JDK 正在引入一个Stream.toList()带有JDK-8180352的 API 。这是一个基准代码，我试图将其性能与现有的进行比较Collectors.toList：

@BenchmarkMode(Mode.All)
@Fork(1)
@State(Scope.Thread)
@Warmup(iterations = 20, time = 1, batchSize = 10000)
@Measurement(iterations = 20, time = 1, batchSize = 10000)
public class CollectorsVsStreamToList {

    @Benchmark
    public List<Integer> viaCollectors() {
        return IntStream.range(1, 1000).boxed().collect(Collectors.toList());
    }

    @Benchmark
    public List<Integer> viaStream() {
        return IntStream.range(1, 1000).boxed().toList();
    }
}

结果总结如下：

Benchmark                                                       Mode  Cnt   Score    Error  Units
CollectorsVsStreamToList.viaCollectors                         thrpt   20  17.321 ±  0.583  ops/s
CollectorsVsStreamToList.viaStream                             thrpt   20  23.879 ±  1.682  ops/s
CollectorsVsStreamToList.viaCollectors                          avgt   20   0.057 ± …

为什么在 Kotlin 中有多种方法可以做同样的事情，这是有原因的吗？

val viaSequence = items.asSequence()
    .filter { it%2 == 0 }
    .map { it*2 }
    .toList()
println(viaSequence)

val viaIterable = items.asIterable()
    .filter { it%2 == 0 }
    .map { it*2 }
    .toList()
println(viaIterable)

val viaStream = items.stream()
    .filter { it%2 == 0 }
    .map { it*2 }
    .toList()
println(viaStream)

我知道以下代码在每个步骤上都会创建一个列表，这会增加 GC 的负载，因此应该避免：

items.filter { it%2 == 0 }.map { it*2 }

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8);
List<Integer> twoEvenSquares = numbers.stream().filter(n -> {
    System.out.println("filtering " + n);
    return n % 2 == 0;
}).map(n -> {
    System.out.println("mapping " + n);
    return n * n;
}).limit(2).collect(Collectors.toList());


for(Integer i : twoEvenSquares)
{
    System.out.println(i);
}

当执行时,输出下面的逻辑来了

filtering 1
filtering 2
mapping 2
filtering 3
filtering 4
mapping 4
4
16

如果流遵循短路概念(我们使用限制流操作),则输出必须如下所示:

filtering 1
filtering 2
filtering 3
filtering 4
mapping 2
mapping 4
4
16

因为在过滤2之后,我们还要找到一个元素来分层限制(2),操作,那么为什么输出不像我解释的那样？

我想通过一个分类器使用Java 8 Stream和Group,但是有多个Collector函数.因此,在分组时,例如计算一个场(或可能是另一个场)的平均值和总和.

我尝试用一​​个例子来简化这一点:

public void test() {
    List<Person> persons = new ArrayList<>();
    persons.add(new Person("Person One", 1, 18));
    persons.add(new Person("Person Two", 1, 20));
    persons.add(new Person("Person Three", 1, 30));
    persons.add(new Person("Person Four", 2, 30));
    persons.add(new Person("Person Five", 2, 29));
    persons.add(new Person("Person Six", 3, 18));

    Map<Integer, Data> result = persons.stream().collect(
            groupingBy(person -> person.group, multiCollector)
    );
}

class Person {
    String name;
    int group;
    int age;

    // Contructor, getter and setter
}

class Data {
    long average;
    long sum;

    public Data(long average, long …

运行以下代码示例以:
"线程中的异常"main"java.lang.StackOverflowError"结束

import java.util.stream.IntStream;
import java.util.stream.Stream;

public class TestStream {

    public static void main(String[] args) {
        Stream<String> reducedStream = IntStream.range(0, 15000)
            .mapToObj(Abc::new)
            .reduce(
                Stream.of("Test")
                , (str , abc) -> abc.process(str)
                , (a , b) -> {throw new IllegalStateException();}
        );
        System.out.println(reducedStream.findFirst().get());
    }

    private static class Abc { 
        public Abc(int id) {
        }

        public Stream<String> process(Stream<String> batch) {
            return batch.map(this::doNothing);
        }

        private String doNothing(String test) {
            return test;
        }
    }
}

究竟是什么导致了这个问题？这段代码的哪一部分是递归的,为什么？

是否有一个操作的任何保证连续和有序流的处理,在遭遇订单？

我的意思是,如果我有这样的代码:

IntStream.range(0, 5)
        .map(i -> {
            myFunction(i);
            return i * 2;
         })
         .boxed()
         .collect(toList());

是否有保证,它将以生成范围的遭遇顺序执行myFunction()调用？

我找到了Stream类的JavaDocs草案,明确说明了这一点:

对于顺序流管道,如果管道源具有已定义的遭遇顺序,则所有操作都以管道源的遭遇顺序执行.

但在官方JavaDocs中,这一行被删除了.它现在仅讨论所选方法的遭遇顺序.该包装java.util.stream文档的副作用款规定:

即使管道被约束以产生与流源的遭遇顺序一致的结果(例如,IntStream.range(0,5).parallel().map(x -> x*2).toArray()必须产生[0, 2, 4, 6, 8]),也不保证将映射器函数应用于各个元素的顺序,或者什么线程为给定元素执行任何行为参数.

但它没有说明顺序流,而且这个例子是针对并行流的(我的理解是顺序流和并行流都是如此,但这是我不确定的部分).

另一方面,它也在订购部分说明:

如果订购了流,则大多数操作都被约束为对其遭遇顺序中的元素进行操作; 如果流的源是List包含的[1, 2, 3],那么执行的结果map(x -> x*2)必须是[2, 4, 6].但是,如果源没有定义的遭遇顺序,则值的任何排列[2, 4, 6]都将是有效结果.

但这一次它以"对元素进行操作"开始,但是示例是关于产生的流,所以我不确定它们是否会考虑副作用,副作用实际上就是这个问题.