Operation Manual: Stream Processing Manual

This article has participated in the good article call order activity, click to see: back end, big front end double track submission, 20,000 yuan prize pool for you to challenge!

>>>> 😜😜😜 Github: 👉 github.com/black-ant CASE Domestic backup: 👉 gitee.com/antblack/ca…

A. The preface

Java streams are a feature that has been around for a long time and can greatly reduce our code, and parallel processing can greatly improve performance by using multiple processor cores in some scenarios.

While Stream syntax can be tricky for beginners to use, this article takes a look at this feature from the shallow to the deep.

This is the beginning, only record the previous combing usage, the next article to see the source code, remember to collect!!!!!

The characteristics of the Stream

• Stream is not a collection, nor is it a data structure, and cannot hold data
• A Stream is somewhat similar to a high-level Iterator and can be used for algorithms and calculations
• Unlike iterators, streams can be parallelized, and the data is divided into segments that are processed in different threads
• Data source, zero or more intermediate operations, and zero or one terminal operation
• All intermediate operations are inert and have no effect until the pipe starts working
• The terminal operation is a bit like a faucet. When the faucet is opened, the water flows and the intermediate operation is performed

Ii. Basic knowledge

2.1 Structure operation

2.1.1 Double colon operation

The double colon operation is to pass the method as an argument to the desired method (Stream), which is a method reference

Example 1: Basic usage

x -> System.out.println(x)
// ------------ 
System.out::println
Copy the code

Case 2: Complex usage

for (String item: list) {
    AcceptMethod.printValur(item);
}
//------------------
list.faorEach(AcceptMethod::printValur);
Copy the code

2.2 Stream creation

2.2.1 Collection and array tools

Based on the case

/ / Collection tools
Collection.stream () : list.stream();

Stream.<String>builder().add("a").add("b").add("c").build();

Stream.of("a"."b"."c")

Stream.generate(() -> "element").limit(10);

Stream.iterate(40, n -> n + 2).limit(20);

Copy the code

Create an integer stream

IntStream.rangeClosed(1.100).reduce(0, Integer::sum);
IntStream.rangeClosed(1.100).parallel().reduce(0, Integer::sum);

// Other basic types of cases
LongStream.rangeClosed(1.3);


Copy the code

Create a parallel flow

// API :
Stream<E> parallelStream(a)

/ / case:
Collection.parallelStream (a)
listOfNumbers.parallelStream(a).reduce(5, Integer::sum);

listOfNumbers.parallelStream().forEach(number ->
    System.out.println(number + "" + Thread.currentThread().getName())
);

Copy the code

Array creation stream

Arrays.stream(intArray).reduce(0, Integer::sum);
Arrays.stream(intArray).parallel().reduce(0, Integer::sum);
Arrays.stream(integerArray).reduce(0, Integer::sum);
Arrays.stream(integerArray).parallel().reduce(0, Integer::sum);
Copy the code

Combined flow

// API: combine two Streams
 <T> Stream<T> concat(Stream<? extends T> a, Stream<? extends T> b)
     
     
/ / case
Stream<Integer> stream1 = Stream.of(1.3.5);
Stream<Integer> stream2 = Stream.of(2.4.6);

Stream<Integer> resultingStream = Stream.concat(stream1, stream2);


// Example: merge three
Stream.concat(Stream.concat(stream1, stream2), stream3);

// Example: stream of merge streamStream.of(stream1, stream2, stream3, stream4)Copy the code

Other cases

// Static factory
1.  Java.util.stream.IntStream.range ( )
2.  Java.nio.file.Files.walk ( )

// Create manually
1.  java.util.Spliterator
2.  Random.ints()
3.  BitSet.stream()
4.  Pattern.splitAsStream(java.lang.CharSequence)
5.  JarFile.stream()
    
// java.io.BufferedReader.lines()
Files.lines(path, Charset.forName("UTF-8"));
Files.lines(path);
Copy the code

supplement

Split flow Case

2.3 Stream Operations

A flow can have multiple Intermediate operations and one Terminal operation, and when the Terminal completes, the flow ends

2.3.1 Intermediate Operation of a flow

Map: indicates the element mapping

// API :
<R> Stream<R> map(Function<? super T, ? extends R> mapper)
    


// Map is passed in the method function, and the Map returns an object
books.stream(a).filter(e -> "Effective Java".equals(e.getValue())).map(Map.Entry::getKey).findFirst(a);

wordList.stream().map(String::toUpperCase).collect(Collectors.toList());

Stream.of(1.2.3).map(n -> n + 1).collect(Collectors.toList());

nums.stream().map( n -> n * n ).collect (Collectors.toList());


Copy the code

flatMap

// flatMap returns a Stream
Stream<List<String>> namesOriginalList = Stream.of(
	Arrays.asList("Pankaj"), 
	Arrays.asList("David"."Lisa"),
	Arrays.asList("Amit"));
//flat the stream from List<String> to String stream
Stream<String> flatStream = namesOriginalList
	.flatMap(strList -> strList.stream());

flatStream.forEach(System.out::println);


Copy the code

mapToXXX

// API : 
IntStream mapToInt(ToIntFunction<? super T> mapper)

MapToXXX is mainly used to convert to

doubleNumbers.stream(a).mapToDouble(Double::doubleValue).sum(a);

customers.stream().mapToInt(Customer::getAge).filter(c -> c > 65).count();

intStream1.mapToObj(c -> (char) c);

Copy the code

A new stream is generated from the elements that are filtered

Predicate is a functional interface
Stream<T> filter(Predicate<? super T> predicate)

// Use the arrow expression in the filter
- tream<Integer> oddIntegers = ints.stream().filter(i -> i.intValue() % 2! =0);
- list.stream().filter(p -> p.startsWith("j")).count()

// Double colons are used in the Filter ::
customers.stream().filter(Customer::hasOverHundredPoints).collect(Collectors.toList());

// Use code blocks in Filter
 customers.stream().filter(c -> {
      try {
          return c.hasValidProfilePhoto();
      } catch (IOException e) {
          //handle exception
      }
      return false;
}).collect(Collectors.toList());
Copy the code

Distinct: go to heavy

nums.stream().filter(num -> num % 2= =0).distinct().collect(Collectors.toList());
list.stream().distinct().collect(Collectors.toList())
Copy the code

Sorted: sorting

// Customize the sorting mode
persons.stream().limit(2).sorted((p1, p2) -> p1.getName().compareTo(p2.getName())).collect(Collectors.toList());

// Use the collator provided by the specified Comparator
List<String> reverseSorted = names2.sorted(Comparator.reverseOrder()).collect(Collectors.toList());

// Use default sort with no parameters
List<String> naturalSorted = names3.sorted().collect(Collectors.toList());
Copy the code

peek

// API: can be used for debugging, mainly to intercept the flow as it passes through a certain point in the pipeline
Stream<T> peek(Consumer<? super T> action)
    
/ / case:
IntStream.range(1.10).peek(System.out::println).sum(a);

// Intercept at multiple intercept points
Stream.of("one"."two"."three"."four")
	.filter(e -> e.length() > 3)
	.peek(e -> System.out.println("Filtered value: " + e))
	.map(String::toUpperCase)
	.peek(e -> System.out.println("Mapped value: " + e))
	.collect(Collectors.toList());

Copy the code

Limit: limit

// API: the number of streams to truncate. You can see that a stream is still returned
Stream<T> limit(long maxSize);

/ / case:
nums.stream().filter(n-> n>2).limit(2).collect(Collectors.toList ())
Copy the code

Skip, skip

// API : 
Stream<T> skip(long n);

/ / case:
nums. stream() .filter(n-> n>2 ).skip (2) . collect( Collectors . toList () );
Copy the code

The parallel parallel flows

// API: returns a parallel equivalent stream, or itself if already parallel
S parallel(a)
boolean isParallel(a)

/ / case:
Object[] listOutput = list.stream().parallel().toArray();
Copy the code

Sequential: a serial stream

// API :
S sequential(a);

/ / case:
Arrays.asList(1.2.3.4).stream().sequential();
Copy the code

Unordered: indicates the unordered state

// Eliminate the encounter order, so that parallel performance can be committed
IntStream.range(1.1 _000_000).unordered().parallel().distinct().toArray();
Copy the code

2.3.2 Aggregation of Terminal operations of streams

Foreach: loop traversal

// API: As you can see, a Consumer function is received
void forEach(Consumer<? super T> action);

// Use the arrow function in foreach
roster.stream().forEach(p -> System.out.println(p.getName()));
Copy the code

ForEachOrdered: A circular flow that is ordered

list.stream().parallel().forEachOrdered(e -> logger.info(e));
Copy the code

Array

stream.toArray(String[]::new);

//reduce: combine the Stream elements
Stream.of("A"."B"."C"."D").reduce("", String::concat);

/ / reduce sum
Stream.of(5.6.7.8).reduce(0, (accumulator, element) -> accumulator + element); ? - Parameters after reduce: the first default value followed by the passed method// min: maximizes the string array
Stream.of(testStrings).max((p1, p2) -> Integer.compare(p1.length(), p2.length()));

// Max: gets the maximum length
br.lines().mapToInt(String::length).max().getAsInt();
Copy the code

collection

• Stream.collect (aggregators.tolist ()) : toList collects all elements of the stream into the List
• Stream.collect (Collectors. ToCollection (ArrayList::new)): Collects the elements of a stream into a collection created by a given source
• Stream.collect (collects.toset ()) : Saves all elements in the stream to a Set, removing the duplicate lock
• stream.collect(Collectors.toCollection(Stack::new))

// API
<R> R collect(Supplier<R> supplier,BiConsumer<R, ? super T> accumulator,BiConsumer<R, R> combiner);
<R, A> R collect(Collector<? super T, A, R> collector);




Map<String, Integer> hashMap = list.stream().collect(Collectors
    .toMap(Function.identity(), String::length));

Map<String, Integer> linkedHashMap = list.stream().collect(Collectors.toMap(
    Function.identity(),
    String::length,
    (u, v) -> u,
    LinkedHashMap::new
));

// Create a Collection object
Stream<Integer> intStream = Stream.of(1.2.3.4);
List<Integer> intList = intStream.collect(Collectors.toList());
System.out.println(intList); //prints [1, 2, 3, 4]

intStream = Stream.of(1.2.3.4); //stream is closed, so we need to create it again
Map<Integer,Integer> intMap = intStream.collect(Collectors.toMap(i -> i, i -> i+10));
System.out.println(intMap); //prints {1=11, 2=12, 3=13, 4=14}


// Create an Array object
Stream<Integer> intStream = Stream.of(1.2.3.4);
Integer[] intArray = intStream.toArray(Integer[]::new);
System.out.println(Arrays.toString(intArray)); //prints [1, 2, 3, 4]

/ / String operations
stream.collect(Collectors.joining()).toString();
list.stream().collect(Collectors.joining("|")) : list.stream().collect(collection.joining ()"| |"."Start--"."--End")) : the middle and before and after the hyphen// Create a Map
books.stream().collect(Collectors.toMap(Book::getIsbn, Book::getName));
// ps : Collector
      
       > toMap(Function
        keyMapper,Function
        valueMapper)
      ,>
    

Copy the code

2.3.3 Calculation of Terminal operation of streams

Min: Returns the smallest element of this stream

// API
Optional<T> min(Comparator<? super T> comparator)Collector<T, ? , Optional<T>>minBy(Comparator<? super T> comparator)

/ / case:
list.stream(a).min(Comparator.comparing(String::valueOf)).ifPresent(e -> System.out.println("Min: " + e));
Copy the code

Max: Returns the largest element of this stream

// API 
Optional<T> max(Comparator<? super T> comparator); Collector<T, ? , Optional<T>> maxBy(Comparator<?super T> comparator) 
    
/ / case:
list.stream().max(Comparator.comparing(String::valueOf)).ifPresent(e -> System.out.println("Max: " + e));

Copy the code

Count: Counts the number of items in the stream

// API : <T> Collector<T, ? , Long>javacounting()/ / case:
Stream.of(1.2.3.4.5).count();
Copy the code

reduce

// API: Reduce is used to perform computation in Stream
Optional<T> reduce(BinaryOperator<T> accumulator); Collector<T, ? , T> reducing(T identity, BinaryOperator<T> op)U reduce(U identity,BiFunction<U, ? super T, U> accumulator,BinaryOperator<U> combiner); Collector<T, ? , U> reducing(U identity,Function<?super T, ? extends U> mapper,BinaryOperator<U> op) 

// The parameters are as follows:Identity: reduced identifier value (also returned with no input element) accumulator: Operation performed// Use case
numbers.reduce((i,j) -> {returni*j; }); numbers.stream().reduce(0, (subtotal, element) -> subtotal + element);
numbers.stream().reduce(0, Integer::sum);

// Association string
letters.stream().reduce("", (partialString, element) -> partialString + element);

// Case correlation
letters.stream().reduce("", (partialString, element) -> partialString.toUpperCase() + element.toUpperCase());

ages.parallelStream().reduce(0, a, b -> a + b, Integer::sum);


// Parallel operation: parallel processing operations must comply with the following operations
1.The result is not affected by the order of operands2.Non-interference: Operations do not affect the data source3.Stateless and deterministic: Operations are stateless and generate the same output for a given input userList.parallelStream().reduce()0, (partialAgeResult, user) -> partialAgeResult + user.getAge(), Integer::sum);

Copy the code

2.4 Search for Terminal operations of streams

AnyMatch: Any element meets the matching condition

// API: Return true as long as one of the elements in Stream matches the predicate passed in
 boolean anyMatch(Predicate<? super T> predicate);


/ / case:
persons.stream(). anyMatch(p -> p.getAge() < 12);
Copy the code

AllMatch: All elements meet the matching condition

// API: Return true for all elements of Stream that match the predicate passed in
boolean allMatch(Predicate<? super T> predicate);

/ / case:
persons.stream(). allMatch(p -> p.getAge() > 18);
Copy the code

FindFirst: Returns the first element in Stream

// API: returns an Optional identifier for the first element
Optional<T> findFirst(a);

/ / case:
students.stream().filter(student ->student .getage()>20 ).findFirst();	
Copy the code

FindAny: Returns any element in Stream

// API: findAny does not necessarily return the first, but either, or an empty Optional if the stream is empty
Optional<T> findAny(a);
Copy the code

NoneMatch: None of the elements meet the match condition

// API: Returns true if none of the elements in Stream match the predicate passed in
boolean noneMatch(Predicate<? super T> predicate);

/ / case:
numbers5.noneMatch(i -> i==10)

Copy the code

2.5 Stream specification

// Reduce: perform further operations on the parameterized collection
    
students.stream().filter(student -> "Computer Science".equals(student.getMajor())).map(Student::getAge).reduce(0, (a, b) -> a + b);

students.stream().filter(student -> "Computer Science".equals(student.getMajor())).map(Student::getAge).reduce(0, Integer::sum);

students.stream().filter(student -> "Computer Science".equals(student.getMajor())).map(Student::getAge).reduce(Integer::sum);

Copy the code

2.6 Grouping streams By

Single-stage grouping

// API : 
public static<T, K> Collector<T, ? , Map<K, List<T>>>groupingBy(Function<?super T, ? extends K> classifier)

students.stream().collect(Collectors.groupingBy(Student::getSchool))
Copy the code

Multistage grouping

/ / function:

/ / case:
students.stream().collect(
    Collectors.groupingBy(Student::getSchool,
    Collectors.groupingBy(Student::getMajor)));
Copy the code

PartitioningBy: specifies a partition. Unlike groupBy, partitions can only be true and false.

// API: As you can see, this is mainly Predicate functionsCollector<T, ? , Map<Boolean, List<T>>> partitioningBy(Predicate<?superT> predicate) Collector<T, ? , Map<Boolean, D>> partitioningBy(Predicate<?super T> predicate,Collector<? super T, A, D> downstream)
    
 / / case:
students.stream().collect(Collectors.partitioningBy(student -> "Wuhan University".equals(student.getSchool())));
Copy the code

3. Use in-depth

3.1 Extending the thread pool

// Set thread pool globally
-D java.util.concurrent.ForkJoinPool.common.parallelism=4

// Manually set this thread
ForkJoinPool customThreadPool = new ForkJoinPool(4);
int sum = customThreadPool.submit(
    () -> listOfNumbers.parallelStream().reduce(0, Integer::sum)).get();
customThreadPool.shutdown();
    
Copy the code

3.2 the Debug process

4. Common cases

Example 1: Each element in List AList is converted to generate another type C and put it into List BList

List<JSONObject> itemjson = new LinkedList<JSONObject>();
List<A> aList = ... 
itemCW.stream().map(c -> {
            JSONObject nodeitem = new JSONObject();
            nodeitem.put("whtype".0);
            return nodeitem;
}).forEach(c -> itemjson.add(c));
Copy the code

Case 2: Loop through a Map

       realmTO.getTemplates().forEach((key, template) -> {
            AnyType type = anyTypeDAO.find(key);
            anyTemplate.set(template);
        });
// For each, you can pass in the key and object, which can be used later
Copy the code

Case 3: Getting a small collection from a large collection

// Get a collection of ids
List<Long> idList = stockList.stream().map(Stock::getId).collect(Collectors.toList());
// Get the skUID set and deduplicate it
List<Long> skuIdList = stockList.stream().map(Stock::getSkuId).distinct().collect(Collectors.toList());
// Get a set of supplierIds (supplierId = int, return List
      
       , boxed)
      
Set<Integer> supplierIdSet = stockList.stream().mapToInt(Stock::getSupplierId).boxed().collect(Collectors.toSet());
Copy the code

Case 4: Grouping and sharding

// Group by skuid
Map<Long, List<Stock>> skuIdStockMap = stockList.stream().collect(Collectors.groupingBy(Stock::getSkuId));
// Filter supplierId=1 and group by skuId
Map<Long, List<Stock>> filterSkuIdStockMap = stockList.stream().filter(s -> s.getSupplierId() == 1).collect(Collectors.groupingBy(Stock::getSkuId));
// Partitions are divided into unavailable and other fragments by state
Map<Boolean, List<Stock>> partitionStockMap = stockList.stream().collect(Collectors.partitioningBy(s -> s.getStatus() == 0));
Copy the code

Case 5: Counting and Summing

Count the number of records whose skuId=1
long skuIdRecordNum = stockList.stream().filter(s -> s.getSkuId() == 1).count();
// Count the total inventory with skuId=1
BigDecimal skuIdAmountSum = stockList.stream().filter(s -> s.getSkuId() == 1).map(Stock::getAmount).reduce(BigDecimal.ZERO, BigDecimal::add);
Copy the code

Case 6: Specific usage

// Group by supplierId and then by skuId. Sort by supplierId and then skuId
Map<Integer, Map<Long, List<Stock>>> supplierSkuStockMap = stockList.stream().collect(Collectors.groupingBy(Stock::getSupplierId, TreeMap::new,
                Collectors.groupingBy(Stock::getSkuId, TreeMap::new, Collectors.toList())));

// supplierId (supplierId) and skuId (skuId)
// sort (); // sort (); // sort ();
stockList.sort(Comparator.comparing(Stock::getSupplierId)
                .thenComparing(Stock::getSkuId, Comparator.reverseOrder()));
Copy the code

Case 7: Sorting by convection

Collections.sort(literals, (final String t, final String t1) -> {
            if (t == null && t1 == null) {
                return 0;
            } else if(t ! =null && t1 == null) {
                return -1; }});// t1 and t2 are the objects to be compared, and the collation method is defined in them. Return true/false for collation
Copy the code

Case 8: Get the first one after filtering the flow

correlationRules.stream().filter(rule -> anyType ! =null && anyType.equals(rule.getAnyType())).findFirst()
    
 // The key is filter and findFirst
Copy the code

Case 9: Conversion from one set of types to another

        List<String> strings = Lists.transform(list, new Function<Integer, String>() {
            @Override
            public String apply(@Nullable Integer integer) {
                returninteger.toString(); }});Copy the code

Case 10: Traversing a collection and returning a collection

return Stream.of(resources).map(resource -> preserveSubpackageName(baseUrlString, resource, path)).collect(Collectors.toList());


    private String preserveSubpackageName(final String baseUrlString, final Resource resource, final String rootPath) {
        try {
            return rootPath + (rootPath.endsWith("/")?"" : "/")
                + resource.getURL().toString().substring(baseUrlString.length());
        } catch (IOException e) {
            throw newUncheckedIOException(e); }}// Note that the following method is called. The direct anonymous method is not written for the time being

Copy the code

Case 11: Simple splicing

// Splice [x, y, z]
String result1 = stream1.collect(Collectors.joining(","."["."]"));
/ / splicing into x | y | z forms
String result2 = stream2.collect(Collectors.joining("|"."".""));
X -> y -> z]
String result3 = stream3.collect(Collectors.joining("- >"."".""));


(String)value.stream().map((i) -> {
                return this.formatSql("{0}", i);
            }).collect(Collectors.joining(","."(".")"));
Copy the code

Case 12: Complex use

buzChanlList.stream()
.map(item -> {
      return null;
	})
.filter(item -> {
       return isok;
     })
.forEach(c -> contentsList.add(c));
Copy the code

Case 13: Syncopated sets

 List<List<Integer>> splitList = Stream.iterate(0, n -> n + 1).limit(limit).parallel().map(a -> list.stream().skip(a * MAX_NUMBER).limit(MAX_NUMBER).parallel().collect(Collectors.toList())).collect(Collectors.toList());

Copy the code

Case 14: Filter operation

collection.stream().filter(person -> "1".equals(person.getGender())).collect(Collectors.toList())
                           
                           

Copy the code

conclusion

The parallelism of streams can greatly increase efficiency when used properly, and we’ll look at the source code in the next article.

reference

@ www.it610.com/article/129…

@ www.baeldung.com/intellij-de…

@ Java 8 Stream – Java Stream – JournalDev

@ www.baeldung.com/java-stream…