Java 동시성(Concurrentcy) Atomic and ConcurrencyMap Part 3

으아, 귀찮음을 이겨내고 수행한다.

ConcurrentHashMap

  ConcurrentMap 인터페이스에서 설명된 메서드들은 인터페이스의 구현체에서 모두 사용가능하다. 이를 가장 정말로, 최고로 유용하게 구현한 것중 지대하게 중요한 구현체가 있는데 그것이 바로 ConcurrentHashMap이다. 이 구현체는 맵에 대해 병행 연산을 수행하기 위해 더 진보된, 강화된 새 메서드들을 탑재하고 있다.

  병행 스트림(Parallel streams)과 마찬가지로 이러한 메서드들은 자바 8에서 제공되는 ForkJoinPool.commonPool()을 통한 특별한 ForkJoinPool을 사용한다. 이 풀은 가용가능한 코어 수에 적합한 미리 정의된 병행유사성(?;parallelism;아 어휘의 한계)을 사용한다. 4개의 코어가 탑재된 본인의 머신에서는 이 값은 3으로 출력된다.

System.out.println(ForkJoinPool.getCommonPoolParallelism());  // 3

  이 값은 JVM 파라미터 설정을 통해 증가되거나 감소될 수 있다.

-Djava.util.concurrent.ForkJoinPool.common.parallelism=5

  이번에도 아래의 코드를 기반으로 하여 메서드들을 살펴볼 것이다. 하지만 이번엔 구현체인 ConcurrentHashMap을 통해 모든 것을 수행할 것이다. 인터페이스 ConcurrentMap을 직접 사용하지 않으므로 해당 클래스의 public 지시 한정 메서드들을 모두 사용할 수 있다.

ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();

map.put("foo", "bar");

map.put("han", "solo");

map.put("r2", "d2");

map.put("c3", "p0");

  자바 8에서는 병행 연산으로 3가지가 제공된다. forEach, search 그리고 reduce. 각각의 연산은 키와 값, 요소(entry) 그리고 키-값 쌍의 4가지 형태로 접근해서 사용하는 것이 가능하다.

  이 모든 메서드들은 첫 번째 인자로 parallelismThreshold라 불리는 값을 취한다. 이 threshold는 한 스레드에서 최소 어느 정도의 컬렉션 사이즈로 연산이 수행되어야 하는지의 값을 명시한다. 가령 이 값을 500으로 잡았고 맵의 크기가 499라면, 모든 연산은 1개의 스레드로만 순차적으로 수행될 것이다. 예시에서는 모두 이 값을 1로 지정정하여 모든 연산이 모두 병행적으로 수행되도록 할 것이다.

ForEach

  forEach() 메서드는 맵의 키-값을 모두 병행적으로 순회(iterate)할 수 있다. BiConsumer 타입의 람다식이 파라미터로 들어가며 해당 BiConsumer의 메서드 인자에는 현재 순환 스텝의 키와 값이 각각 대입된다. 병행 연산이 수행되는지 잘 보이도록 해당 스레드의 이름을 콘솔에 출력하도록 하였다. 여기서 본인의 경우 최대 3개의 스레드가 사용가능하단 점을 명심하길 바란다.

map.forEach(1, (key, value) ->

    System.out.printf("key: %s; value: %s; thread: %s\n",

        key, value, Thread.currentThread().getName()));

// key: r2; value: d2; thread: main

// key: foo; value: bar; thread: ForkJoinPool.commonPool-worker-1

// key: han; value: solo; thread: ForkJoinPool.commonPool-worker-2

// key: c3; value: p0; thread: main

Search

  search() 메서드는 BiFunction 타입을 인자로 취하며 해당 람다식에선 찾고자 하는 키-값의 조건이 나타나면 null 이 아닌 반환값을 리턴하도록 하고 탐색 중인 키-값이 조건에 맞지 않는 경우, 즉 찾는 값이 아닌 경우 null을 리턴하도록 구현해야 한다. null이 아닌 값을 받자마자 결과는 리턴되고 추가적인 탐색 작업은 취소된다. ConcurrentHashMap은 정렬되지 않은 자료형이란 것을 명심하자. 그렇기에 탐색 조건으로 넘겨진 람다식의 구현은 맵의 처리 순서에 의존적이도록 작성되면 안된다. 이 경우, 여러 요소에서 동시에 검색 조건을 만족하는 경우가 나타날 수 있는데 이 경우 결과는 예상할 수가 없게 되기 때문이다.

String result = map.search(1, (key, value) -> {

    System.out.println(Thread.currentThread().getName());

    if ("foo".equals(key)) {

        return value;

    }

    return null;

});

System.out.println("Result: " + result);

// ForkJoinPool.commonPool-worker-2

// main

// ForkJoinPool.commonPool-worker-3

// Result: bar

여기는 맵의 값만을 단독으로 사용하여 검색 조건을 지정한 예시이다.

String result = map.searchValues(1, value -> {

    System.out.println(Thread.currentThread().getName());

    if (value.length() > 3) {

        return value;

    }

    return null;

});

System.out.println("Result: " + result);

// ForkJoinPool.commonPool-worker-2

// main

// main

// ForkJoinPool.commonPool-worker-1

// Result: solo

Reduce

  reduce() 메서드는 Java8 Stream에서 이미 소개했었다. BiFunction 타입의 람다 표현식을 두개 파라미터로 받는다. 첫째 람다식에서는 각각의 키-값 쌍을 가용 타입(any-type)의 단일 값으로 변경하는 작업이 정의되어야 하고, 둘째 람다식에서는 이러한 모든 변환된 값들은 또 최종 하나의 결과로 통합하는 로직이 작성되어야 한다. 여기서 null은 무시된다.

String result = map.reduce(1,

    (key, value) -> {

        System.out.println("Transform: " + Thread.currentThread().getName());

        return key + "=" + value;

    },

    (s1, s2) -> {

        System.out.println("Reduce: " + Thread.currentThread().getName());

        return s1 + ", " + s2;

    });

System.out.println("Result: " + result);

// Transform: ForkJoinPool.commonPool-worker-2

// Transform: main

// Transform: ForkJoinPool.commonPool-worker-3

// Reduce: ForkJoinPool.commonPool-worker-3

// Transform: main

// Reduce: main

// Reduce: main

// Result: r2=d2, c3=p0, han=solo, foo=bar

이제까지 Java 8 Concurrency에 대한 튜토리얼의 3-3의 부분을 즐겨 읽었기를 바란다. 라고 글을 쓰며 저자는 말을 마치고 있다. 모든 코드는 깃에 올라가 있다고 하니 원본을 찾아 맨 밑의 링크를 클릭하면 될 것이다.

[링크]