item 46) 스트림에서는 부작용 없는 함수를 사용하라

스트림 :  함수형 스타일 오퍼레이션을 지원하는 클래스

 

스트림의 특징

No storage : 저장을 위한 자료구조가 아니라 소스(자료구조, 배열, I/O 채널 등)로부터 자료를 받아서 계산을 위한 파이프라인

함수형 : 스트림의 연산(operation) 은 결과를 생산만 하지, 소스를 수정하지 않음 ex) 스트림을 필터링한다고 해서 기존 리스트의  원소를 제거 하지는 않음

Laziness-Seeking : 대부분의 스트림 연산(filter, map, 중복 제거 등) 은 lazy 하게 작동한다. (스트림 연산이 최종 연산이 호출되는 시점까지 지연됨)

 List<Integer> strings = List.of(1,2,3,4,5);
        strings.stream()
                .map(x->{
                    System.out.println(x);
                    return x*x+1;
                }).filter(x -> x>3);
                
                //종단 연산이 없으면 중간 연산은 일어나지 않는다.

 

Possibly unbounded : 컬렉션은 크기가 한정되있는 반면 스트림은 그렇지 않다. 무한한 스트림에 대해서 short-circuting 연산을 finite 시간에 연산이 가능하다.(limit, anyMatch 등등)

Stream<Integer> infiniteStream = Stream.iterate(0, i -> i + 1);
infiniteStream.limit(5).forEach(System.out::println);

Consumable : 스트림의 각 원소는 한번만 방문 가능하다. 재방문 하기 위해서는 새로운 스트림을 만들어야한다.

 

 

- 순수 함수  : 오직 입력만이 결과값에 영향을 주는 함수(다른 외부 가변 상태 참조 X, 함수 스스로도 외부의 상태 변경 X)

- 부수 효과 (side-effect) : 자신의 스코프 밖의 변수 상태 등을 변경하는 메소드, 프로시저 

(A side effect method is a method which modifies some state variable value/arguments passed having a consequence beyond its scope,)

 

side-effect의 예

• Logging to the console/file
• Writing to the screen
• Writing to a file/network
• Triggering any external process — db calls
• Invoking any other functions with side-effects

#Good : Not changing the state of the parameters passed, returning a value from the method.

#Bad : Changing the state of the parameters passed, however method does _not_ return a value.

#The Ugly: Changing the state of the parameters passed and also returning a value from the method.

 

  // 스트림을 가장한 반복적 코드(외부 변수 freq를 변경함)
  
    Map<String, Long> freq = new HashMap<>();
   
    try(
   
    Stream<String> words = new Scanner(file).tokens())
            words.forEach(word -> {
                freq.merge(word.toLowerCase(), 1L, Long::sum);
            });
    }

   Map<String, Long> freq;
        try (Stream<String> words = new Scanner(file).tokens()) {
            freq = words.collect(groupingBy(String::toLowerCase, counting()));
        }

 종단 연산(forEach 등)은 앞에서 수행한 연산 결과를 보여주는 일 이상을 하면 안된다. 

 

side effect가 없는 함수를 써야하는 이유

 

차이는 병렬 스트림에서 발생하다. interenc 함수를 보면 peek에서 외부의 상태(listOfStrings)를 변경한다.

하지만 이 변경은 최종 연산인 get()이 호출 되기 전까지 일어나지 않는다. (lazy exectuion)

그러므로 여러 쓰레드가 get를 호출하는 시점에서 listOfStrings의 상태를 바꾸려고 하니 ConcurrentConcurrentModificationException 이 발생하는 것이다.

(참고로 ConcurrentModificationException은 어떤 쓰레드가 iterator가 진행중인 Collection을 수정하려고 할 때 발생한다.) 즉 락이 걸려있는 컬렉션을 수정하려고 할때 발생하는 것 .

public class SideEffect {
    static void interferenc(){
        try {
            List<String> listOfStrings =
                    new ArrayList<>(Arrays.asList("one", "two"));

            // This will fail as the peek operation will attempt to add the
            // string "three" to the source after the terminal operation has
            // commenced.

            String concatenatedString = listOfStrings
                    .stream()

                    // Don't do this! Interference occurs here.
                    .peek(s -> listOfStrings.add("three"))

                    .reduce((a, b) -> a + " " + b)
                    .get();

            System.out.println("Concatenated string: " + concatenatedString);

        } catch (Exception e) {
            System.out.println("Exception caught: " + e.toString());
        }
    }

    static void func1(){
        List<Integer> matched = new ArrayList<>();
        List<Integer> elements = new ArrayList<>();

        for(int i=0 ; i< 10000 ; i++) {
            elements.add(i);
        }

         elements.parallelStream()
                .forEach(e -> {
                    if(e>=100) {
                        System.out.println(Thread.currentThread().getId() +  " " + matched.size());
                        matched.add(e);
                    }
                });
                
                /*
                
                31 0
                31 1
                31 2
                30 0
                34 0
                30 4
                22 0
                27 0
                ...
                */
        System.out.println(matched.size());




    }
    static void func2(){
        List<Integer> elements = new ArrayList<>();
        for(int i=0 ; i< 10000 ; i++) {
            elements.add(i);
        }
        List<Integer> matched = elements.parallelStream()
                .filter(e -> e >= 100)
                .collect(Collectors.toList());
        System.out.println(matched.size());
    }

    public static void main(String[] args) {

        interferenc();
        System.out.println("-----------");
        System.out.println();
       // func1();
        System.out.println("-----------");
        System.out.println();
      //  func2();

    }
}

 

 

이번엔 func 1, func2를 비교해보자.

func1에서는 forEach에서 외부 변수인 matched의 상태를 바꾸려고 한다. 

결과를 보면 알수 있듯이 여러 쓰레드가 mathced를 접근하고 있고 별다른 lock이 걸려있지 않기 때문에 matched size가 매 실행마다 다른 것이다.

병렬 스트림에서 스트림 연산을 쓰레드 안전하게 사용할수 있는 것이 collect 함수이다.(In particular, the collect method is designed to perform the most common stream operations that have side effects in a parallel-safe manner)

 

Collectors : 스트림 종단 연산으로 스트림의 원소를 한데 모아 컬렉션으로 변환할수 있음(List , Map, Set 등등)

java.util.stream.Collectors 클래스의 메소드를 할용 (toList 등등);

 

toList() 리스트 변환

toSet() Set 변환

toCollection(collectionFactory) 프로그래머가 정의한 컬렉션 타입 변환

 

toMap() 맵으로 변환

스트림 원소를 키로 매핑하는 keyMapper와 값을 매핑하는 valueMapper함수를 인수로 받음

Collector<T,?,Map<K,U>> toMap(Function<? super T, ? extends K> keyMapper, Function<? super T,? extends U> valueMapper)

private static final Map<String, Operation> stringToEnum = 
    Stream.of(values()).collect(
        toMap(Obejct::toString, e->e));

 

스트림의 중복 원소가 있을 경우 key가 중복되어 IllegalStateException이 발생

충돌을 방지 하기 위해 병합 함수(merge function)를 인수로 제공

 

ex) 같은 키가 들어왔을때 | 로 value를 묶어줌

  Stream<String> s = Stream.of("apple", "banana", "apricot", "orange", "apple");
    Map<Character, String> m = s.collect(Collectors.toMap(s1 -> s1.charAt(0), s1 -> s1, (oldVal, newVal) -> oldVal + "|" + newVal)); 
    
    
 
    /*
    a apple|apricot|apple
	b banana
	o orange
    */

 

BinaryOperator( 인자 2개를 받고 1개를 리턴하는 함수형 타입)을 넘겨 줘서 기존 원소와 비교하는 merge 함수를 넘길수도 있음

    // 음악가와 그 음악가의 베스트 앨범을 연관짓는 예제
    Map<Artist, Album> topHits = albums.collect(
    // BinaryOperator 정적 임포트한  maxBy 정적패터리 메서드 사용
    toMap(Album::artist, a->a, maxBy(comparing(Album::sales))));
    //toMap(keyMapper, valueMapper, (oldVal, newVal) -> newVal) 가장 마지막에 넣은 값이 유효값
    

 

bishonbopanna.medium.com/java-side-effect-methods-good-bad-and-ugly-8ffa697323ec

docs.oracle.com/javase/tutorial/collections/streams/parallelism.html#side_effects