작업 중 발생한 Redis 동시성 문제 발생과 처리

 

 

Collector 모듈 동작 중 동시성 문제가 발생했다.

원인은 꽤 명확한 것인데, Redis 객체 insert에 해당하는 Collection 객체가 Synchronized하지 않기 때문.

 

 

   ...
   
   //스케줄러 동작 여하에 따라 데이터 임시 보관용 List
    private final List<MetricDto> keepedMetricDtos = new ArrayList<>(); //Metric Table
    private final List<EtcDto> keepedEtcDtos = new ArrayList<>(); //예외 체크용

    //Redis Saver
    private final MetricRedisSaver metricRedisSaver = new MetricRedisSaver();

    //Psql
    private final AgentSaver agentSaver = new AgentSaver();
    private final MetricSaver metricSaver = new MetricSaver();
    private final EtcValueSaver etcValueSaver = new EtcValueSaver();
    private final SumAndGaugeSaver sumAndGaugeSaver = new SumAndGaugeSaver();
    private final HistogramSaver histogramSaver = new HistogramSaver();




    public void save(List<MetricDto> metricDtoList) {
        checkMetricInnerList(metricDtoList);
        if(checkIsScheduling()) return;   //테이블 재생성 로직 동작 중에는 로직 중지
        else jdbcBatchStart();
    }

    /** Batch Start **/
    public void jdbcBatchStart() {
        try {
            agentSaver.insert(keepedMetricDtos.stream().map(MetricDto::getAgentName).filter(Objects::nonNull).collect(Collectors.toSet()));
            metricSaver.insert(keepedMetricDtos);       //Metric table
            sumAndGaugeSaver.insert(keepedMetricDtos);  //sum and gauge table
            histogramSaver.insert(keepedMetricDtos);    //hist table
            etcValueSaver.insert(keepedEtcDtos);        //unknown table
            metricRedisSaver.insert(keepedMetricDtos);  //Redis save **** 이 부분에서 동시성 문제가 발생한다. ***
            if (LOG_SHOW_METRIC_SAVEDATA_LENGTH) addLog(keepedMetricDtos); //Log
        } catch (Exception e) {
            ErrorLogger.log("jdbcBatchStart batch fail", this.getClass().getName(), e);
        } finally {
            //Data Clear -> exception 시 데이터 유실됨.
            keepedMetricDtos.clear();
            keepedEtcDtos.clear();
        }
    }

 

아무래도 RedisSaver 객체의 비동기 로직 때문에 이의 문제가 발생한 것 같다.

 

Redis name Publisher 객체 보유 및 pub-sub / pipeline / streams를 수행하는 주 객체

 

 

결국 Method 단위에 Synchronized를 걸던가, 현재 공유하여 사용되는 keep .. List 객체를 동기화 처리 해야 한다.

조금 알아봤는데, 다음과 같은 대안이 있었다.

 

1. synchronizedList

기본적으로 개별 메서드 호출에 대한 동기화를 제공한다.  하지만 복합 연산(예: contains와 add의 조합)이나 이터레이션 중 수정에는 여전히 문제가 발생할 수 있다.

 

• 복합 연산: 여러 메서드 호출을 조합한 작업에서는 synchronizedList만으로는 충분하지 않다. 해당 블록 전체를 명시적으로 동기화해야 한다.
• 이터레이션 중 수정: synchronizedList는 이터레이션 중 리스트가 수정되면 ConcurrentModificationException이 발생할 수 있다. 이터레이션 시에도 리스트를 동기화 블록으로 감싸야 한다.

2. 복잡한 연산에 대한 대처

• CopyOnWriteArrayList: 모든 쓰기 연산에서 내부 배열을 복사하여 동작하므로, 이터레이션 중 동시 수정 문제를 피할 수 있다. 쓰기 연산이 적고 읽기 연산이 많은 경우 적합하다.
• ConcurrentHashMap: 리스트 대신 맵이 필요하다면, 세분화된 동기화를 제공하는 ConcurrentHashMap을 사용하는 것이 좋다.

 

 

현재 로직 상 딱히 복합한 Collection method나 연산을 사용하지 않고, 단순 List 순회 뒤 copy하는 로직이므로, Collection 객체 대신 SyncronizedList를 사용했다.

 

 

2024-09-02 23:56:24 [grpc-default-executor-1] WARN  util.log.ErrorLogger - [ERR LOG] jdbcBatchStart batch fail
2024-09-02 23:56:24 [grpc-default-executor-1] WARN  util.log.ErrorLogger - [POSITION] psql.insert.metrics.MetricDataSaveManager
2024-09-02 23:56:24 [grpc-default-executor-1] WARN  util.log.ErrorLogger - ----------------------------------------
java.util.ConcurrentModificationException
	at java.util.HashMap$HashIterator.nextNode(HashMap.java:1445)
	at java.util.HashMap$ValueIterator.next(HashMap.java:1474)
	at psql.insert.metrics.saver.SumAndGaugeSaver.insert(SumAndGaugeSaver.java:51)
	at psql.insert.metrics.MetricDataSaveManager.jdbcBatchStart(MetricDataSaveManager.java:58)
	at psql.insert.metrics.MetricDataSaveManager.save(MetricDataSaveManager.java:50)
	at grpc.MetricServiceImpl.export(MetricServiceImpl.java:21)
	at io.opentelemetry.proto.collector.metrics.v1.MetricsServiceGrpc$MethodHandlers.invoke(MetricsServiceGrpc.java:256)
	at io.grpc.stub.ServerCalls$UnaryServerCallHandler$UnaryServerCallListener.onHalfClose(ServerCalls.java:182)
	at io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.halfClosed(ServerCallImpl.java:351)
	at io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1HalfClosed.runInContext(ServerImpl.java:861)
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
2024-09-02 23:56:32 [grpc-default-executor-3] INFO  util.log.NoticeLogger - [NOTICE] [trace]save => Agent Name(Groupware_webmail_0829) :: [spans(6), attributes(62)]

 

 

여전히 동시성 처리 문제가 발생한다.

순회의 경우 상관없을 줄 알았는데, Iterator 사용으로 제약이 생기는 것으로 추정된다.

 

비동기 로직 부분만 Synchronized Block으로 감싸주었다.

    /** Batch Start **/
    public void jdbcBatchStart() {
        try {
            agentSaver.insert(keepedMetricDtos.stream().map(MetricDto::getAgentName).filter(Objects::nonNull).collect(Collectors.toSet()));
            metricSaver.insert(keepedMetricDtos);       //Metric table
            sumAndGaugeSaver.insert(keepedMetricDtos);  //sum and gauge table
            histogramSaver.insert(keepedMetricDtos);    //hist table
            etcValueSaver.insert(keepedEtcDtos);        //unknown table
            synchronized (keepedMetricDtos) {
                metricRedisSaver.insert(keepedMetricDtos);  // Redis save
            }
            if (LOG_SHOW_METRIC_SAVEDATA_LENGTH) addLog(keepedMetricDtos); //Log
        } catch (Exception e) {
            ErrorLogger.log("jdbcBatchStart batch fail", this.getClass().getName(), e);
        } finally {
            //Data Clear -> exception 시 데이터 유실됨.
            keepedMetricDtos.clear();
            keepedEtcDtos.clear();
        }
    }

 

 

에러 없이 정상적으로 돌아가는 모습. 

코드 확인할 때 비동기 부분을 면밀하게 확인해야겠다.