diff --git a/src/main/java/com/eactive/eai/common/logger/DBLogTransactionLogger.java b/src/main/java/com/eactive/eai/common/logger/DBLogTransactionLogger.java index 7271bfb..94a891a 100644 --- a/src/main/java/com/eactive/eai/common/logger/DBLogTransactionLogger.java +++ b/src/main/java/com/eactive/eai/common/logger/DBLogTransactionLogger.java @@ -1,6 +1,7 @@ package com.eactive.eai.common.logger; +import java.util.List; import java.util.Properties; import com.eactive.eai.adapter.ElinkAdapter; @@ -176,6 +177,83 @@ public class DBLogTransactionLogger implements TransactionLogger { } } + /** + * 1. 기능 : 여러 건의 로깅 정보를 한 트랜잭션으로 적재 (COMMIT 1회) + * 2. 처리 개요 : + * - log()의 건별 처리와 달리 COMMIT 횟수를 배치 크기만큼 줄인다. + * - LOG_TYPE 판정 규칙은 log()과 동일하게 유지한다. + * 3. 주의사항 + * - 실시간 모니터링(EAIServiceMonitor) 전달은 발행측 EAILogSender.send()에서 + * 이미 처리하므로 여기서는 다루지 않는다. + * + * @param items {EAIMessage, Properties} 쌍의 목록 + **/ + public void logBatch(List items) { + if (items == null || items.isEmpty()) return; + logCount += items.size(); + try { + insertLogBatch(items); + } catch (Exception e) { + errCount++; + if (logger.isError()) logger.error("DBLogTransactionLogger] logBatch ERROR. - " + e.getMessage(), e); + } + } + + public static void insertLogBatch(List items) { + if (items == null || items.isEmpty()) return; + + // Direct DB Logging - 판정 규칙은 log()과 동일 + String logType = "DB"; + String setLogType = PropManager.getInstance().getProperty("LOG_TYPE"); + if (setLogType != null) { + logType = setLogType; + } + + if (!"DB".equals(logType) || !EAIDBLogControl.isEnable()) { + for (Object[] item : items) { + writeFileLog((EAIMessage) item[0], (Properties) item[1]); + } + return; + } + + EAILogBatchWriter writer = ApplicationContextProvider.getContext().getBean(EAILogBatchWriter.class); + try { + writer.writeBatch(items); + } catch (Exception be) { + // 배치는 한 건만 실패해도 트랜잭션 전체가 롤백된다. + // 정상 건까지 유실되지 않도록 건별 독립 트랜잭션으로 재시도한다. + if (logger.isError()) { + logger.error("DBLogTransactionLogger] insertLogBatch failed, retry one by one. size=" + items.size(), be); + } + for (Object[] item : items) { + EAIMessage eaiMessage = (EAIMessage) item[0]; + Properties prop = (Properties) item[1]; + try { + writer.writeOne(eaiMessage, prop); + } catch (Exception e) { + String message = e.getMessage(); + // DB Connection Error일 경우에만 DB 로깅을 중단한다. + if ("ConnectionError".equals(message) || StringUtils.contains(message, "JDBCConnectionException") + || StringUtils.contains(message, "Unable to acquire JDBC Connection")) { + EAIDBLogControl.setEnable(false); + } + if (logger.isError()) { + logger.error("DBLogTransactionLogger] insertLogBatch single retry failed. - " + message, e); + } + writeFileLog(eaiMessage, prop); + } + } + } + } + + private static void writeFileLog(EAIMessage eaiMessage, Properties prop) { + try { + EAIFileLogger.getInstance().setLog(eaiMessage, prop); + } catch (Exception fe) { + if (logger.isError()) logger.error("DBLogTransactionLogger] file log failed. - " + fe.getMessage(), fe); + } + } + public static void insertLog(EAIMessage eaiMessage, Properties prop) throws EAILogException { String guidLogPrefix = "DBLogTransactionLogger] GUID["+ eaiMessage.getMapper().getGuid(eaiMessage.getStandardMessage()) +"] UUID["+eaiMessage.getSvcOgNo()+"] "; diff --git a/src/main/java/com/eactive/eai/common/logger/EAILogBatchWriter.java b/src/main/java/com/eactive/eai/common/logger/EAILogBatchWriter.java new file mode 100644 index 0000000..ed93db1 --- /dev/null +++ b/src/main/java/com/eactive/eai/common/logger/EAILogBatchWriter.java @@ -0,0 +1,49 @@ +package com.eactive.eai.common.logger; + +import java.util.List; +import java.util.Properties; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Propagation; +import org.springframework.transaction.annotation.Transactional; + +import com.eactive.eai.common.message.EAIMessage; + +/** + * 1. 기능 : 비동기 거래로그를 여러 건 묶어 한 트랜잭션으로 적재 + * 2. 처리 개요 : + * - EAILogDAO는 클래스 레벨 @Transactional(기본 propagation REQUIRED)이므로 + * writeBatch 안에서 호출하면 별도 트랜잭션을 열지 않고 바깥 트랜잭션에 합류한다. + * - 결과적으로 N건이 COMMIT 1회로 처리되어 Oracle log file sync 대기가 1/N로 줄어든다. + * 3. 주의사항 + * - 배치 중 한 건이라도 예외가 나면 트랜잭션 전체가 롤백된다. + * 호출측(DBLogTransactionLogger.insertLogBatch)에서 건별 재시도로 폴백해야 한다. + */ +@Service +public class EAILogBatchWriter { + + @Autowired + private EAILogDAO dao; + + /** + * N건을 하나의 트랜잭션으로 적재한다. (COMMIT 1회) + * + * @param items {EAIMessage, Properties} 쌍의 목록 + */ + @Transactional + public void writeBatch(List items) throws Exception { + for (Object[] item : items) { + dao.addEAISvcLog((EAIMessage) item[0], (Properties) item[1]); + } + } + + /** + * 배치 실패 시 건별 재시도용. 각 건이 독립 트랜잭션이므로 + * 특정 건의 실패가 나머지 건에 영향을 주지 않는다. + */ + @Transactional(propagation = Propagation.REQUIRES_NEW) + public void writeOne(EAIMessage message, Properties prop) throws Exception { + dao.addEAISvcLog(message, prop); + } +} diff --git a/src/main/java/com/eactive/eai/common/logger/EAILogSender.java b/src/main/java/com/eactive/eai/common/logger/EAILogSender.java index da7eb59..60ee2de 100644 --- a/src/main/java/com/eactive/eai/common/logger/EAILogSender.java +++ b/src/main/java/com/eactive/eai/common/logger/EAILogSender.java @@ -1,5 +1,6 @@ package com.eactive.eai.common.logger; +import java.util.List; import java.util.Properties; import org.apache.commons.lang3.SerializationUtils; @@ -122,6 +123,30 @@ public class EAILogSender { } public static void logDirect(EAIMessage message, Properties prop) throws EAILogException { - txLogger.log(message, prop); + txLogger.log(message, prop); + } + + /** + * 여러 건을 한 트랜잭션(COMMIT 1회)으로 적재한다. + * 비동기 로깅 컨슈머(CustomEventHandler)가 모아둔 배치를 넘길 때 사용한다. + * + * @param items {EAIMessage, Properties} 쌍의 목록 + */ + public static void logDirectBatch(List items) { + if (items == null || items.isEmpty()) return; + + if (txLogger instanceof DBLogTransactionLogger) { + ((DBLogTransactionLogger) txLogger).logBatch(items); + return; + } + + // 배치를 지원하지 않는 TransactionLogger 구현이면 건별 처리로 폴백한다. + for (Object[] item : items) { + try { + txLogger.log((EAIMessage) item[0], (Properties) item[1]); + } catch (Exception e) { + logger.error("logDirectBatch fallback failed.", e); + } + } } } diff --git a/src/main/java/com/eactive/eai/common/logger/HttpLoggingService.java b/src/main/java/com/eactive/eai/common/logger/HttpLoggingService.java index a2bfb6f..5d80099 100644 --- a/src/main/java/com/eactive/eai/common/logger/HttpLoggingService.java +++ b/src/main/java/com/eactive/eai/common/logger/HttpLoggingService.java @@ -1,11 +1,14 @@ package com.eactive.eai.common.logger; +import java.util.List; + import com.eactive.eai.common.logger.mapper.HttpAdapterExtraLogMapper; import com.eactive.eai.common.util.Logger; import com.eactive.eai.data.entity.onl.logger.HttpAdapterExtraLog; import org.apache.commons.lang.StringUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; @Service @@ -22,6 +25,20 @@ public class HttpLoggingService { @Autowired private HttpAdapterExtraLogFileLogger fileLogger; + /** + * 여러 건을 한 트랜잭션(COMMIT 1회)으로 적재한다. + * HttpAdapterExtraLogLogger가 @Transactional(REQUIRED)이므로 이 트랜잭션에 합류한다. + * + * 배치 중 한 건이라도 실패하면 전체가 롤백되므로, + * 호출측에서 insertHttpAdapterExtraLog()로 건별 재시도해야 한다. + */ + @Transactional + public void insertHttpAdapterExtraLogBatch(List voList) throws Throwable { + for (HttpAdapterExtraLogVo vo : voList) { + dbLogger.save(mapper.toEntity(vo)); + } + } + public void insertHttpAdapterExtraLog(HttpAdapterExtraLogVo httpAdapterExtraLogVo) throws Throwable{ HttpAdapterExtraLog httpAdapterExtraLog = mapper.toEntity(httpAdapterExtraLogVo); if (EAIDBLogControl.isEnable()) { diff --git a/src/main/java/com/eactive/eai/common/logger/async/CustomEventHandler.java b/src/main/java/com/eactive/eai/common/logger/async/CustomEventHandler.java index d919060..d842bdf 100644 --- a/src/main/java/com/eactive/eai/common/logger/async/CustomEventHandler.java +++ b/src/main/java/com/eactive/eai/common/logger/async/CustomEventHandler.java @@ -3,59 +3,124 @@ package com.eactive.eai.common.logger.async; import java.util.ArrayList; import java.util.List; -import com.eactive.eai.common.logger.EAILogException; import com.eactive.eai.common.logger.EAILogSender; import com.eactive.eai.common.message.EAIMessage; import com.eactive.eai.common.util.Logger; import com.lmax.disruptor.EventHandler; +import com.lmax.disruptor.LifecycleAware; +import com.lmax.disruptor.TimeoutHandler; -public class CustomEventHandler implements EventHandler { +/** + * 1. 기능 : 거래로그 이벤트를 모아 한 트랜잭션(COMMIT 1회)으로 적재 + * 2. 처리 개요 : + * - Disruptor의 endOfBatch는 "지금 링버퍼에 더 처리할 이벤트가 없다"는 신호다. + * 이것을 flush 조건으로 쓰면 한산할 때는 건당 즉시 적재되어 지연이 늘지 않고, + * 부하가 몰릴 때만 배치가 커진다. 별도 타임아웃 flush 스레드가 필요 없다. + * - batchSize는 하한이 아니라 상한이다. "100건 모일 때까지 대기"가 아니라 + * "한 트랜잭션이 100건을 넘지 않게 끊는다"는 의미다. + * - 안전망으로 TimeoutHandler를 구현한다. 유휴 상태가 지속되면 Disruptor가 + * onTimeout()을 호출하므로, 버퍼에 남은 로그가 방치되지 않는다. + * (WaitStrategy가 TimeoutBlockingWaitStrategy = 설정값 "TIME"일 때 동작. 기본값) + * 3. 주의사항 + * - LoggingEvent.clear()는 EAIMessage 내용까지 비우므로 버퍼에 담은 뒤 호출하면 안 된다. + * 슬롯 참조만 끊고, EAIMessage 해제는 적재 완료 후 releaseMessages()에서 처리한다. + */ +public class CustomEventHandler implements EventHandler, LifecycleAware, TimeoutHandler { static Logger logger = Logger.getLogger(Logger.LOGGER_DEFAULT); - String name; - int sleepMs; - private int batchSize = 1; - - private int count = 0; - private final List eventList = new ArrayList<>(); - - public CustomEventHandler() { - } + + private final String name; + private final int sleepMs; + private final int batchSize; + + private final List buffer; public CustomEventHandler(String name, int sleepMs, int batchSize) { this.name = name; this.sleepMs = sleepMs; - this.batchSize = batchSize; + this.batchSize = (batchSize < 1) ? 1 : batchSize; + this.buffer = new ArrayList(this.batchSize); } @Override public void onEvent(LoggingEvent event, long sequence, boolean endOfBatch) throws Exception { - if(sleepMs > 0) Thread.sleep(sleepMs); - if(batchSize > 1) { - eventList.add(event); - if (++count >= batchSize) { - processBatch(); - eventList.clear(); - count = 0; - } + if (sleepMs > 0) Thread.sleep(sleepMs); + + EAIMessage message = event.getMessage(); + if (message != null) { + buffer.add(new Object[] { message, event.getProperty() }); + } + + // 링버퍼 슬롯은 재사용되므로 참조만 끊는다. (event.clear() 사용 금지 - 상단 주석 참고) + event.setMessage(null); + event.setProperty(null); + + if (endOfBatch || buffer.size() >= batchSize) { + flush(); } - else { - EAIMessage message = event.getMessage(); - if(logger.isInfo()) { - logger.info(String.format("CustomWorkHandler: %s LoggingEvent: %s %s\n" - ,name, message.getSvcOgNo() ,message.getLogPssSno()) - ); - } - EAILogSender.logDirect(event.getMessage(), event.getProperty()); - if(event != null) { - event.clear(); - event = null; - } - } } - private void processBatch() throws EAILogException { - for(LoggingEvent event:eventList) { - EAILogSender.logDirect(event.getMessage(), event.getProperty()); + private void flush() { + if (buffer.isEmpty()) return; + int size = buffer.size(); + // info 레벨이 꺼져 있으면 시각 측정 자체를 하지 않는다. + final boolean measure = logger.isInfo(); + final long t0 = measure ? System.currentTimeMillis() : 0L; + try { + EAILogSender.logDirectBatch(buffer); + } catch (Throwable th) { + logger.error(String.format("%s] batch log failed. size=%d", name, size), th); + } finally { + // DB 적재 시간만 측정한다. releaseMessages()는 EAIMessage 100건의 + // setBizData(null)/svcMsgs.clear()를 도는 비용이라 측정에 섞이면 안 된다. + final long elapsed = measure ? (System.currentTimeMillis() - t0) : 0L; + releaseMessages(); + buffer.clear(); + if (measure) { + logger.info("{} batch logging. size={}, elapsed={}ms", name, size, elapsed); + } + } + } + + /** 적재가 끝난 EAIMessage의 내부 버퍼를 해제한다. (기존 LoggingEvent.clear()가 하던 역할) */ + private void releaseMessages() { + for (Object[] item : buffer) { + EAIMessage message = (EAIMessage) item[0]; + if (message == null) continue; + try { + message.clear(); + } catch (Exception e) { + // 해제 실패는 적재 결과에 영향이 없으므로 무시한다. + } + } + } + + /** + * 유휴 시 Disruptor가 호출하는 안전망. + * 정상 흐름에서는 endOfBatch로 이미 flush되지만, onEvent가 예외로 중단되어 + * 버퍼가 남은 뒤 거래가 끊기는 경우를 대비한다. + */ + @Override + public void onTimeout(long sequence) throws Exception { + if (buffer.isEmpty()) return; + if (logger.isDebug()) { + logger.debug(String.format("%s] flush by timeout. remain=%d", name, buffer.size())); + } + flush(); + } + + @Override + public void onStart() { + if (logger.isWarn()) { + logger.warn(String.format(">> %s start. batchSize = %d", name, batchSize)); + } + } + + /** 컨슈머 스레드 종료 시 버퍼에 남은 로그를 반드시 적재한다. */ + @Override + public void onShutdown() { + flush(); + if (logger.isWarn()) { + logger.warn(String.format("<< %s shutdown.", name)); } } } diff --git a/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingBatchEventHandler.java b/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingBatchEventHandler.java new file mode 100644 index 0000000..d01c651 --- /dev/null +++ b/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingBatchEventHandler.java @@ -0,0 +1,135 @@ +package com.eactive.eai.common.logger.async; + +import java.util.ArrayList; +import java.util.List; + +import com.eactive.eai.common.logger.EAIDBLogControl; +import com.eactive.eai.common.logger.HttpAdapterExtraLogVo; +import com.eactive.eai.common.logger.HttpLoggingService; +import com.eactive.eai.common.util.ApplicationContextProvider; +import com.eactive.eai.common.util.Logger; +import com.lmax.disruptor.EventHandler; +import com.lmax.disruptor.LifecycleAware; +import com.lmax.disruptor.TimeoutHandler; + +/** + * 1. 기능 : HTTP 헤더 로그 이벤트를 모아 한 트랜잭션(COMMIT 1회)으로 적재 + * 2. 처리 개요 : + * - endOfBatch를 flush 조건으로 사용한다. 한산할 때는 건당 즉시 적재되고 + * 부하가 몰릴 때만 배치가 커지므로 별도 타임아웃 스레드가 필요 없다. + * - batchSize는 하한이 아니라 상한이다. "100건 모일 때까지 대기"가 아니라 + * "한 트랜잭션이 100건을 넘지 않게 끊는다"는 의미다. + * - 안전망으로 TimeoutHandler를 구현한다. 유휴 시 Disruptor가 onTimeout()을 + * 호출하므로 버퍼에 남은 로그가 방치되지 않는다. + * (WaitStrategy가 TimeoutBlockingWaitStrategy = 설정값 "TIME"일 때 동작. 기본값) + * - 배치 실패 시 기존 건별 경로(insertHttpAdapterExtraLog)로 재시도한다. + * 그 경로가 DB 장애 판정과 파일로그 폴백을 이미 담고 있다. + * 3. 주의사항 + * - HttpLoggingService 빈은 필드에 캐싱한다. 이벤트마다 타입 기반 getBean을 + * 호출하면 컨슈머 처리량이 떨어진다. + */ +public class HttpLoggingBatchEventHandler implements EventHandler, LifecycleAware, TimeoutHandler { + static Logger logger = Logger.getLogger(Logger.LOGGER_DEFAULT); + + private final String name; + private final int batchSize; + + private final List buffer; + + private HttpLoggingService service; + + public HttpLoggingBatchEventHandler(String name, int batchSize) { + this.name = name; + this.batchSize = (batchSize < 1) ? 1 : batchSize; + this.buffer = new ArrayList(this.batchSize); + } + + private HttpLoggingService service() { + if (service == null) { + service = ApplicationContextProvider.getContext().getBean(HttpLoggingService.class); + } + return service; + } + + @Override + public void onEvent(HttpLoggingEvent event, long sequence, boolean endOfBatch) throws Exception { + HttpAdapterExtraLogVo vo = event.getHttpAdapterExtraLogVo(); + if (vo != null) { + buffer.add(vo); + } + // 링버퍼 슬롯은 재사용되므로 참조를 끊는다. VO 내용은 유지된다. + event.clear(); + + if (endOfBatch || buffer.size() >= batchSize) { + flush(); + } + } + + private void flush() { + if (buffer.isEmpty()) return; + int size = buffer.size(); + // info 레벨이 꺼져 있으면 시각 측정 자체를 하지 않는다. + final boolean measure = logger.isInfo(); + final long t0 = measure ? System.currentTimeMillis() : 0L; + try { + if (EAIDBLogControl.isEnable()) { + service().insertHttpAdapterExtraLogBatch(buffer); + if (logger.isDebug()) { + logger.debug(String.format("%s] flushed %d http log(s) in one transaction", name, size)); + } + } else { + writeEach(); + } + } catch (Throwable th) { + // 배치는 한 건만 실패해도 전체가 롤백된다. 건별로 재시도해 정상 건을 살린다. + logger.error(String.format("%s] batch http log failed, retry one by one. size=%d", name, size), th); + writeEach(); + } finally { + buffer.clear(); + } + if (measure) { + logger.info("{} batch logging. size={}, elapsed={}ms", name, size, System.currentTimeMillis() - t0); + } + } + + /** 기존 건별 경로. DB 장애 판정과 파일로그 폴백이 이 안에 있다. */ + private void writeEach() { + for (HttpAdapterExtraLogVo vo : buffer) { + try { + service().insertHttpAdapterExtraLog(vo); + } catch (Throwable th) { + logger.error("failed to insert async http log ", th); + } + } + } + + /** + * 유휴 시 Disruptor가 호출하는 안전망. + * 정상 흐름에서는 endOfBatch로 이미 flush되지만, onEvent가 예외로 중단되어 + * 버퍼가 남은 뒤 거래가 끊기는 경우를 대비한다. + */ + @Override + public void onTimeout(long sequence) throws Exception { + if (buffer.isEmpty()) return; + if (logger.isDebug()) { + logger.debug(String.format("%s] flush by timeout. remain=%d", name, buffer.size())); + } + flush(); + } + + @Override + public void onStart() { + if (logger.isWarn()) { + logger.warn(String.format(">> %s start. batchSize = %d", name, batchSize)); + } + } + + /** 컨슈머 스레드 종료 시 버퍼에 남은 로그를 반드시 적재한다. */ + @Override + public void onShutdown() { + flush(); + if (logger.isWarn()) { + logger.warn(String.format("<< %s shutdown.", name)); + } + } +} diff --git a/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingEvent.java b/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingEvent.java index e403c17..30b1145 100644 --- a/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingEvent.java +++ b/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingEvent.java @@ -8,6 +8,14 @@ import lombok.Data; public class HttpLoggingEvent { private HttpAdapterExtraLogVo httpAdapterExtraLogVo; + /** + * 링버퍼 슬롯의 참조만 끊는다. + * VO 자체는 컨슈머가 배치 버퍼에 담아 사용하므로 내용을 비우면 안 된다. + */ + public void clear() { + this.httpAdapterExtraLogVo = null; + } + public final static EventFactory EVENT_FACTORY = new EventFactory() { public HttpLoggingEvent newInstance() { return new HttpLoggingEvent(); diff --git a/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingPoolObject.java b/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingPoolObject.java index 29b5f15..746a6bc 100644 --- a/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingPoolObject.java +++ b/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingPoolObject.java @@ -17,6 +17,8 @@ public class HttpLoggingPoolObject { Disruptor disruptor = null; RingBuffer ringBuffer = null; int id = 0; + // 실제 링버퍼 크기로 생성자에서 설정한다. shutdown()이 이 값과 remainingCapacity를 + // 비교하므로 하드코딩하면 queue.size 변경 시 종료되지 않는다. int queueMax = (int)Math.pow(2, 10); int workerSize = 0; @@ -25,7 +27,6 @@ public class HttpLoggingPoolObject { } private WaitStrategy getWaitStrategy(String waitStrategy) { - WaitStrategy ws = null; if(ConfigKeys.LOGGER_ASYNC_WAITSTRATEGY_BLOCK.equals(waitStrategy)) { // throughput and low-latency are not as important as CPU resource return new BlockingWaitStrategy(); @@ -47,14 +48,23 @@ public class HttpLoggingPoolObject { if(ConfigKeys.LOGGER_ASYNC_WAITSTRATEGY_YIELD.equals(waitStrategy)) { return new YieldingWaitStrategy(); } - return ws; + // 알 수 없는 값이면 null이 아니라 기본값(TIME)을 돌려준다. + // null을 넘기면 컨슈머 스레드가 NPE로 죽어 로깅이 통째로 멈춘다. + // TimeoutBlockingWaitStrategy여야 배치 핸들러의 onTimeout 안전망도 동작한다. + if(waitStrategy != null && logger.isWarn()) { + logger.warn(String.format(">> unknown waitStrategy [%s], fallback to TIME(100ms)", waitStrategy)); + } + return new TimeoutBlockingWaitStrategy(100 * 1000, TimeUnit.MICROSECONDS); } - public HttpLoggingPoolObject(int id, int queueSize, int workerSize, String waitStrategy) { + public HttpLoggingPoolObject(int id, int queueSize, int workerSize, String waitStrategy, int batchSize) { this.id = id; + this.queueMax = queueSize; + this.workerSize = workerSize; if(logger.isWarn()) { logger.warn(String.format(">> Disruptor-%d queueSize = %d", id, queueSize)); logger.warn(String.format(">> Disruptor-%d workerSize = %d", id, workerSize)); + logger.warn(String.format(">> Disruptor-%d batchSize = %d", id, batchSize)); logger.warn(String.format(">> Disruptor-%d waitStrategy = %s", id, waitStrategy)); } CustomThreadFactory tFactory = new CustomThreadFactory(); @@ -63,12 +73,21 @@ public class HttpLoggingPoolObject { ProducerType.SINGLE, getWaitStrategy(waitStrategy)); // BlockingWaitStrategy | SleepingWaitStrategy | YieldingWaitStrategy | BusySpinWaitStrategy - WorkHandler[] handlers = new WorkHandler[workerSize]; - for(int i=0; i< handlers.length; i++) { - WorkHandler handler = new HttpLoggingWorkHandler(); - handlers[i] = handler; + if(workerSize > 1) { + // 건별 COMMIT 경로. WorkHandler에는 endOfBatch가 없어 배치 적재가 불가능하다. + WorkHandler[] handlers = new WorkHandler[workerSize]; + for(int i=0; i< handlers.length; i++) { + WorkHandler handler = new HttpLoggingWorkHandler(); + handlers[i] = handler; + } + disruptor.handleEventsWithWorkerPool(handlers); + } + else { + // 배치 COMMIT 경로. 컨슈머 병렬도는 디스크럽터 풀 개수(pool.maxsize)로 확보한다. + HttpLoggingBatchEventHandler handler = + new HttpLoggingBatchEventHandler(String.format("HttpLoggingBatchEventHandler%d-%d", id, 0), batchSize); + disruptor.handleEventsWith(handler); } - disruptor.handleEventsWithWorkerPool(handlers); try { diff --git a/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingPoolObjectFactory.java b/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingPoolObjectFactory.java index f1362c8..6bfb304 100644 --- a/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingPoolObjectFactory.java +++ b/src/main/java/com/eactive/eai/common/logger/async/HttpLoggingPoolObjectFactory.java @@ -18,7 +18,8 @@ public class HttpLoggingPoolObjectFactory extends BasePooledObjectFactory disruptor = null; RingBuffer ringBuffer = null; int id = 0; + // 실제 링버퍼 크기로 생성자에서 설정한다. shutdown()이 이 값과 remainingCapacity를 + // 비교하므로 하드코딩하면 queue.size 변경 시 종료되지 않는다. int queueMax = (int)Math.pow(2, 10); int workerSize = 0; @@ -30,7 +32,6 @@ public class LoggingPoolObject { } private WaitStrategy getWaitStrategy(String waitStrategy) { - WaitStrategy ws = null; if(ConfigKeys.LOGGER_ASYNC_WAITSTRATEGY_BLOCK.equals(waitStrategy)) { // throughput and low-latency are not as important as CPU resource return new BlockingWaitStrategy(); @@ -52,33 +53,45 @@ public class LoggingPoolObject { if(ConfigKeys.LOGGER_ASYNC_WAITSTRATEGY_YIELD.equals(waitStrategy)) { return new YieldingWaitStrategy(); } - return ws; + // 알 수 없는 값이면 null이 아니라 기본값(TIME)을 돌려준다. + // null을 넘기면 컨슈머 스레드가 NPE로 죽어 로깅이 통째로 멈춘다. + // TimeoutBlockingWaitStrategy여야 배치 핸들러의 onTimeout 안전망도 동작한다. + if(waitStrategy != null && logger.isWarn()) { + logger.warn(String.format(">> unknown waitStrategy [%s], fallback to TIME(100ms)", waitStrategy)); + } + return new TimeoutBlockingWaitStrategy(100 * 1000, TimeUnit.MICROSECONDS); } - public LoggingPoolObject(int id, int queueSize, int workerSize, String waitStrategy) { + public LoggingPoolObject(int id, int queueSize, int workerSize, String waitStrategy, int batchSize) { this.id = id; + this.queueMax = queueSize; + this.workerSize = workerSize; if(logger.isWarn()) { logger.warn(String.format(">> Disruptor-%d queueSize = %d", id, queueSize)); logger.warn(String.format(">> Disruptor-%d workerSize = %d", id, workerSize)); + logger.warn(String.format(">> Disruptor-%d batchSize = %d", id, batchSize)); logger.warn(String.format(">> Disruptor-%d waitStrategy = %s", id, waitStrategy)); } CustomThreadFactory tFactory = new CustomThreadFactory(); disruptor = new Disruptor(LoggingEvent.EVENT_FACTORY, queueSize, tFactory, - ProducerType.SINGLE, + ProducerType.SINGLE, getWaitStrategy(waitStrategy)); // BlockingWaitStrategy | SleepingWaitStrategy | YieldingWaitStrategy | BusySpinWaitStrategy if(workerSize > 1) { + // 건별 COMMIT 경로. WorkHandler에는 endOfBatch가 없어 배치 적재가 불가능하다. + // 커밋 횟수를 줄이려면 worker.size=1로 두고 아래 배치 EventHandler를 사용한다. WorkHandler[] handlers = new WorkHandler[workerSize]; for(int i=0; i< handlers.length; i++) { // TODO : 현재는 delay 없이 처리하도록 하고, 추후 DB부하를 줄이려면 sleep을 정의. - CustomWorkHandler handler = new CustomWorkHandler(String.format("CustomWorkHandler%d-%d",id, i), 0, 1); + CustomWorkHandler handler = new CustomWorkHandler(String.format("CustomWorkHandler%d-%d",id, i), 0, 1); handlers[i] = handler; } - disruptor.handleEventsWithWorkerPool(handlers); + disruptor.handleEventsWithWorkerPool(handlers); } else { - CustomEventHandler handler = new CustomEventHandler(String.format("CustomEventHandler%d-%d",id, 0), 0, 1); + // 배치 COMMIT 경로. 컨슈머 병렬도는 디스크럽터 풀 개수(pool.maxsize)로 확보한다. + CustomEventHandler handler = new CustomEventHandler(String.format("CustomEventHandler%d-%d",id, 0), 0, batchSize); disruptor.handleEventsWith(handler); } diff --git a/src/main/java/com/eactive/eai/common/logger/async/LoggingPoolObjectFactory.java b/src/main/java/com/eactive/eai/common/logger/async/LoggingPoolObjectFactory.java index d036a68..7d53f15 100644 --- a/src/main/java/com/eactive/eai/common/logger/async/LoggingPoolObjectFactory.java +++ b/src/main/java/com/eactive/eai/common/logger/async/LoggingPoolObjectFactory.java @@ -19,7 +19,8 @@ public class LoggingPoolObjectFactory extends BasePooledObjectFactory asyncDummyData = new ConcurrentHashMap(); @@ -71,6 +72,7 @@ public class ElinkConfig implements ConfigKeys { sb.append( String.format("%s = %s\n", LOGGER_ASYNC_INITPOOLS, asyncPoolInitSize) ); sb.append( String.format("%s = %s\n", LOGGER_ASYNC_QUEUES, asyncQueueSize) ); sb.append( String.format("%s = %s\n", LOGGER_ASYNC_WORKERS, asyncWorkers) ); + sb.append( String.format("%s = %s\n", LOGGER_ASYNC_BATCHSIZE, asyncBatchSize) ); sb.append( String.format("%s = %s\n", LOGGER_ASYNC_WAITSTRATEGY, waitStrategy) ); sb.append(">> Async Dummy Configuration\n"); sb.append(String.format("%s = %s\n", HTTP_ASYNC_DEFAULT_DUMMY_DATA, asyncDefaultDummyData)); @@ -169,6 +171,16 @@ public class ElinkConfig implements ConfigKeys { asyncWorkers = 16; } + try { + sCount = env.getProperty(LOGGER_ASYNC_BATCHSIZE, "100"); + asyncBatchSize = Integer.parseInt(sCount); + if (asyncBatchSize < 1) { + asyncBatchSize = 1; + } + } catch (Exception ex) { + asyncBatchSize = 100; + } + try { waitStrategy = env.getProperty(LOGGER_ASYNC_WAITSTRATEGY, LOGGER_ASYNC_WAITSTRATEGY_TIME); } catch (Exception ex) { @@ -271,6 +283,14 @@ public class ElinkConfig implements ConfigKeys { ElinkConfig.asyncWorkers = asyncWorkers; } + public static int getAsyncBatchSize() { + return asyncBatchSize; + } + + public static void setAsyncBatchSize(int asyncBatchSize) { + ElinkConfig.asyncBatchSize = asyncBatchSize; + } + public static String getWaitStrategy() { return waitStrategy; }