✨TIL - 동시성 문제 해결

문제발생

여러 애플리케이션에서 한 테이블로 로그성 데이터를 남겼다.

그러기도 했고, 단말에서 부터 이중~사중으로 호출되는 경우가 있어 운영 중 락이 걸리는 위험한 경우가 발생했다.

서버 로그에는 문제가 없었지만, 해당 시간 대의 SQL session을 확인 했을 때

insert 쿼리가 한개씩 여러번 실행되는 것이 아닌 …

한번데 4개의 동일한 insert 쿼리가 실행되는 세션을 발견하였다.

일단 단말 원인을 찾아 해결하되, 추가로 서버에서도 설정을 해주기로 했다.

임시해결

조회 결과를 로그성 테이블에 남기는 코드라 잠시 주석처리하고, 기록 남기는 것을 잠시 중단하기로 하였다.

Linked Blocking Queue로 해결

우리 서비스는 api 실행 순서가 보장되어야하는 서비스라 Linked Blocking Queue 를 사용해 서버에서의 처리를 추가하기로 하였다. 애플리케이션이 실행되면 모든 요청들을 큐로 받고 하나씩실행하게 하는 방법이다. 그럼 여러 요청이 동시에 들어온다고 하더라도 큐에 넣고 하나씩 꺼내 사용하기 때문에 동시성 문제를 해결할 수 있을 것이라고 판단하였다.

　
프로세스는 메모리 상에서 실행중인 프로그램을 말하며,
스레드는 이 프로세스

　
프로세스는 메모리 상에서 실행중인 프로그램을 말하며,
스레드는 이 프로세스

ttpUploadServerHandler : 모든 요청 큐 등록

public class HttpUploadServerHandler extends SimpleChannelInboundHandler<HttpObject> {
   
    // 블락킹 큐 선언
    private final LinkedBlockingQueue<RequestTask> queue;
    public HttpUploadServerHandler(LinkedBlockingQueue<RequestTask> queue) {
      this.queue = queue;
    }   

		/**
      * 데이터가 수신 되었을 때.
      * 수신된 데이터는 네티의 ByteBuf 객체에 저장되어 있으며 두번째 인자 msg를 통해 접근가능하다.
     */
    @Override
    public void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
		
        try {
          // ✔ 간단히 URL과 파라미터만 출력
          log.info("▶ 큐 등록 - Method: {}, URI: {}, Params: {}", request.getMethod().name(), request.getUri(), httpDataAttributes);
          // 큐에 작업 넣기 - process() 에서 uriController 호출
          queue.put(new RequestTask(ctx, new HashMap<>(httpDataAttributes), this, request));
          log.info("▶ 큐 등록 완료");
        } catch (Exception e) {
          log.error("▷ 큐 등록 실패 - Method: {}, URI: {}, Params: {}", request.getMethod().name(), request.getUri(), httpDataAttributes, e);
        }
						
		}

RequestTask

import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpRequest;
import kr.co.bzsys.logicserver.http.HttpUploadServerHandler;

import java.util.Map;

/**
 * 큐에 등록된 요청 작업 처리 클래스
 * - Netty의 ChannelHandlerContext 및 요청 데이터를 보관하고 있다가
 * - process() 호출 시 핸들러에 데이터를 주입하고 URI 요청 처리를 실행한다.
 */
public class RequestTask {
	private final ChannelHandlerContext ctx;
	private final Map<String, Object> httpDataAttributes;
	private final HttpUploadServerHandler handler;
	private final HttpRequest request;
	
	public RequestTask(ChannelHandlerContext ctx, Map<String, Object> httpDataAttributes,
					   HttpUploadServerHandler handler, HttpRequest request) {
		this.ctx = ctx;
		this.httpDataAttributes = httpDataAttributes;
		this.handler = handler;
		this.request = request;
	}
	
	public void process() throws Exception {
		handler.setHttpDataAttributes(httpDataAttributes);
		handler.uriController(ctx, request);
	}
	
	public HttpRequest getRequest() {
		return request;
	}
	
	public Map<String, Object> getHttpDataAttributes() {
		return httpDataAttributes;
	}
	
	public HttpMethod getMethod() {
		return request.getMethod();
	}
	
	@Override
	public String toString() {
		return getRequestData();
	}
	
	public String getRequestData() {
		return String.format(
				"\n▶ RequestTask 정보\n" +
						"   - URI       : %s\n" +
						"   - Method    : %s\n" +
						"   - Client IP : %s\n" +
						"   - Params    : %s",
				request.getUri(),
				request.getMethod().name(),
				ctx.channel().remoteAddress(),
				httpDataAttributes
		);
	}
}

RequestQueueProcessor

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.LinkedBlockingQueue;

/**
 * 큐를 처리하는 쓰레드
 * - 애플리케이션 실행 시 시작되어 대기 상태에 들어간다.
 * - queue.put()이 호출되면 즉시 깨어나 해당 작업(RequestTask)을 수행한다.
 */
@Slf4j
public class RequestQueueProcessor extends Thread {
	private final LinkedBlockingQueue<RequestTask> queue;
	
	public RequestQueueProcessor(LinkedBlockingQueue<RequestTask> queue) {
		this.queue = queue;
	}
	
	@Override
	public void run() {
		while (!isInterrupted()) {
			try {
				RequestTask task = queue.take();
				log.info("▶ 큐 꺼냄 - Method: {}, URI: {}, Params: {}", task.getMethod().name(), task.getRequest().getUri(), task.getHttpDataAttributes());
				log.info(task.toString());
				task.process();
			} catch (InterruptedException e) {
				log.error("▷ 큐 꺼냄 실패", e);
				Thread.currentThread().interrupt(); // 인터럽트 상태 복구
				break; // 스레드 종료
			} catch (Exception e) {
				log.error("Error processing task", e);
				log.error("RequestTask 처리 중 예외", e);
			}
		}
		log.info("RequestQueueProcessor 스레드 종료");
	}
}

이런식으로 큐에 담아 하나씩 꺼내 처리하는 로직을 구현하였고, 테스트 결과 또한 큐잉으로 인한 병목이나 대기시간이 발생하지 않고 실시간르로 바로 처리 되는 것을 확인하였다.