Good Morning Brady!

회사 후배가 질문했다. 멀티 스레드 환경에서 List 형태의 데이터를 처리하고 싶은데  List에는 중복된 데이터가 존재할 수 있다고 했다. 문제는 어떤 스레드에서 처리 중인 데이터가 다른 스레드에서 연속해서 처리되면 안 된다고 했다. 약간 고민되긴 했지만 synchronized 키워드가 떠올랐다. 사실 Java를 처음 입문했을 때  동시성이 떨어진다는 이유로 웹 환경에서는 synchronized를 잘 사용하지 않는다고 얘기를 들어서 처음부터 나의 관심 밖의 키워드였다. 하지만 후배가 질문한 내용을 보면 멀티 스레드 환경에서 동시성 제어를 통해서 충분히 해결할 수 있다.

List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 1, 2);

ExecutorService es = Executors.newFixedThreadPool(7);

ExecutionMeasurer.measure(() -> {
	List<CompletableFuture<Void>> futures = list.stream()
			.map(num -> CompletableFuture.runAsync(() -> {
				synchronized (num) {
					ExecutionMeasurer.delay(1000);
				}
			}, es))
			.collect(Collectors.toList());

	futures.forEach(CompletableFuture::join);
});

es.shutdown();

list에는 총 7개의 요소가 있고 각각의 요소를 처리하는데 1,000ms가 걸린다고 가정해 보자. 총 7개의 스레드로 처리를 하면 몇 초가 걸릴까? 중복된 요소 1, 2가 존재하기 때문에 7개의 스레드로 실행하더라도  synchronized에 의해서 대략 2초 정도의 시간이 소요된다. 주어진 자원을 100% 다 사용하지 못했지만 최소한의 lock으로 중복 처리 없이 동시성을 높였다고 할 수 있다. synchronized 키워드는 평소가 자주 사용하지는 않지만 상황에 따라서는 유용하게 사용할 수도 있다.

위 예제는 단일 JVM 구조에서 가볍게 실행될 경우이고 분산 환경에서는 좀 더 다른 고민이 필요할 것이다. 데이터 분포는 어떠한지 분배는 어떻게 처리할지 또는 lock의 단위를 어떻게 할지 고민해봐야 한다.

@Transactional
public void read(int no) {
	Sample sample = jdbcTemplate.queryForObject(
			"select * from sample where no = ? for update ",
			new Object[] { no },
			new BeanPropertyRowMapper<>(Sample.class));

	delay(500);

	jdbcTemplate.update(
			"update sample set count = ? where no = ?",
			sample.getCount() + 1, sample.getNo());
}

위 샘플 코드를 아래와 같이 멀티 스레드로 실행하면 SELECT ~ FOR UPDATE까지는 동시에 접근이 되지만 비관적 동시성으로 인해 결국 각각의 요청이 0.5초 이상 기다렸다가 순차적으로 실행이 되는 것을 확인할 수 있다.

@Test
public void run() throws Exception {
	ExecutorService executorService = Executors.newFixedThreadPool(10);

	List<CompletableFuture<Void>> futures = IntStream.rangeClosed(1, 10)
			.mapToObj(i -> CompletableFuture.runAsync(() ->
					sampleService.read(1), executorService))
			.collect(toList());

	futures.forEach(CompletableFuture::join);
}

@Transactional
public void read(int no) {
	Sample sample = jdbcTemplate.queryForObject(
			"select * from sample where no = ?",
			new Object[] { no },
			new BeanPropertyRowMapper<>(Sample.class));

	delay(500);

	int affectedRow = jdbcTemplate.update(
			"update sample set count = ? where no = ? and count = ?",
			sample.getCount() + 1, sample.getNo(), sample.getCount());

	if (affectedRow == 0) {
		throw new RuntimeException("이미 업데이트가 완료 되었습니다.");
	}
}