spring / / 2023. 8. 18. 09:45

Spring 분산환경에서 동시성 처리 방안

여러 개의 쓰레드 경합이 발생하는 상황에서 동시성 처리하기 위해서 일반적으로 synchronized와 같이 동기화를 사용하면 되지만 여러 개의 프로세스 환경에서는 처리가 되지 않는다.
왜냐하면 synchronized는 하나의 프로세스에서만 처리가 되고 다른 프로세스에서는 동작하지 않기 때문에 이를 위해 처리할 수 있는 방안을 알아보자.

이를 해결할 수 있는 방법은 몇 가지가 있는데 아래와 같은 방법에 대한 테스트를 진행해 보았다.

  1. 기본 실행
  2. 비관적 락(Pessimistic Lock) 사용
  3. Hazelcast Distrubuted Lock 이용
  4. Spring Integration Lock 이용
  5. 어노테이션 이용 (Spring Integration)

기본 시나리오

동시성 문제가 발생할 수 있는 하나의 시나리오를 만들어 보자.

한 명의 사용자는 회원가입 시 하나의 쿠폰을 발행을 받고 이를 사용한다고 해보자.

쿠폰은 한 개만 발행되므로 한 번만 사용할 수 있고 동시에 두 개를 사용했을 경우 사용하지 못하게 막아줘야 한다. 하지만 두 개의 프로세스에서 동시에 사용을 하는 경우에 발생하는 문제를 어떻게 해결할 수 있는 지 알아보자.

  1. 사용자 등록 (쿠폰 등록)
  2. 쿠폰 사용 API 생성
  3. 동시에 2개의 프로세스에서 쿠폰 사용 API 호출

환경 구성

  • Spring Boot 2.7.4
  • Java 8
  • JPA/H2
  • Hazelcast (Server-client)

기본 소스

pom.xml

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-tx</artifactId>
    </dependency>
    <dependency>
        <groupId>com.h2database</groupId>
        <artifactId>h2</artifactId>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
</dependencies>

사용자 Controller

@RestController
@RequestMapping("users")
@RequiredArgsConstructor
public class UserController {

    private final UserService userService;

    private final UserCouponService userCouponService;

    @GetMapping("{userId}")
    public User find(@PathVariable int userId) {
        return userService.find(userId);
    }

    @PostMapping
    public void register(@RequestBody UserCdo userCdo) {
        User user = new User(userCdo.getUserId(), userCdo.getName());
        userService.save(user);
    }

    @PutMapping("{userId}/coupons/issue")
    public void useCoupon(@PathVariable int userId) {
        userCouponService.issueCoupon(userId);
    }
}

사용자 Service

@Service
@RequiredArgsConstructor
@Transactional
public class UserService {

    private final UserRepository userRepository;

    public User find(int userId) {
        return userRepository.findById(userId)
            .map(UserJpo::toDomain).orElseThrow(() -> new RuntimeException("User not found : " + userId));
    }

    public void save(User user) {
        userRepository.save(new UserJpo(user));
    }


    public void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

사용자 쿠폰 Service

@Service
@RequiredArgsConstructor
@Slf4j
public class UserCouponService {

    private final UserService userService;

    @Transactional
    public void issueCoupon(int userId) {
        User user = userService.find(userId);

        userService.sleep(3000); // 동시성을 유발하기 위해 3초간 대기시킨다.

        if (user.getCouponCount() <= 0) {
            throw new IllegalArgumentException(user.getName() + "'s coupon is not available. coupon count is " + user.getCouponCount());
        }

        user.useCoupon();
        userService.save(user);

        log.debug(user.getName() + "'s coupon issued.");
    }
}

사용자 Repository

@Repository
public interface UserRepository extends JpaRepository<UserJpo, Integer> {
}

사용자 Jpa entity

@Entity
@NoArgsConstructor
@Getter
public class UserJpo {

    @Id
    private int userId;

    private String name;

    private int couponCount;

    public UserJpo(User user) {
        this.userId = user.getUserId();
        this.name = user.getName();
        this.couponCount = user.getCouponCount();
    }

    public User toDomain() {
        return new User(this.userId, this.name, this.couponCount);
    }

}

사용자 Domain

@NoArgsConstructor
@AllArgsConstructor
@Getter
public class User {

    private int userId;

    private String name;

    private int couponCount;

    public User(int userId, String name) {
        this.userId = userId;
        this.name = name;
        this.couponCount = 1;
    }

    public void useCoupon() {
        this.couponCount--;
    }
}

Spring Boot Application

@SpringBootApplication
public class UserApplication {

    public static void main(String[] args) {
        SpringApplication.run(UserApplication.class, args);
    }
}

application.yml

server:
  port: 0 # 2개의 인스턴스를 임의의 포트번호로 띄우기 위함

spring:
  application:
    name: user-app
  jpa:
    database-platform: org.hibernate.dialect.H2Dialect
    hibernate:
      ddl-auto: create
  datasource:
    url: jdbc:h2:~/${spring.application.name}-app;AUTO_SERVER=TRUE
    username: sa
    password:
    driverClassName: org.h2.Driver

테스트 실행

1. 기본 실행

위의 소스로 기본 실행을 해보자.

UserApplication을 2개 실행하자.

Intellij에서 실행할 때 기본적으로 1개만 실행이 되므로 Allow multiple instances를 선택해줘야 한다.

실행을 intellij http를 사용하여 실행한다.

  • 사용자 등록을 하고 쿠폰 발행을 동시에 실행
  • sleep이 3초로 걸려있으므로 두 번째 쿠폰발행 API를 3초 안에 실행하면 된다.
실행
### 사용자 등록
POST http://localhost:54349/users
content-Type: application/json

{
  "userId": 1,
  "name": "홍길동"
}
### 쿠폰 발행 (1번 서버)
PUT http://localhost:54349/users/1/coupons/issue
### 쿠폰 발행 (2번 서버)
PUT http://localhost:54121/users/1/coupons/issue
결과
# 1번 프로세스의 응답
홍길동's coupon issued.

# 2번 프로세스의 응답
홍길동's coupon issued.

1개만 발행되고 나머지는 실패할 것으로 기대했지만, 실행 결과 2개의 프로세스에서 모두 쿠폰이 발행됨을 확인할 수 있다.
홍길동에게 발행된 쿠폰은 1개인데 2개가 발행되므로 문제가 발생한 것이다.

이러한 상황에서 동시성 이슈를 어떻게 처리할 수 있는 지 알아보자.

2. 비관적 락(Pessimistic lock)

가장 쉽게 할 수 있는 방법은 해당 DB에 비관적 락(Pessmistic Lock)을 사용하는 것이다. 즉, User 테이블의 홍길동에 대해 row-level lock을 적용하는 것이다. 2개의 프로세스가 접근할 때 먼저 접근한 쪽에서 row-level lock을 잡고 완료되기 전까지 나머지 thread를 대기시키는 방식이다.

이렇게 함으로써 2번째 들어온 thread는 1번째가 완료되기 전까지 row를 읽기조차 하지 못하고 대기한다. 그리고 1번째 thread가 완료되고 나면 2번째 thread가 실행된다.

소스
@Repository
public interface UserRepository extends JpaRepository<UserJpo, Integer> {

    @Lock(LockModeType.PESSIMISTIC_WRITE)
    Optional<UserJpo> findByUserId(int userId);
}

UserRepository에 @Lock(LockModeType.PESSIMISTIC_WRITE)을 추가하여 비관적 락을 사용할 수 있게 한다.

public void issueCoupon(int userId) {
        User user = userService.findByUserId(userId); //  findById가 아닌 findByUserId를 사용 (비관적 락 적용)
        ...
}
실행
### 사용자 등록
POST http://localhost:54349/users
content-Type: application/json

{
  "userId": 1,
  "name": "홍길동"
}
### 쿠폰 발행 (1번 서버)
PUT http://localhost:54349/users/1/coupons/issue
### 쿠폰 발행 (2번 서버)
PUT http://localhost:54121/users/1/coupons/issue
결과
# 1번 프로세스의 응답
홍길동's coupon issued.

# 2번 프로세스의 응답
java.lang.IllegalArgumentException: 홍길동's coupon is not available. coupon count is 0

우리가 예상한 대로 정상적으로 잘 실행되는 것을 알 수 있다.

3. Hazelcast Distributed Lock 이용

두 번째 방법은 hazelcast를 사용하여 lock을 이용하는 것이다.

소스

Hazelcast 의존성 추가

<dependency>
    <groupId>com.hazelcast</groupId>
    <artifactId>hazelcast-all</artifactId>
    <version>4.1</version>
</dependency>

hazelcast는 server-client로 설정을 하였다.

개인적으로 사용하는 Hazelcast 서버가 이미 있어서 여기서는 클라이언트 설정만 추가했다.

hazelcast-client.yaml 파일을 resources 하위에 생성한다.

hazelcast-client:
  cluster-name: my-cluster
  network:
    cluster-members:
      - 172.16.xxx,xxx # hazelcast 서버 주소

서비스 로직

@Service
@RequiredArgsConstructor
@Slf4j
public class UserCouponHazelcastService {

    private final UserService userService;

    private final HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();

    @Transactional
    public void issueCoupon(int userId) {
        Lock lock = hazelcastInstance.getCPSubsystem().getLock("user-coupon");

        try {
            lock.lock();
            log.debug("[" + userId + "] lock");

            User user = userService.find(userId);
            userService.sleep(3000); // 동시성을 유발하기 위해 3초간 대기시킨다.

            if (user.getCouponCount() <= 0) {
                throw new IllegalArgumentException(user.getName() + "'s coupon is not available. coupon count is " + user.getCouponCount());
            }

            user.useCoupon();
            userService.save(user);

            log.debug(user.getName() + "'s coupon issued.");
        } finally {
            try {
                log.debug("[" + userId + "] unlock");
                //hazelcastLocker.unlock("user-coupon");
                lock.unlock();

                Thread.sleep(100); // 추가 작업
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }
}
실행
### 사용자 등록
POST http://localhost:54349/users
content-Type: application/json

{
  "userId": 1,
  "name": "홍길동"
}
### 쿠폰 발행 (1번 서버)
PUT http://localhost:54349/users/1/coupons/issue
### 쿠폰 발행 (2번 서버)
PUT http://localhost:54121/users/1/coupons/issue
결과
# 1번 프로세스의 응답
홍길동's coupon issued.

# 2번 프로세스의 응답
홍길동's coupon issued.

예상한 기대와는 다르게 2개의 쿠폰이 발급되었다.

lock이 제대로 실행되지 않았던 것일까? 하지만 로그를 확인하면 2개의 프로세스에서 lock과 unlock이 순서대로 찍힌 것을 확인할 수 있다. lock은 정상적으로 실행된 것을 알 수 있다.

# 1번 프로세스
2023-08-14 08:52:53.918 DEBUG 34421 --- [o-auto-1-exec-6] .UserCouponHazelcastService : [1] lock
2023-08-14 08:52:56.920 DEBUG 34421 --- [o-auto-1-exec-6] .UserCouponHazelcastService : [1] unlock

# 2번 프로세스
2023-08-14 08:52:56.921 DEBUG 34425 --- [o-auto-1-exec-3] .UserCouponHazelcastService : [1] lock
2023-08-14 08:52:59.926 DEBUG 34425 --- [o-auto-1-exec-3] .UserCouponHazelcastService : [1] unlock

1번 프로세스가 08:52:56.920에 unlock이 되었고 2번 프로세스가 08:52:56.921에 lock이 되었으니 2번째 실행할 때는 더 이상 발급이 되지 않아야 하지만 발급이 되는 문제가 발생한다.

그래서 소스를 확인해보니 트랜잭션에 의해 발생하는 문제가 아닐까 생각되어 트랜잭션 로그를 찍어보기로 했다.

트랜잭션 로그를 찍기 위해 application.yml에 아래 코드를 추가한다.

logging:
  level:
    org.springframework.transaction.interceptor: trace

그리고 다시 실행해보자

# 1번 프로세스
2023-08-14 08:57:42.945 DEBUG 34421 --- [o-auto-1-exec-8] .UserCouponHazelcastService : [1] unlock
2023-08-14 08:57:43.051 TRACE 34421 --- [o-auto-1-exec-8] o.s.t.i.TransactionInterceptor           : Completing transaction for [example.user.service.hazelcast.UserCouponHazelcastService.issueCoupon]

# 2번 프로세스
2023-08-14 08:57:41.166 TRACE 34425 --- [o-auto-1-exec-8] o.s.t.i.TransactionInterceptor           : Getting transaction for [example.user.service.hazelcast.UserCouponHazelcastService.issueCoupon]
2023-08-14 08:57:42.946 DEBUG 34425 --- [o-auto-1-exec-8] .UserCouponHazelcastService : [1] lock

1번 프로세스에서 unlock은 08:57:42.945에 실행되었지만 트랜잭션 commit은 08:57:43.051에 실행되었다.
2번 프로세스는 08:57:42.946에 lock을 얻었지만 이는 1번 트랜잭션이 커밋되기 전 시점(08:57:43.051)인 것이다. 그래서 아직 쿠폰발급이 커밋되지 않은 상태로 2번에서 조회가 되어 문제가 발생한 것이다.

이렇 듯 lock의 설정, 해제는 트랜잭션과 무관하므로 내부에서 동시에 구현하면 문제가 발생할 여지가 많다.

이런 문제를 해결하기 위해 트랜잭션 외부에서 lock을 설정 및 해제하는 코드를 작성해보았다.

UserCouponHazelcastService에서는 트랜잭션 처리를 하고 wrapper 클래스(UserCouponHazelcastWrapperService)에서 lock을 제어하게 코드를 작성해보았다.

@Service
@RequiredArgsConstructor
@Slf4j
public class UserCouponHazelcastWrapperService {

    private final UserCouponHazelcastService userCouponHazelcastService;

    private final HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();

    public void issueCoupon(int userId) {
        Lock lock = hazelcastInstance.getCPSubsystem().getLock("user-coupon");

        try {
            log.debug("[" + userId + "] lock");
            lock.lock();

            userCouponHazelcastService.issueCoupon(userId);
        } finally {
            try {
                log.debug("[" + userId + "] unlock");
                lock.unlock();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }
}
@Service
@RequiredArgsConstructor
@Slf4j
public class UserCouponHazelcastService {

    private final UserService userService;

    @Transactional
    public void issueCoupon(int userId) {
        User user = userService.find(userId);
        userService.sleep(3000); // 동시성을 유발하기 위해 3초간 대기시킨다.

        if (user.getCouponCount() <= 0) {
            throw new IllegalArgumentException(user.getName() + "'s coupon is not available. coupon count is " + user.getCouponCount());
        }

        user.useCoupon();
        userService.save(user);

        log.debug(user.getName() + "'s coupon issued.");
    }
}

두 개의 프로세스에서 실행을 해보면 잘 실행이 된다. (1번만 쿠폰발급 성공, 2번 실패)

로그를 보면

# 1번 프로세스
2023-08-14 09:15:33.407 TRACE 35035 --- [o-auto-1-exec-7] o.s.t.i.TransactionInterceptor           : Completing transaction for [example.user.service.hazelcast.UserCouponHazelcastService.issueCoupon]
2023-08-14 09:15:33.408 DEBUG 35035 --- [o-auto-1-exec-7] .u.s.h.UserCouponHazelcastWrapperService : [1] unlock

# 2번 프로세스
2023-08-14 09:15:31.293 DEBUG 35037 --- [o-auto-1-exec-2] .u.s.h.UserCouponHazelcastWrapperService : [1] lock
2023-08-14 09:15:33.409 TRACE 35037 --- [o-auto-1-exec-2] o.s.t.i.TransactionInterceptor           : Getting transaction for [example.user.service.hazelcast.UserCouponHazelcastService.issueCoupon]
java.lang.IllegalArgumentException: 홍길동's coupon is not available. coupon count is 0
    at example.user.service.hazelcast.UserCouponHazelcastService.issueCoupon(UserCouponHazelcastService.java:23) ~[classes/:na]
    at example.user.service.hazelcast.UserCouponHazelcastService$$FastClassBySpringCGLIB$$1c326dde.invoke(<generated>) ~[classes/:na]

1번 프로세스에서 09:15:33.407에 커밋이 되었고 2번 프로세스에서 09:15:33.409에 트랜잭션이 시작되었으니 이미 1번에서 발급완료된 후에 새로운 트랜잭션이 시작되었으니 정상적으로 동작한 것이다.

대신 이런 방식으로 사용을 하면 모든 트랜잭션이 실행될 때까지는 대기가 걸리기 때문에 성능에 많은 영향을 줄 수 있을 듯 하니 개선이 필요할 것이다.
예를 들면 getLock할 때 userId별 lock을 가져오는 방식을 사용

javahazelcastInstance.getCPSubsystem().getLock("user-coupon-" + userId);

4. Spring Integration Lock 이용

Spring Integration에도 Lock 인터페이스가 존재한다. 이를 통해 여러 프로세스 환경에서 배타적 제어를 위해 사용할 수 있다.

실제 구현체는 jdbc, redis, hazelcast, zookeeper 등을 사용할 수 있는데 아래 예제는 jdbc를 활용하는 방법이다.

소스

pom.xml에 추가

<dependency>
   <groupId>org.springframework.integration</groupId>
   <artifactId>spring-integration-jdbc</artifactId>
</dependency>
<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency>
   <groupId>org.postgresql</groupId>
   <artifactId>postgresql</artifactId>
</dependency>

Config 클래스 추가

@Configuration
public class JdbcLockConfig {
    @Bean
    public DefaultLockRepository DefaultLockRepository(DataSource dataSource) {
        return new DefaultLockRepository(dataSource);
    }

    @Bean
    public JdbcLockRegistry jdbcLockRegistry(LockRepository lockRepository) {
        return new JdbcLockRegistry(lockRepository);
    }
}

테이블 생성

CREATE TABLE INT_LOCK (
    LOCK_KEY char(36) NOT NULL,
    REGION varchar(100) NOT NULL,
    CLIENT_ID char(36) DEFAULT NULL,
    CREATED_DATE timestamp NOT NULL,
    PRIMARY KEY (LOCK_KEY,REGION)
);

Controller

@PutMapping("{userId}/coupons/issue")
    @SneakyThrows
    public void useCouponBySpringIntegrationLock(@PathVariable int userId) {
        Lock lock = lockRegistry.obtain("user-" + userId);

        boolean lockAcquired = lock.tryLock(1, TimeUnit.SECONDS);
        log.debug("lockAcquired : " + lockAcquired);
        if (lockAcquired) {
            try {
                userCouponSpringIntegrationService.issueCoupon(userId);
            } finally {
                lock.unlock();
            }
        } else {
            throw new IllegalArgumentException("userId is locked: " + userId);
        }
    }

Service

@Service
@RequiredArgsConstructor
@Slf4j
public class UserCouponSpringIntegrationService {

    private final UserService userService;

    @Transactional
    @SneakyThrows
    public void issueCoupon(int userId) {
        User user = userService.find(userId);

        userService.sleep(3000); // 동시성을 유발하기 위해 3초간 대기시킨다.

        if (user.getCouponCount() <= 0) {
            throw new IllegalArgumentException(user.getName() + "'s coupon is not available. coupon count is " + user.getCouponCount());
        }

        user.useCoupon();
        userService.save(user);

        log.debug(user.getName() + "'s coupon issued.");
    }
}
실행
### 사용자 등록
POST http://localhost:54349/users
content-Type: application/json

{
  "userId": 1,
  "name": "홍길동"
}
### 쿠폰 발행 (1번 서버)
PUT http://localhost:54349/users/1/coupons/issue
### 쿠폰 발행 (2번 서버)
PUT http://localhost:54121/users/1/coupons/issue
결과
# 1번 프로세스의 응답
홍길동's coupon issued.

# 2번 프로세스의 응답
java.lang.IllegalArgumentException: userId is locked: 1

정상적으로 잘 실행되는 것을 확인할 수 있다.

5. 어노테이션 이용 (Spring Integration)

4번과 같이 구현을 하면 동작은 잘 되지만 비즈니스 로직이 복잡해 지는 문제가 있다. 그래서 이를 AOP를 활용하여 구현하면 로직이 깔끔해지는 장점이 있다.

소스

DistributedLock 어노테이션 추가

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface DistributedLock {

    String key();

    TimeUnit timeUnit() default TimeUnit.SECONDS;

    long waitTime() default 3L;

    long leaseTime() default 3L;
}

Aspect 

@Aspect
@Component
@RequiredArgsConstructor
@Slf4j
@Order(Ordered.HIGHEST_PRECEDENCE) // 
public class DistributedLockAop {
    private static final String LOCK_PREFIX = "lock_";

    private final LockTransaction lockTransaction;

    private final LockRegistry lockRegistry;

    @Around("@annotation(example.aop.DistributedLock)")
    public Object lock(final ProceedingJoinPoint joinPoint) throws Throwable {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        Method method = signature.getMethod();
        DistributedLock distributedLock = method.getAnnotation(DistributedLock.class);

        String key = LOCK_PREFIX + CustomSpringELParser.getDynamicValue(signature.getParameterNames(), joinPoint.getArgs(), distributedLock.key());
        Lock lock = lockRegistry.obtain(key);

        try {
            boolean available = lock.tryLock(distributedLock.waitTime(), distributedLock.timeUnit());
            if (!available) {
                return false;
            }

            return lockTransaction.proceed(joinPoint);  
        } catch (InterruptedException e) {
            throw new InterruptedException();
        } finally {
            try {
                lock.unlock();
                log.debug("DistributedLockAop unlock");
            } catch (IllegalMonitorStateException e) {
                log.info("Already UnLock {} {}", method.getName(), "key", key);
            }
        }
    }
}

비즈니스 트랜잭션과 무관한 Lock의 트랜잭션용 클래스

@Component
public class LockTransaction {

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public Object proceed(final ProceedingJoinPoint joinPoint) throws Throwable {
        return joinPoint.proceed();
    }
}
public class CustomSpringELParser {

    public static Object getDynamicValue(String[] parameterNames, Object[] args, String key) {
        ExpressionParser parser = new SpelExpressionParser();
        StandardEvaluationContext context = new StandardEvaluationContext();

        for (int i = 0; i < parameterNames.length; i++) {
            context.setVariable(parameterNames[i], args[i]);
        }

        return parser.parseExpression(key).getValue(context, Object.class);
    }
}

사용방법

@Transactional
@DistributedLock(key = "#user") // @DistributedLock를 붙이면 된다.
public void issueCoupon(int userId) {
    User user = userService.find(userId);

    userService.sleep(3000); // 동시성을 유발하기 위해 3초간 대기시킨다.

    if (user.getCouponCount() <= 0) {
        throw new IllegalArgumentException(user.getName() + "'s coupon is not available. coupon count is " + user.getCouponCount());
    }

    user.useCoupon();
    userService.save(user);

    log.debug(user.getName() + "'s coupon issued.");
}

참고자료

https://spring.io/blog/2019/06/19/spring-tips-distributed-locks-with-spring-integration

https://github.com/megadotnet/Distributed-Lock-hazelcast

https://www.javacodegeeks.com/2020/12/overview-of-implementing-distributed-locks.html

https://blog.voidmainvoid.net/107

https://github.com/alturkovic/distributed-lock

https://helloworld.kurly.com/blog/distributed-redisson-lock/

반응형
저작자표시
  • 네이버 블로그 공유
  • 네이버 밴드 공유
  • 페이스북 공유
  • 카카오스토리 공유
spring 관련 글
글 더보기

티스토리툴바