Java 동시성 컬렉션과 유틸리티

1
// 일반 HashMap: 멀티스레드 환경에서 안전하지 않음
2
Map<String, Integer> map = new HashMap<>();
3

4
// 스레드 1
5
map.put("count", 1);
6

7
// 스레드 2 (동시 실행)
8
map.put("count", 2);  // Race Condition!
9

10
// 최악의 경우 내부 구조가 깨져서 무한 루프 발생 가능 (Java 7 이전)

1
// Hashtable: 모든 메서드가 synchronized
2
Hashtable<String, Integer> table = new Hashtable<>();
3

4
// 문제: 모든 연산이 전체 테이블을 잠금
5
table.put("key1", 1);  // 전체 락
6
table.get("key2");     // 읽기도 전체 락 (비효율적!)

1
// 개념적 구조 (실제 코드는 아님)
2
class ConcurrentHashMap<K, V> {
3
    Segment<K, V>[] segments = new Segment[16];  // 기본 16개 세그먼트
4

5
    static class Segment<K, V> {
6
        ReentrantLock lock = new ReentrantLock();
7
        HashEntry<K, V>[] table;
8
    }
9
}

1
// 간소화된 put 구조
2
public V put(K key, V value) {
3
    int hash = hash(key);
4

5
    for (Node<K,V>[] tab = table;;) {
6
        Node<K,V> f = tabAt(tab, i);  // volatile read
7

8
        if (f == null) {
9
            // 빈 버킷: CAS로 원자적 삽입
10
            if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value))) {
11
                break;  // 성공!
12
            }
13
        } else {
14
            // 충돌 발생: 해당 버킷만 synchronized
15
            synchronized (f) {
16
                // 버킷 내에서만 락 (매우 세밀한 락)
17
                // 체이닝 또는 트리에 삽입
18
            }
19
        }
20
    }
21
}

1
// Hashtable: 전체 락 (최악)
2
synchronized (전체 테이블) {
3
    // 모든 연산
4
}
5

6
// ConcurrentHashMap (Java 7): 세그먼트 락 (중간)
7
// 실제로는 ReentrantLock을 상속한 Segment를 사용
8
segments[i].lock();
9
try {
10
    // 해당 세그먼트만
11
} finally {
12
    segments[i].unlock();
13
}
14

15
// ConcurrentHashMap (Java 8+): 버킷 락 (최고)
16
synchronized (bucket[i]) {
17
    // 해당 버킷만
18
}

1
import java.util.concurrent.ConcurrentHashMap;
2

3
// 사용자 세션 관리
4
public class SessionManager {
5
    private ConcurrentHashMap<String, UserSession> sessions = new ConcurrentHashMap<>();
6

7
    public void createSession(String sessionId, UserSession session) {
8
        // putIfAbsent: 없을 때만 삽입 (원자적)
9
        UserSession existing = sessions.putIfAbsent(sessionId, session);
10
        if (existing != null) {
11
            throw new IllegalStateException("Session already exists");
12
        }
13
    }
14

15
    public void updateLastAccessTime(String sessionId) {
16
        // compute: 키의 값을 원자적으로 계산
17
        sessions.computeIfPresent(sessionId, (id, session) -> {
18
            session.setLastAccessTime(System.currentTimeMillis());
19
            return session;
20
        });
21
    }
22

23
    public void removeExpiredSessions() {
24
        long now = System.currentTimeMillis();
25
        long timeout = 30 * 60 * 1000;  // 30분
26

27
        // 스레드 안전하게 순회하면서 삭제
28
        sessions.entrySet().removeIf(entry ->
29
            now - entry.getValue().getLastAccessTime() > timeout
30
        );
31
    }
32
}
33

34
// 카운터 (원자적 증가)
35
public class UrlClickCounter {
36
    private ConcurrentHashMap<String, Long> clicks = new ConcurrentHashMap<>();
37

38
    public void recordClick(String url) {
39
        // merge: 키가 있으면 함수 적용, 없으면 초기값 삽입
40
        clicks.merge(url, 1L, (oldValue, one) -> oldValue + 1);
41

42
        // 또는 compute 사용
43
        // clicks.compute(url, (k, v) -> (v == null) ? 1L : v + 1);
44
    }
45

46
    public long getClicks(String url) {
47
        return clicks.getOrDefault(url, 0L);
48
    }
49
}

1
ConcurrentHashMap<String, List<String>> map = new ConcurrentHashMap<>();
2

3
// 잘못된 코드: 경쟁 조건 발생!
4
if (!map.containsKey("users")) {
5
    map.put("users", new ArrayList<>());  // 사이에 다른 스레드가 끼어들 수 있음
6
}
7
map.get("users").add("Alice");
8

9
// 올바른 코드: 원자적 연산 사용
10
map.computeIfAbsent("users", k -> new ArrayList<>()).add("Alice");
11
// 주의: 반환된 ArrayList 자체는 thread-safe하지 않으므로,
12
// 동시 수정이 필요하면 CopyOnWriteArrayList 등을 사용해야 한다

1
import java.util.concurrent.CopyOnWriteArrayList;
2

3
public class CopyOnWriteArrayList<E> {
4
    private volatile Object[] array;
5

6
    public boolean add(E e) {
7
        synchronized (lock) {
8
            Object[] oldArray = array;
9
            int len = oldArray.length;
10

11
            // 배열 전체를 복사 (비용이 큼!)
12
            Object[] newArray = Arrays.copyOf(oldArray, len + 1);
13
            newArray[len] = e;
14

15
            // volatile write로 배열 교체
16
            array = newArray;
17
        }
18
        return true;
19
    }
20

21
    public E get(int index) {
22
        // 락 없음! volatile read만
23
        return (E) array[index];
24
    }
25
}

1
// 이벤트 리스너 관리 (전형적인 사용 사례)
2
public class EventBus {
3
    private CopyOnWriteArrayList<EventListener> listeners = new CopyOnWriteArrayList<>();
4

5
    // 리스너 등록: 드물게 발생
6
    public void addListener(EventListener listener) {
7
        listeners.add(listener);  // 배열 복사 발생
8
    }
9

10
    // 이벤트 발생: 매우 자주 발생
11
    public void publishEvent(Event event) {
12
        // 락 없이 빠르게 순회
13
        for (EventListener listener : listeners) {
14
            listener.onEvent(event);
15
        }
16
    }
17
}
18

19
// 설정 관리
20
public class ConfigurationManager {
21
    private CopyOnWriteArrayList<String> allowedIps = new CopyOnWriteArrayList<>();
22

23
    // 설정 변경: 드물게 발생 (관리자가 수동으로)
24
    public void addAllowedIp(String ip) {
25
        allowedIps.add(ip);
26
    }
27

28
    // IP 검증: 매 요청마다 발생 (매우 빈번)
29
    public boolean isAllowed(String ip) {
30
        return allowedIps.contains(ip);  // 락 없이 빠르게 확인
31
    }
32
}

1
import java.util.concurrent.BlockingQueue;
2
import java.util.concurrent.ArrayBlockingQueue;
3

4
public class ProducerConsumer {
5
    private BlockingQueue<Task> queue = new ArrayBlockingQueue<>(10);
6

7
    // Producer 스레드
8
    public void producer() throws InterruptedException {
9
        Task task = createTask();
10
        queue.put(task);  // 큐가 꽉 차면 대기 (블록)
11
    }
12

13
    // Consumer 스레드
14
    public void consumer() throws InterruptedException {
15
        Task task = queue.take();  // 큐가 비면 대기 (블록)
16
        process(task);
17
    }
18
}

1
public class ArrayBlockingQueue<E> {
2
    private final Object[] items;
3
    private int count;  // 현재 요소 수
4

5
    private final Lock lock = new ReentrantLock();
6
    private final Condition notEmpty = lock.newCondition();
7
    private final Condition notFull = lock.newCondition();
8

9
    public void put(E e) throws InterruptedException {
10
        lock.lock();
11
        try {
12
            while (count == items.length) {
13
                notFull.await();  // 꽉 참: 대기
14
            }
15
            enqueue(e);
16
            notEmpty.signal();  // Consumer 깨우기
17
        } finally {
18
            lock.unlock();
19
        }
20
    }
21

22
    public E take() throws InterruptedException {
23
        lock.lock();
24
        try {
25
            while (count == 0) {
26
                notEmpty.await();  // 비어있음: 대기
27
            }
28
            E item = dequeue();
29
            notFull.signal();  // Producer 깨우기
30
            return item;
31
        } finally {
32
            lock.unlock();
33
        }
34
    }
35
}

1
BlockingQueue<Task> queue = new ArrayBlockingQueue<>(100);  // 최대 100개
2

3
// 용량 초과 시 대기
4
queue.put(task);  // 블록
5
queue.offer(task, 1, TimeUnit.SECONDS);  // 1초 타임아웃

1
// API 요청 처리 제한
2
public class RateLimitedApiHandler {
3
    private BlockingQueue<ApiRequest> queue = new ArrayBlockingQueue<>(1000);
4

5
    public boolean handleRequest(ApiRequest request) {
6
        // 큐가 꽉 차면 거부 (서버 과부하 방지)
7
        return queue.offer(request);
8
    }
9
}

1
BlockingQueue<Task> unbounded = new LinkedBlockingQueue<>();  // 무제한
2
BlockingQueue<Task> bounded = new LinkedBlockingQueue<>(100);  // 100개 제한

1
// 두 개의 락으로 성능 향상
2
class LinkedBlockingQueue<E> {
3
    private final ReentrantLock takeLock = new ReentrantLock();
4
    private final ReentrantLock putLock = new ReentrantLock();
5

6
    // put과 take가 동시에 실행 가능!
7
}

1
// 우선순위 작업 큐
2
BlockingQueue<Task> queue = new PriorityBlockingQueue<>(10,
3
    (t1, t2) -> Integer.compare(t2.priority, t1.priority)  // 높은 우선순위 먼저
4
);
5

6
queue.put(new Task("Low", 1));
7
queue.put(new Task("High", 10));
8
queue.put(new Task("Medium", 5));
9

10
queue.take();  // "High" (우선순위 10)
11
queue.take();  // "Medium" (우선순위 5)
12
queue.take();  // "Low" (우선순위 1)

1
BlockingQueue<Task> queue = new SynchronousQueue<>();
2

3
// Producer
4
queue.put(task);  // Consumer가 take() 호출할 때까지 블록
5

6
// Consumer
7
Task task = queue.take();  // Producer가 put() 호출할 때까지 블록

1
// Executors.newCachedThreadPool() 내부 구조
2
ThreadPoolExecutor executor = new ThreadPoolExecutor(
3
    0, Integer.MAX_VALUE,
4
    60L, TimeUnit.SECONDS,
5
    new SynchronousQueue<>()  // 작업을 바로 스레드에 전달
6
);

1
import java.util.concurrent.*;
2

3
public class AsyncLogger {
4
    private BlockingQueue<LogMessage> queue = new LinkedBlockingQueue<>(10000);
5
    private volatile boolean running = true;
6

7
    // 로거 시작
8
    public void start() {
9
        Thread writerThread = new Thread(() -> {
10
            while (running || !queue.isEmpty()) {
11
                try {
12
                    LogMessage msg = queue.poll(1, TimeUnit.SECONDS);
13
                    if (msg != null) {
14
                        writeToFile(msg);  // I/O 작업
15
                    }
16
                } catch (InterruptedException e) {
17
                    Thread.currentThread().interrupt();
18
                    break;
19
                }
20
            }
21
        });
22
        writerThread.setDaemon(false);
23
        writerThread.start();
24
    }
25

26
    // 로그 기록 (논블로킹)
27
    public void log(String message) {
28
        LogMessage msg = new LogMessage(message, System.currentTimeMillis());
29

30
        // offer: 큐가 꽉 차면 false 반환 (블록하지 않음)
31
        if (!queue.offer(msg)) {
32
            System.err.println("Log queue full, dropping message: " + message);
33
        }
34
    }
35

36
    // 종료
37
    public void shutdown() {
38
        running = false;
39
    }
40

41
    private void writeToFile(LogMessage msg) {
42
        // 파일에 쓰기 (느린 I/O)
43
    }
44
}

1
import java.util.concurrent.locks.ReentrantLock;
2

3
public class Counter {
4
    private final ReentrantLock lock = new ReentrantLock();
5
    private int count = 0;
6

7
    // 기본 사용
8
    public void increment() {
9
        lock.lock();
10
        try {
11
            count++;
12
        } finally {
13
            lock.unlock();  // 반드시 finally에서 해제
14
        }
15
    }
16

17
    // tryLock: 데드락 방지
18
    public boolean tryIncrement() {
19
        if (lock.tryLock()) {
20
            try {
21
                count++;
22
                return true;
23
            } finally {
24
                lock.unlock();
25
            }
26
        }
27
        return false;  // 락 획득 실패
28
    }
29

30
    // 타임아웃
31
    public boolean incrementWithTimeout() throws InterruptedException {
32
        if (lock.tryLock(1, TimeUnit.SECONDS)) {
33
            try {
34
                count++;
35
                return true;
36
            } finally {
37
                lock.unlock();
38
            }
39
        }
40
        return false;  // 1초 안에 락 획득 못함
41
    }
42

43
    // 인터럽트 가능
44
    public void incrementInterruptibly() throws InterruptedException {
45
        lock.lockInterruptibly();  // 인터럽트 가능
46
        try {
47
            count++;
48
        } finally {
49
            lock.unlock();
50
        }
51
    }
52
}

1
ReentrantLock fairLock = new ReentrantLock(true);  // FIFO 순서 보장
2

3
// 대기 순서대로 락 획득
4
// 단점: 성능 저하 (컨텍스트 스위칭 증가)

1
import java.util.concurrent.locks.ReadWriteLock;
2
import java.util.concurrent.locks.ReentrantReadWriteLock;
3

4
public class CachedData {
5
    private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
6
    private Map<String, String> cache = new HashMap<>();
7

8
    // 읽기: 여러 스레드 동시 실행 가능
9
    public String get(String key) {
10
        rwLock.readLock().lock();
11
        try {
12
            return cache.get(key);
13
        } finally {
14
            rwLock.readLock().unlock();
15
        }
16
    }
17

18
    // 쓰기: 배타적 실행
19
    public void put(String key, String value) {
20
        rwLock.writeLock().lock();
21
        try {
22
            cache.put(key, value);
23
        } finally {
24
            rwLock.writeLock().unlock();
25
        }
26
    }
27
}

1
읽기 락 보유 중:
2
  - 다른 읽기 락 획득 가능
3
  - 쓰기 락 획득 불가능
4

5
쓰기 락 보유 중:
6
  - 다른 읽기 락 획득 불가능
7
  - 다른 쓰기 락 획득 불가능

1
public class Cache<K, V> {
2
    private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
3
    private final Map<K, V> cache = new HashMap<>();
4

5
    public V get(K key) {
6
        rwLock.readLock().lock();
7
        try {
8
            V value = cache.get(key);
9
            if (value != null) {
10
                return value;
11
            }
12
        } finally {
13
            rwLock.readLock().unlock();
14
        }
15

16
        // 캐시 미스: 쓰기 락으로 전환
17
        rwLock.writeLock().lock();
18
        try {
19
            // Double-check (다른 스레드가 이미 로드했을 수 있음)
20
            V value = cache.get(key);
21
            if (value == null) {
22
                value = loadFromDatabase(key);
23
                cache.put(key, value);
24
            }
25
            return value;
26
        } finally {
27
            rwLock.writeLock().unlock();
28
        }
29
    }
30

31
    public void invalidate(K key) {
32
        rwLock.writeLock().lock();
33
        try {
34
            cache.remove(key);
35
        } finally {
36
            rwLock.writeLock().unlock();
37
        }
38
    }
39
}

1
import java.util.concurrent.locks.StampedLock;
2

3
public class Point {
4
    private final StampedLock sl = new StampedLock();
5
    private double x, y;
6

7
    // 낙관적 읽기: 가장 빠름
8
    public double distanceFromOrigin() {
9
        long stamp = sl.tryOptimisticRead();  // 락 없이 stamp 획득
10
        double currentX = x;
11
        double currentY = y;
12

13
        if (!sl.validate(stamp)) {
14
            // 쓰기가 발생했음: 읽기 락으로 재시도
15
            stamp = sl.readLock();
16
            try {
17
                currentX = x;
18
                currentY = y;
19
            } finally {
20
                sl.unlockRead(stamp);
21
            }
22
        }
23

24
        return Math.sqrt(currentX * currentX + currentY * currentY);
25
    }
26

27
    // 쓰기
28
    public void move(double deltaX, double deltaY) {
29
        long stamp = sl.writeLock();
30
        try {
31
            x += deltaX;
32
            y += deltaY;
33
        } finally {
34
            sl.unlockWrite(stamp);
35
        }
36
    }
37
}

1
import java.util.concurrent.atomic.AtomicInteger;
2

3
public class AtomicCounter {
4
    private AtomicInteger count = new AtomicInteger(0);
5

6
    public void increment() {
7
        count.incrementAndGet();  // 원자적 증가
8
    }
9

10
    public int get() {
11
        return count.get();
12
    }
13

14
    // CAS 사용 예제
15
    public boolean compareAndSetTo100() {
16
        int current = count.get();
17
        return count.compareAndSet(current, 100);
18
    }
19
}

1
// 간소화된 코드 — Java 8+에서는 Unsafe.getAndAddInt()를 사용하며,
2
// Java 9+에서는 VarHandle 기반으로 변경되었다
3
public class AtomicInteger {
4
    private volatile int value;
5

6
    public final int incrementAndGet() {
7
        int current, next;
8
        do {
9
            current = value;
10
            next = current + 1;
11
        } while (!compareAndSet(current, next));  // CAS 루프
12
        return next;
13
    }
14

15
    // native 메서드 (하드웨어 CAS 사용)
16
    public final native boolean compareAndSet(int expect, int update);
17
}

1
// 1. synchronized: 느림 (락 오버헤드)
2
private int count = 0;
3
public synchronized void increment() {
4
    count++;
5
}
6

7
// 2. AtomicInteger: 빠름 (락 프리)
8
private AtomicInteger count = new AtomicInteger(0);
9
public void increment() {
10
    count.incrementAndGet();
11
}
12

13
// 3. LongAdder: 매우 빠름 (경쟁이 심할 때)
14
private LongAdder count = new LongAdder();
15
public void increment() {
16
    count.increment();
17
}

1
import java.util.concurrent.atomic.LongAdder;
2

3
public class HighContentionCounter {
4
    private LongAdder count = new LongAdder();
5

6
    // 여러 스레드가 동시에 호출해도 매우 빠름
7
    public void increment() {
8
        count.increment();
9
    }
10

11
    // 최종 합계
12
    public long sum() {
13
        return count.sum();  // 느림 (모든 셀 합산)
14
    }
15
}

1
import java.util.concurrent.atomic.AtomicReference;
2

3
public class ImmutableCache {
4
    private AtomicReference<ImmutableMap<String, String>> cache
5
        = new AtomicReference<>(ImmutableMap.of());
6

7
    public void put(String key, String value) {
8
        ImmutableMap<String, String> oldCache, newCache;
9
        do {
10
            oldCache = cache.get();
11
            newCache = ImmutableMap.<String, String>builder()
12
                .putAll(oldCache)
13
                .put(key, value)
14
                .build();
15
        } while (!cache.compareAndSet(oldCache, newCache));
16
    }
17

18
    public String get(String key) {
19
        return cache.get().get(key);
20
    }
21
}

1
import java.util.concurrent.CountDownLatch;
2

3
public class ParallelTaskRunner {
4
    public void runTasks() throws InterruptedException {
5
        int numTasks = 5;
6
        CountDownLatch latch = new CountDownLatch(numTasks);
7

8
        // 5개 작업 시작
9
        for (int i = 0; i < numTasks; i++) {
10
            new Thread(() -> {
11
                doWork();
12
                latch.countDown();  // 카운트 감소
13
            }).start();
14
        }
15

16
        // 모든 작업이 완료될 때까지 대기
17
        latch.await();
18
        System.out.println("All tasks completed!");
19
    }
20
}
21

22
// 실무 예제: 애플리케이션 시작 대기
23
public class Application {
24
    private CountDownLatch startupLatch = new CountDownLatch(3);
25

26
    public void start() throws InterruptedException {
27
        // 3개 서비스 시작
28
        new Thread(() -> {
29
            initDatabase();
30
            startupLatch.countDown();
31
        }).start();
32

33
        new Thread(() -> {
34
            initCache();
35
            startupLatch.countDown();
36
        }).start();
37

38
        new Thread(() -> {
39
            initMessageQueue();
40
            startupLatch.countDown();
41
        }).start();
42

43
        // 모든 서비스가 준비될 때까지 대기
44
        startupLatch.await();
45
        System.out.println("Application ready!");
46
    }
47
}

1
import java.util.concurrent.CyclicBarrier;
2

3
public class ParallelMergeSort {
4
    public void sort(int[] array) throws Exception {
5
        int numThreads = 4;
6
        CyclicBarrier barrier = new CyclicBarrier(numThreads, () -> {
7
            // 모든 스레드가 barrier에 도달하면 실행
8
            System.out.println("All threads reached barrier!");
9
        });
10

11
        for (int i = 0; i < numThreads; i++) {
12
            int start = i * (array.length / numThreads);
13
            int end = (i + 1) * (array.length / numThreads);
14

15
            new Thread(() -> {
16
                sortPartition(array, start, end);
17
                try {
18
                    barrier.await();  // 다른 스레드 대기
19
                } catch (Exception e) {
20
                    e.printStackTrace();
21
                }
22
            }).start();
23
        }
24
    }
25
}

1
import java.util.concurrent.Semaphore;
2

3
// 커넥션 풀
4
public class ConnectionPool {
5
    private Semaphore semaphore;
6
    private List<Connection> connections;
7

8
    public ConnectionPool(int size) {
9
        this.semaphore = new Semaphore(size);
10
        this.connections = new ArrayList<>(size);
11
        for (int i = 0; i < size; i++) {
12
            connections.add(createConnection());
13
        }
14
    }
15

16
    public Connection acquire() throws InterruptedException {
17
        semaphore.acquire();  // 허가 획득 (없으면 대기)
18
        return getConnection();
19
    }
20

21
    public void release(Connection conn) {
22
        returnConnection(conn);
23
        semaphore.release();  // 허가 반환
24
    }
25
}
26

27
// API Rate Limiter
28
public class RateLimiter {
29
    private Semaphore semaphore;
30

31
    public RateLimiter(int requestsPerSecond) {
32
        this.semaphore = new Semaphore(requestsPerSecond);
33

34
        // 1초마다 허가 리필
35
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
36
        scheduler.scheduleAtFixedRate(() -> {
37
            semaphore.release(requestsPerSecond - semaphore.availablePermits());
38
        }, 1, 1, TimeUnit.SECONDS);
39
    }
40

41
    public boolean tryAcquire() {
42
        return semaphore.tryAcquire();
43
    }
44
}