djb-eapim/elink-online-common
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

비동기 로그 배치 커밋 적용

Browse Source
This commit is contained in:
curry772
2026-08-12 13:15:58 +09:00
parent 3f75bd9b49
commit 398c26537f
13 changed files with 488 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/eactive/eai/common/logger/DBLogTransactionLogger.java
+78
View File
@@ -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<Object[]> 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<Object[]> 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()+"] ";
src/main/java/com/eactive/eai/common/logger/EAILogBatchWriter.java
+49
View File
@@ -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<Object[]> 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);
}
}
src/main/java/com/eactive/eai/common/logger/EAILogSender.java
+25
View File
@@ -1,5 +1,6 @@
package com.eactive.eai.common.logger;
import java.util.List;
import java.util.Properties;
import org.apache.commons.lang3.SerializationUtils;
@@ -124,4 +125,28 @@ public class EAILogSender {
public static void logDirect(EAIMessage message, Properties prop) throws EAILogException {
txLogger.log(message, prop);
}
/**
* 여러 건을 한 트랜잭션(COMMIT 1회)으로 적재한다.
* 비동기 로깅 컨슈머(CustomEventHandler)가 모아둔 배치를 넘길 때 사용한다.
*
* @param items {EAIMessage, Properties} 쌍의 목록
*/
public static void logDirectBatch(List<Object[]> 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);
}
}
}
}
src/main/java/com/eactive/eai/common/logger/HttpLoggingService.java
+17
View File
@@ -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<HttpAdapterExtraLogVo> 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()) {
src/main/java/com/eactive/eai/common/logger/async/CustomEventHandler.java
+95 -30
View File
@@ -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<LoggingEvent> {
/**
* 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<LoggingEvent>, 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<LoggingEvent> eventList = new ArrayList<>();
private final String name;
private final int sleepMs;
private final int batchSize;
public CustomEventHandler() {
}
private final List<Object[]> 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<Object[]>(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;
}
}
else {
EAIMessage message = event.getMessage();
if(logger.isInfo()) {
logger.info(String.format("CustomWorkHandler: %s LoggingEvent: %s %s\n"
,name, message.getSvcOgNo() ,message.getLogPssSno())
);
if (message != null) {
buffer.add(new Object[] { message, event.getProperty() });
}
EAILogSender.logDirect(event.getMessage(), event.getProperty());
if(event != null) {
event.clear();
event = null;
// 링버퍼 슬롯은 재사용되므로 참조만 끊는다. (event.clear() 사용 금지 - 상단 주석 참고)
event.setMessage(null);
event.setProperty(null);
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 {
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);
}
}
}
private void processBatch() throws EAILogException {
for(LoggingEvent event:eventList) {
EAILogSender.logDirect(event.getMessage(), event.getProperty());
/** 적재가 끝난 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));
}
}
}
src/main/java/com/eactive/eai/common/logger/async/HttpLoggingBatchEventHandler.java
+135
View File
@@ -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<HttpLoggingEvent>, LifecycleAware, TimeoutHandler {
static Logger logger = Logger.getLogger(Logger.LOGGER_DEFAULT);
private final String name;
private final int batchSize;
private final List<HttpAdapterExtraLogVo> buffer;
private HttpLoggingService service;
public HttpLoggingBatchEventHandler(String name, int batchSize) {
this.name = name;
this.batchSize = (batchSize < 1) ? 1 : batchSize;
this.buffer = new ArrayList<HttpAdapterExtraLogVo>(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));
}
}
}
src/main/java/com/eactive/eai/common/logger/async/HttpLoggingEvent.java
+8
View File
@@ -8,6 +8,14 @@ import lombok.Data;
public class HttpLoggingEvent {
private HttpAdapterExtraLogVo httpAdapterExtraLogVo;
/**
* 링버퍼 슬롯의 참조만 끊는다.
* VO 자체는 컨슈머가 배치 버퍼에 담아 사용하므로 내용을 비우면 안 된다.
*/
public void clear() {
this.httpAdapterExtraLogVo = null;
}
public final static EventFactory<HttpLoggingEvent> EVENT_FACTORY = new EventFactory<HttpLoggingEvent>() {
public HttpLoggingEvent newInstance() {
return new HttpLoggingEvent();
src/main/java/com/eactive/eai/common/logger/async/HttpLoggingPoolObject.java
+22 -3
View File
@@ -17,6 +17,8 @@ public class HttpLoggingPoolObject {
Disruptor<HttpLoggingEvent> disruptor = null;
RingBuffer<HttpLoggingEvent> 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
if(workerSize > 1) {
// 건별 COMMIT 경로. WorkHandler에는 endOfBatch가 없어 배치 적재가 불가능하다.
WorkHandler<HttpLoggingEvent>[] 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);
}
try {
src/main/java/com/eactive/eai/common/logger/async/HttpLoggingPoolObjectFactory.java
+2 -1
View File
@@ -18,7 +18,8 @@ public class HttpLoggingPoolObjectFactory extends BasePooledObjectFactory<HttpLo
int queueSize = ElinkConfig.getAsyncQueueSize();
int workers = ElinkConfig.getAsyncWorkers();
String waitStrategy = ElinkConfig.getWaitStrategy();
return new HttpLoggingPoolObject(i++, queueSize, workers, waitStrategy);
int batchSize = ElinkConfig.getAsyncBatchSize();
return new HttpLoggingPoolObject(i++, queueSize, workers, waitStrategy, batchSize);
}
@Override
src/main/java/com/eactive/eai/common/logger/async/LoggingPoolObject.java
+17 -4
View File
@@ -22,6 +22,8 @@ public class LoggingPoolObject {
Disruptor<LoggingEvent> disruptor = null;
RingBuffer<LoggingEvent> 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,14 +53,23 @@ 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();
@@ -69,6 +79,8 @@ public class LoggingPoolObject {
getWaitStrategy(waitStrategy));
// BlockingWaitStrategy | SleepingWaitStrategy | YieldingWaitStrategy | BusySpinWaitStrategy
if(workerSize > 1) {
// 건별 COMMIT 경로. WorkHandler에는 endOfBatch가 없어 배치 적재가 불가능하다.
// 커밋 횟수를 줄이려면 worker.size=1로 두고 아래 배치 EventHandler를 사용한다.
WorkHandler<LoggingEvent>[] handlers = new WorkHandler[workerSize];
for(int i=0; i< handlers.length; i++) {
// TODO : 현재는 delay 없이 처리하도록 하고, 추후 DB부하를 줄이려면 sleep을 정의.
@@ -78,7 +90,8 @@ public class LoggingPoolObject {
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);
}
src/main/java/com/eactive/eai/common/logger/async/LoggingPoolObjectFactory.java
+2 -1
View File
@@ -19,7 +19,8 @@ public class LoggingPoolObjectFactory extends BasePooledObjectFactory<LoggingPoo
int queueSize = ElinkConfig.getAsyncQueueSize();
int workers = ElinkConfig.getAsyncWorkers();
String waitStrategy = ElinkConfig.getWaitStrategy();
return new LoggingPoolObject(i++, queueSize, workers, waitStrategy);
int batchSize = ElinkConfig.getAsyncBatchSize();
return new LoggingPoolObject(i++, queueSize, workers, waitStrategy, batchSize);
}
@Override
src/main/java/com/eactive/eai/env/ConfigKeys.java
Vendored
+2
View File
@@ -13,6 +13,8 @@ public interface ConfigKeys {
public static final String LOGGER_ASYNC_INITPOOLS = "logger.async.pool.initsize";
public static final String LOGGER_ASYNC_QUEUES = "logger.async.queue.size";
public static final String LOGGER_ASYNC_WORKERS = "logger.async.worker.size";
// 한 트랜잭션(COMMIT 1회)에 묶어 적재할 최대 로그 건수
public static final String LOGGER_ASYNC_BATCHSIZE = "logger.async.batch.size";
public static final String LOGGER_ASYNC_WAITSTRATEGY = "logger.async.worker.waitstrategy";
public static final String LOGGER_ASYNC_WAITSTRATEGY_BLOCK = "BLOCK";
src/main/java/com/eactive/eai/env/ElinkConfig.java
Vendored
+20
View File
@@ -37,6 +37,7 @@ public class ElinkConfig implements ConfigKeys {
private static int asyncPoolInitSize = 8;
private static int asyncQueueSize = 1024;
private static int asyncWorkers = 16;
private static int asyncBatchSize = 100;
private static String waitStrategy = LOGGER_ASYNC_WAITSTRATEGY;
@@ -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;
}