@Entity
public class Coupon {
@Id @GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
private Long userId;
public Coupon() {
}
public Coupon(Long userId) {
this.userId = userId;
}
public Long getId() {
return id;
}
}
Coupon Repository
public interface CouponRepository extends JpaRepository<Coupon, Long> {
}
Coupon Service
@Service
public class ApplyService {
private final CouponRepository couponRepository;
public ApplyService(CouponRepository couponRepository) {
this.couponRepository = couponRepository;
}
public void apply(Long userId) {
// 쿠폰 개수 세기
long count = couponRepository.count();
// 쿠폰 100개 초과이면 발급 안함
if (count > 100) {
return;
}
// 그게 아니라면 쿠폰 발급
couponRepository.save(new Coupon(userId));
}
}
쿠폰 1개 발급 테스트
@SpringBootTest
class ApplyServiceTest {
@Autowired
private ApplyService applyService;
@Autowired
private CouponRepository couponRepository;
@Test
public void 한번만응모() {
applyService.apply(1L);
long count = couponRepository.count();
assertThat(count).isEqualTo(1);
}
}
만약 동시에 쿠폰 발급 요청이 들어온다면 어떤 문제가 생길까?
쿠폰이 99개 발급 됐을 때, 쿠폰 발급 요청이 동시에 들어오면 101개가 발급될 수도 있다.(레이스 컨디션) - 아래에 설명
쿠폰 1000개 발급 테스트(멀티스레드활용)
💡 ExecutorsService 병렬작업을 간단하게 도와주는 Java API
💡 CountDownLatch 다른 Thread에서 수행하는 작업을 기다리도록 도와주는 클래스
Test code
@Test
public void 여러명응모() {
int threadCount = 1000;
ExecutorService executorService = Executors.newFixedThreadPool(32); // 2개의 스레드로 구성된 고정 스레드 풀을 생성
// for문을 통해 1000개 요청
for (int i = 0; i < threadCount; i++) {
long userId = i;
executorService.submit(() -> {
applyService.apply(userId);
});
}
latch.await();
// 쿠폰 개수 확인
long count = couponRepository.count();
// 쿠폰 개수 100개인지 확인
assertThat(count).isEqualTo(100);
}
(문제점-1)
결과는 100개보다 더 많은 쿠폰을 발급하게 된다.
왜냐하면 레이스 컨디션이 발생했기 때문이다.
💡 레이스 컨디션(race condition) 두 개 이상의 쓰레드가 공유 데이터에 access를 하고 동시에 작업을 하려고 할 때 발생하는 문제이다. 경주(race)하는 상태(condition)라고 생각하면 될 듯하다.
쿠폰이 99개 발급 됐을 때, 쿠폰 발급 요청이 동시에 들어오면 101개가 발급될 수도 있다.
싱글 쓰레드로 작업한다면 레이스 컨디션은 일어나지 않겠지만, 쿠폰 발급 로직 전체를 싱글 쓰레드로 작업하게 된다면 성능이 좋지 않을 것이다. (먼저 요청한 사람의 쿠폰이 발급된 이후에 다른 사람들의 쿠폰이 발급 가능해지기 때문)
자바에서 지원하는 synchronized를 해결방법으로 생각할 수 있겠지만, 레이스 컨디션이 다시 발생함으로 적절하지 않다.
💡 Redis를 사용하는 이유? Redis는 싱글스레드를 기반으로 동작하여 레이스 컨디션을 해결할 수 있을 뿐만 아니라 incr은 성능도 빠른 명령어이다. redis incr명령어를 통해 key:value+1
해결방법
쿠폰 발급 전에 쿠폰 카운트를 1증가시키고 리턴되는 값이 100보다 크다면 이미 100개 이상 발급되었다는 뜻임으로 쿠폰 발급을 더 이상 안되게 한다.
CouponCountRepository
레디스 명령어를 실행할 repository 생성(class임. interface아님)
@Repository
public class CouponCountRepository {
// redis 명령어 실행을 위한 redisTemplate 변수 추가
private final RedisTemplate<String, String> redisTemplate;
public CouponCountRepository(RedisTemplate<String, String> redisTemplate) {
this.redisTemplate = redisTemplate;
}
// redis의 incr명령어를 사용학 위한 increment method 작성
public Long increment() {
return redisTemplate
.opsForValue()
.increment("coupon_count");
}
}
순서
시간
Thread-1
Redis - Coupon Count
Thread-2
1
10:00
start - 10:00
incr counpon_count
end - 10:02
99개
2
10:01
99개
wait...
3
10:02
100개
start - 10:02
incr counpon_count
end - 10:03
4
10:03
create coupon
101개
5
101개
failed create coupon
(문제발생-2)
발급하는 쿠폰의 개수가 많아질수록 (문제1)RDB에 부하를 주게 된다. 이는 서비스 지연 혹은 오류로 이어진다.
만약 사용하는 RDB가 쿠폰 전용 DB가 아니라 다양한 곳에서 사용하고 있었다면 (문제2)다른 서비스까지 장애가 발생할 수 있다.
C:\Users\lsi66>docker ps -a
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
3cf1f8ac51a3 wurstmeister/zookeeper "/bin/sh -c '/usr/sb…" 7 hours ago Up 36 seconds 22/tcp, 2888/tcp, 3888/tcp, 0.0.0.0:2181->2181/tcp zookeeper
1ea80a0a6060 wurstmeister/kafka:2.12-2.5.0 "start-kafka.sh" 7 hours ago Up 2 seconds 0.0.0.0:9092->9092/tcp kafka
bbce7b8bd410 redis "docker-entrypoint.s…" 8 hours ago Up About a minute 0.0.0.0:6379->6379/tcp myredis
d62f1870038f mysql:latest "docker-entrypoint.s…" 44 hours ago Up 2 minutes 0.0.0.0:3306->3306/tcp, 33060/tcp my-mysq
스프링에서는 손쉽게 설정값들을 설정할 수 있도록 ConsumerFactory라는 인터페이스를 제공한다.
설정값들을 담기 위한 맵을 변수로 추가해준다.
@Configuration
public class KafkaConsumerConfig {
// # 순서1 : ConsumerFactory 설정값 설정
Map<String, Object> config = new HashMap<>();
config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
config.put(ConsumerConfig.GROP_ID_CONFIG, "group_1");
config.put(ConsumerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
config.put(ConsumerConfig.VALUE_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
return new DefaultKafkaConsmerFactory<>(config);
}
// # 순서 2 : 토픽으로부터 메시지를 전달받기 위한 kafka-listener를 만드는 kafka-listener-container-factory 생성
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Long> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, Long> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
}
(패키지)consumer -CouponCreatedConsumer
@Component
public class CouponCreatedConsumer {
// # 순서 3 : 토픽에 전송된 데이터를 가져오기 위한 컨슈머 작업
private final CouponRepository couponRepository;
public CouponCreatedConsumer(CouponRepository couponRepository) {
this.couponRepository = couponRepository;
}
@KafkaListener(topics = "coupon_create", groupId = "group_1")
public void listener(Long userId) {
// 순서 4 : 쿠폰 생성
couponRepository.save(new Coupon(userId));
}
}
(패키지) domain -Coupon
@Entity
public class Coupon {
@Id @GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
private Long userId;
public Coupon() {
}
public Coupon(Long userId) {
this.userId = userId;
}
public Long getId() {
return id;
}
}
(패키지) repository -CouponRepository
public interface CouponRepository extends JpaRepository<Coupon, Long> {
}
테스트 코드 실행
@Test
public void 여러명응모() {
int threadCount = 1000;
ExecutorService executorService = Executors.newFixedThreadPool(32); // 2개의 스레드로 구성된 고정 스레드 풀을 생성
// for문을 통해 1000개 요청
for (int i = 0; i < threadCount; i++) {
long userId = i;
executorService.submit(() -> {
applyService.apply(userId);
});
}
latch.await();
// 쿠폰 개수 확인
long count = couponRepository.count();
// 쿠폰 개수 100개인지 확인
assertThat(count).isEqualTo(100);
}
fail : 100개 아닌 11개만 생성이 되었음..
왜?
순서
Time
Test Case
Producer
Consumer
1
10:00
테스트케이스 시작
데이터 수신중...
2
10:01
데이터 전송 완료
데이터 처리...
3
10:02
테스트케이스 종료
데이터 처리...
4
10:03
데이터 처리...
5
10:04
데이터 처리완료
테스트 케이스는 데이터가 전송이 완료된 시점을 기준으로 쿠폰의 개수를 가져오고
컨슈머에서는 그 시점에 아직 모든 쿠폰을 생성하지 않았기 때문에 테스트 케이스가 실패한다.
쿠폰이 100개가 정상적으로 생성되는지 확인하기 위해 Tread slip을 활용해보자.
@Test
public void 여러명응모() throws InterruptedException {
int threadCount = 100;
ExecutorService executorService = Executors.newFixedThreadPool(32);
CountDownLatch latch = new CountDownLatch(threadCount);
// for문을 통해 1000개 요청
for (int i = 0; i < threadCount; i++) {
long userId = i;
executorService.submit(() -> {
applyService.apply(userId);
});
}
latch.await();
Thread.sleep(10000); // 👀 여기 추가
// 쿠폰 개수 확인
long count = couponRepository.count();
// 쿠폰 개수 100개인지 확인
assertThat(count).isEqualTo(100);
}
테스트 성공
결론
카프카를 사용한다면 API에서 직접 쿠폰을 생성할 때에 비해서 처리량을 조절할 수있게 된다.
처리량을 조절함에 따라서 데이터베이스의 부하를 줄일 수 있다는 장점이 있다.
하지만 테스트 케이스처럼 쿠폰 생성까지 약간의 텀이 발생한다는 단점이 있다.
💡 처리량 조절? Kafka 는 토픽에 있는 데이터를 순차적으로 가져와서 처리하게 됩니다.Consumer 가 1개가 있고 토픽에 데이터가 100개가 있다고 가정할때 Consumer 에서는 1번 데이터를 가져와서 처리가완료되면 2번 데이터를 가져와서 처리합니다. 10:00시에 100명의 유저가 1번씩 요청을 보내게 됐을때 API 에서 직접 처리를 한다면 데이터베이스에 100번의 요청이 한번에 몰리게 될 것입니다. 하지만 이 요청을 카프카 프로듀서를 이용하여 토픽에 전송하고 Consumer 를 이용하여 데이터를 처리한다면 데이터베이스에 요청하는양을 조절할 수 있게됩니다. 이것을 처리량조절이 가능하다고 표현하였습니다.
발급 가능한 쿠폰 개수를 1인당 1개로 제한하기(요구사항 변경)
방법1. 데이터 베이스에서 userId와 coupon type를 uniqueKey를 활용하여 1개만 생성되도록 데이터베이스 레벨에서 막는다. 하지만 보통 한 userId가 여러 coupon type을 갖기 때문에 탈락.
방법2. 범위로 락을 잡고 처음에 쿠폰 발급 여부를 가져와서 판단.
쿠폰을 아직 발급하지 않았다면 쿠폰 발급
위에 예시는 API에서는 쿠폰 발급 가능 여부만 판단하고, 실제 쿠폰 생성은 consumer에서 생성
실제 consumer가 쿠폰 생성하는 데에는 텀이 발생하기 때문에, 그 텀에 또 쿠폰 요청하면 쿠폰이 발급되지 않았다고 판단하여, 한명에서 쿠폰이 2개 발급되는 경우가 발생할 수 있기 때문에 탈락.
public void apply(Long userId) {
// lock start
// 쿠폰 발급 여부
// if 발급됐다면) return
Long count = couponCountRepository.increment();
// 쿠폰 100개 초과이면 발급 안함
if (count > 100) {
return;
}
couponCreateProducer.create(userId);
// lock end
}
API에서 쿠폰을 발급한다고 해도 lock범위가 너무 넓어진다.
다른 요청들은 이 lock이 끝날 때까지 이 로직에 접근하지 못하기 때문에, 성능이 안좋아질 수 있다.
방법3. set 자료구조 활용(✔)
💡 Set set은 값을 유니크하게 저장할 수 있는 자료구조 값은 2번 저장해도 1개만 남기 때문에 중복 여부를 빠르게 확인할 수 있다. Redis에서도 Set을 지원한다.
sadd test 1
>> 1
// 정상적으로 추가 되었다면 1개 추가 되었기 때문에 1 리턴
// ✔ 추가 되었다면 추가된 vaule의 개수를 리턴
// 똑같은 key를 넣었을 경우, 0 리턴
set에 존재하지 않으면 발급, 존재하면 발급하지 않는다.
(패키지)repository - AppliedUserRepository
public class AppliedUserRepository {
// #1 SET을 관리할 레포지토리 생성
// #2 redis 명령을 수행해야 하기 때문에 redisTemplate생성
private final RedisTemplate<String, String> redisTemplate;
public AppliedUserRepository(RedisTemplate<String, String> redisTemplate) {
this.redisTemplate = redisTemplate;
}
// #3 set에 데이터를 넣기 위한 메소드를 만들어 준다.
// ✔ 이 부분이 redis의 명령어 수행부분이다. sadd applied_user userId.toString()와 동일.
public Long add(Long userId) {
return redisTemplate
.opsForSet()
.add("applied_user", userId.toString());
}
}
(패키지)service - ApplyService
public void apply(Long userId) {
Long apply = appliedUserRepository.add(userId); // 👀여기에 추가
// 경우1. 쿠폰 발급이 1개 아니라면(쿠폰발급X)
if (apply != 1) {
return;
}
// 경우2. 쿠폰 발급
Long count = couponCountRepository.increment();
// 쿠폰 100개 초과이면 발급 안함
if (count > 100) {
return;
}
// 그게 아니라면 쿠폰 발급
couponCreateProducer.create(userId);
}
(패키지)test - ApplyServiceTest
@Test
public void 한명당_한개의쿠폰만_발급() throws InterruptedException {
int threadCount = 100;
ExecutorService executorService = Executors.newFixedThreadPool(32);
CountDownLatch latch = new CountDownLatch(threadCount);
// for문을 통해 1000개 요청
for (int i = 0; i < threadCount; i++) {
long userId = i;
executorService.submit(() -> {
applyService.apply(1L); // user 1이 100번의 요청을 하게 되지만 결과적으로 1개의 쿠폰만 발급을 해야한다.
});
}
latch.await();
Thread.sleep(10000);
// 쿠폰 개수 확인
long count = couponRepository.count();
// 쿠폰 개수 100개인지 확인
assertThat(count).isEqualTo(1); // 1개의 쿠폰만 발급
}
(컨슈머에서) 쿠폰 발급 중 에러가 발생한다면?
컨슈퍼에서 토픽에 있는 데이터를 가져간 후에 쿠폰을 발급하는 중 에러가 발생하면 쿠폰은 발급되지 않았는데 쿠폰 개수만 올라가는 문제가 발생할 수 있다.(100개보다 적은 쿠폰 개수 발급)
domain - FailedEnvet
@Entity
public class FailedEvent {
// 쿠폰 발급에 실패한 데이터를 담기 위한 Entity
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
private Long userId;
public FailedEvent() {
}
public FailedEvent(Long userId) {
this.userId = userId;
}
}
repository - FailedEventRepository
public interface FailedEventRepository extends JpaRepository<FailedEvent, Long> {
}
consumer - CouponCreatedConsumer
@Component
public class CouponCreatedConsumer {
// # 순서 3 : 토픽에 전송된 데이터를 가져오기 위한 컨슈머 작업
private final CouponRepository couponRepository;
private final FailedEventRepository failedEventRepository;
private final Logger logger = LoggerFactory.getLogger(CouponCreatedConsumer.class);
public CouponCreatedConsumer(CouponRepository couponRepository, FailedEventRepository failedEventRepository) {
this.couponRepository = couponRepository;
this.failedEventRepository = failedEventRepository;
}
@KafkaListener(topics = "coupon_create", groupId = "group_1")
public void listener(Long userId) {
// 순서 4 : 쿠폰 생성
try {
couponRepository.save(new Coupon(userId));
} catch (Exception e) {
// 쿠폰 발급을 하다가 error가 발생하면 logger를 사용해서 log를 남겨준다.
logger.error("failed to create coupon::" + userId);
// failedEventRepository에 쿠폰발급 실패한 userId를 저장한다.
failedEventRepository.save(new FailedEvent(userId));
}
}
}
쿠폰 발급을 진행하다가 에러가 발생하면 Failed Event에 실패한 이벤트를 저장한다.
이후에 배치 프로그램에서 Failed Event에 쌓인 데이터들을 주기적으로 읽어서 쿠폰을 발급한다면, 결과적으로 쿠폰 100개가 모두 발급될 것이다.
