Commit @2ac5d34bec3a53b3783af3d99e641e222b180383

장동신 2024-12-18

리포트 SQL, 리포트 서비스 쓰레드 수정

@2ac5d34bec3a53b3783af3d99e641e222b180383

266590d

2ac5d34

src/main/java/com/munjaon/server/cache/mapper/ReportMapper.java

--- src/main/java/com/munjaon/server/cache/mapper/ReportMapper.java

+++ src/main/java/com/munjaon/server/cache/mapper/ReportMapper.java


     ReportDto getReportForUser(String userId);
     List<ReportDto> getReportListForUser(String userId);
     int deleteReport(String msgId);
-    int deleteBulkReport(Map<String, String> reqMap);
+    int deleteBulkReport(Map<String, Object> reqMap);
 }

266590d

2ac5d34

src/main/java/com/munjaon/server/cache/service/ReportService.java

--- src/main/java/com/munjaon/server/cache/service/ReportService.java

+++ src/main/java/com/munjaon/server/cache/service/ReportService.java


     }
 
     public int deleteBulkReport(String msgId, String userId) {
-        Map<String, String> reqMap = new HashMap<>();
+        Map<String, Object> reqMap = new HashMap<>();
         reqMap.put("msgId", msgId);
         reqMap.put("userId", userId);
 
         return deleteBulkReport(reqMap);
     }
 
-    public int deleteBulkReport(Map<String, String> reqMap) {
+    public int deleteBulkReport(Map<String, Object> reqMap) {
         return reportMapper.deleteBulkReport(reqMap);
     }
 }

266590d

2ac5d34

src/main/java/com/munjaon/server/config/RunnerConfiguration.java

--- src/main/java/com/munjaon/server/config/RunnerConfiguration.java

+++ src/main/java/com/munjaon/server/config/RunnerConfiguration.java


     public CommandLineRunner getRunnerBeanForReportQueue() {
         try {
             String serviceName = "REPORT_QUEUE";
-            ReportQueueServerService reportQueueServerService = new ReportQueueServerService(serviceName);
-            reportQueueServerService.start();
+            ReportQueueServer reportQueueServer = new ReportQueueServer(serviceName);
+            reportQueueServer.start();
         } catch (Exception e) {
             throw new RuntimeException(e);
         }

266590d

2ac5d34

src/main/java/com/munjaon/server/server/dto/ReportUserDto.java

--- src/main/java/com/munjaon/server/server/dto/ReportUserDto.java

+++ src/main/java/com/munjaon/server/server/dto/ReportUserDto.java


     private int maxQueueCount;                //Report Queue에 유지할 건수
     /* 요청을 처리중인지 여부 */
     private boolean isRunningMode;
+    /* Report Queue 처리중인지 여부 */
+    private boolean isQueueMode;
 
     public int isAlive() {
         if (isLogin) {

2ac5d34

src/main/java/com/munjaon/server/server/service/ReportQueueServer.java (added)

+++ src/main/java/com/munjaon/server/server/service/ReportQueueServer.java

...	...	@@ -0,0 +1,77 @@
	1	+package com.munjaon.server.server.service;
	2	+
	3	+import com.munjaon.server.server.dto.ReportUserDto;
	4	+import com.munjaon.server.server.queue.ReportUserQueue;
	5	+import com.munjaon.server.server.task.ReportQueueTask;
	6	+import org.json.simple.JSONObject;
	7	+
	8	+import java.util.List;
	9	+
	10	+public class ReportQueueServer extends Service {
	11	+ private final ReportUserQueue reportUserQueue = ReportUserQueue.getInstance();
	12	+ private final int maxWriteCount;
	13	+ private final int maxQueueCount;
	14	+
	15	+ public ReportQueueServer(String serviceName) {
	16	+ super(serviceName);
	17	+ this.maxWriteCount = Integer.parseInt(getProp("MAX_WRITE_COUNT").trim());
	18	+ this.maxQueueCount = Integer.parseInt(getProp("MAX_QUEUE_COUNT").trim());
	19	+ }
	20	+
	21	+ @Override
	22	+ public void checkReady() {
	23	+ this.IS_READY_YN = true;
	24	+ }
	25	+
	26	+ @Override
	27	+ public void initResources() {
	28	+ saveSystemLog("REPORT_QUEUE_SERVICE : RESOURCES INITIALIZING ... ...");
	29	+
	30	+ saveSystemLog("REPORT_QUEUE_SERVICE : RESOURCES INITIALIZED ... ...");
	31	+ }
	32	+
	33	+ @Override
	34	+ public void releaseResources() {
	35	+ saveSystemLog("REPORT_QUEUE_SERVICE : SERVER RESOURCE RELEASING ... ...");
	36	+
	37	+ saveSystemLog("REPORT_QUEUE_SERVICE : SERVER RESOURCE RELEASED ... ...");
	38	+ }
	39	+
	40	+ @Override
	41	+ public void doService() {
	42	+ saveSystemLog("REPORT_QUEUE_SERVICE : SERVER SERVICE STARTED ... ...");
	43	+ while (isRun()) {
	44	+ try {
	45	+ doQueueService();
	46	+ Thread.sleep(10);
	47	+ } catch (Exception e) {
	48	+ throw new RuntimeException(e);
	49	+ }
	50	+ }
	51	+ saveSystemLog("REPORT_QUEUE_SERVICE : SERVER SERVICE STOPPED ... ...");
	52	+ }
	53	+
	54	+ private void doQueueService() {
	55	+ List<ReportUserDto> reportUserList = reportUserQueue.getUsers();
	56	+ if (reportUserList == null \|\| reportUserList.isEmpty()) {
	57	+ return;
	58	+ }
	59	+
	60	+ for (ReportUserDto reportUserDto : reportUserList) {
	61	+ /* Report Queue 작업중인지 체크 */
	62	+ if (reportUserDto.isQueueMode()) {
	63	+ continue;
	64	+ }
	65	+ /* Report Queue 작업을 실행한다. */
	66	+ reportUserDto.setQueueMode(true);
	67	+ reportUserDto.setMaxWriteCount(this.maxWriteCount);
	68	+ reportUserDto.setMaxQueueCount(this.maxQueueCount);
	69	+ new ReportQueueTask(reportUserDto, logger).run();
	70	+ }
	71	+ }
	72	+
	73	+ @Override
	74	+ public JSONObject monitorService() {
	75	+ return null;
	76	+ }
	77	+}

266590d

2ac5d34

src/main/java/com/munjaon/server/server/task/ReportQueueTask.java

--- src/main/java/com/munjaon/server/server/task/ReportQueueTask.java

+++ src/main/java/com/munjaon/server/server/task/ReportQueueTask.java


 import com.munjaon.server.server.dto.ReportDto;
 import com.munjaon.server.server.dto.ReportUserDto;
 import com.munjaon.server.util.LogUtil;
+import lombok.SneakyThrows;
 
 import java.text.SimpleDateFormat;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;

 public class ReportQueueTask implements Runnable {
     public static final SimpleDateFormat sdf = new SimpleDateFormat("[MM-dd HH:mm:ss]");
     public static final String LOG_DATE_FORMAT = "[MM-dd HH:mm:ss]";
+    private final int NO_REPORT_SLEEP_TIME = 3000;
+    private final int REPORT_SLEEP_TIME = 10;
 
     private final ReportUserDto reportUserDto;
+    private final ReportQueue reportQueue;
     private final LogUtil logger;
+
+    private boolean isProcessRun = false;
 
     public ReportQueueTask(ReportUserDto reportUserDto, LogUtil logger) {
         this.reportUserDto = reportUserDto;
         this.logger = logger;
+
+        /* 유효성 체크 */
+        if (validQueueTask()) {
+            this.reportQueue = reportUserDto.getReportQueue();
+            /* 리포트큐에 최대 크기까지 쓰고 모두 리포트는 전송했는지 체크 : 테스트후 적용 */
+            if (reportQueue.isWriteLimit(reportUserDto.getMaxQueueCount())) {
+                /* Report Queue : Report를 전부 전송한 경우 Truncate & 리포트 큐에 저장 로직 수행, Report Queue 저장로직 수행 */
+                if (reportQueue.isTruncateQueue(reportUserDto.getMaxQueueCount())) {
+                    try {
+                        reportQueue.truncateQueue();
+                        /* 정상적으로 Truncate가 완료된 경우만 isProcessRun : true */
+                        this.isProcessRun = true;
+                    } catch (Exception ignored) {}
+                }
+            } else {
+                /* Report Queue가 최대 쓰기 크기에 도달하지 않은 경우 : Report Queue 저장로직 수행 */
+                this.isProcessRun = true;
+            }
+        } else {
+            this.reportQueue = null;
+        }
+
+        /* Report Queue 모드로 설정 : Report Thread 제어용 */
+        reportUserDto.setQueueMode(true);   // Report Queue 모드로 설정
     }
 
-    @Override
-    public void run() {
+    private boolean validQueueTask() {
         if (reportUserDto == null || reportUserDto.getUserId() == null) {
-            return;
+            return false;
         }
         if (reportUserDto.getReportQueue() == null || reportUserDto.getReportQueue().isOpen() == false) {
-            return;
+            return false;
         }
 
-        ReportQueue reportQueue = reportUserDto.getReportQueue();
-        /* 리포트큐에 최대 크기까지 쓰고 모두 리포트는 전송했는지 체크 : 테스트후 적용 */
-        if (reportQueue.isWriteLimit(reportUserDto.getMaxQueueCount())) {
-            if (reportQueue.isTruncateQueue(reportUserDto.getMaxQueueCount())) {
-                try {
-                    reportQueue.truncateQueue();
-                } catch (Exception ignored) {}
-            } else {
-                return;
-            }
-        }
-        ReportService reportService = (ReportService) CacheService.REPORT_SERVICE.getService();
-        List<ReportDto> list = reportService.getReportListForUser(reportUserDto.getUserId());
-        if (list == null || list.isEmpty()) {
-            return;
-        }
+        return true;
+    }
 
-        StringBuilder builder = new StringBuilder();
-        for (ReportDto dto : list) {
-            try {
-                if (builder.isEmpty()) {
-                    builder.append(dto.getMsgId());
-                } else {
-                    builder.append(",").append(dto.getMsgId());
+    @SneakyThrows
+    @Override
+    public void run() {
+        /* 정상적인 Report Queue 저장로직을 수행할 경우 */
+        int cntReport = 0;
+        if (isProcessRun) {
+            ReportService reportService = (ReportService) CacheService.REPORT_SERVICE.getService();
+            List<ReportDto> list = reportService.getReportListForUser(reportUserDto.getUserId());
+
+            /* Report Queue 저장 & DB 데이터 삭제 */
+            if (list != null && list.size() > 0) {
+                List<String> msgList = new ArrayList<>();
+                /* Report Queue에 저장 */
+                for (ReportDto dto : list) {
+                    boolean isError = false;
+                    try {
+                        reportQueue.pushReportToQueue(dto);
+                        msgList.add(dto.getMsgId());
+                        cntReport++;    // Report를 처리한 카운트
+                        saveLog("reportDto : " + dto.toString());
+                    } catch (Exception e) {
+                        saveLog("ReportQueueTask ERROR");
+                        saveLog(e);
+                        isError = true;
+                    }
+                    /* 에러인 경우 */
+                    if (isError) {
+                        break;
+                    }
                 }
-                saveLog("reportDto : " + dto.toString());
-                reportQueue.pushReportToQueue(dto);
-            } catch (Exception e) {
-                throw new RuntimeException(e);
+
+                /* DB에서 데이터 삭제 */
+                if (msgList.size() > 0) {
+                    Map<String, Object> reqMap = new HashMap<>();
+                    reqMap.put("userId", reportUserDto.getUserId());
+                    reqMap.put("msgList", msgList);
+                    reportService.deleteBulkReport(reqMap);
+                }
             }
         }
 
-        Map<String, String> reqMap = new HashMap<>();
-        reqMap.put("userId", reportUserDto.getUserId());
-        reqMap.put("msgId", builder.toString());
-        reportService.deleteBulkReport(reqMap);
+        /* Thread Sleep */
+        if (cntReport > 0) {
+            saveLog("ReportQueueTask is Run ... ... [ID : " + reportUserDto.getUserId() + "] [REPORT COUNT : " + cntReport + "]");
+            Thread.sleep(REPORT_SLEEP_TIME);
+        } else {
+            Thread.sleep(NO_REPORT_SLEEP_TIME);
+        }
+
+        /* Queue 실해모드 해제 */
+        reportUserDto.setQueueMode(false);   // Report Queue 모드로 설정
     }
 
     private void saveSystemLog(Object obj) {

266590d

2ac5d34

src/main/resources/sqlmap/report_sql.xml

--- src/main/resources/sqlmap/report_sql.xml

+++ src/main/resources/sqlmap/report_sql.xml


     </delete>
 
     <delete id="deleteBulkReport">
+        DELETE FROM mj_msg_report
+        WHERE USER_ID = #{userId}
+            AND MSG_ID IN
+        <foreach item="item" index="index" collection="msgList"
+                 open="(" separator="," close=")">
+            #{item}
+        </foreach>
+    </delete>
+
+    <delete id="deleteBulkReport_bak">
         DELETE SRC FROM mj_msg_report SRC
         INNER JOIN (
             with recursive
             T as ( select #{msgId} as items),
-            N as ( select 1 as n union select n + 1 from N, T
-                where n <![CDATA[<=]]> length(items) - length(replace(items, ',', '')))
-            select distinct substring_index(substring_index(items, ',', n), ',', -1)
-            MSG_ID from N, T
+        N as ( select 1 as n union select n + 1 from N, T
+        where n <![CDATA[<=]]> length(items) - length(replace(items, ',', '')))
+        select distinct substring_index(substring_index(items, ',', n), ',', -1)
+        MSG_ID from N, T
         ) DEST ON SRC.MSG_ID = DEST.MSG_ID
         WHERE SRC.USER_ID = #{userId}
     </delete>

Add a comment

Open 0
Closed 0

List

...	...	@@ -270,8 +270,8 @@
270	270	public CommandLineRunner getRunnerBeanForReportQueue() {
271	271	try {
272	272	String serviceName = "REPORT_QUEUE";
273		- ReportQueueServerService reportQueueServerService = new ReportQueueServerService(serviceName);
274		- reportQueueServerService.start();
	273	+ ReportQueueServer reportQueueServer = new ReportQueueServer(serviceName);
	274	+ reportQueueServer.start();
275	275	} catch (Exception e) {
276	276	throw new RuntimeException(e);
277	277	}

...	...	@@ -11,5 +11,5 @@
11	11	ReportDto getReportForUser(String userId);
12	12	List<ReportDto> getReportListForUser(String userId);
13	13	int deleteReport(String msgId);
14		- int deleteBulkReport(Map<String, String> reqMap);
	14	+ int deleteBulkReport(Map<String, Object> reqMap);
15	15	}

...	...	@@ -29,14 +29,14 @@
29	29	}
30	30
31	31	public int deleteBulkReport(String msgId, String userId) {
32		- Map<String, String> reqMap = new HashMap<>();
	32	+ Map<String, Object> reqMap = new HashMap<>();
33	33	reqMap.put("msgId", msgId);
34	34	reqMap.put("userId", userId);
35	35
36	36	return deleteBulkReport(reqMap);
37	37	}
38	38
39		- public int deleteBulkReport(Map<String, String> reqMap) {
	39	+ public int deleteBulkReport(Map<String, Object> reqMap) {
40	40	return reportMapper.deleteBulkReport(reqMap);
41	41	}
42	42	}

...	...	@@ -25,6 +25,8 @@
25	25	private int maxQueueCount; //Report Queue에 유지할 건수
26	26	/* 요청을 처리중인지 여부 */
27	27	private boolean isRunningMode;
	28	+ /* Report Queue 처리중인지 여부 */
	29	+ private boolean isQueueMode;
28	30
29	31	public int isAlive() {
30	32	if (isLogin) {

...	...	@@ -6,10 +6,12 @@
6	6	import com.munjaon.server.server.dto.ReportDto;
7	7	import com.munjaon.server.server.dto.ReportUserDto;
8	8	import com.munjaon.server.util.LogUtil;
	9	+import lombok.SneakyThrows;
9	10
10	11	import java.text.SimpleDateFormat;
11	12	import java.time.LocalDateTime;
12	13	import java.time.format.DateTimeFormatter;
	14	+import java.util.ArrayList;
13	15	import java.util.HashMap;
14	16	import java.util.List;
15	17	import java.util.Map;
...	...	@@ -17,60 +19,106 @@
17	19	public class ReportQueueTask implements Runnable {
18	20	public static final SimpleDateFormat sdf = new SimpleDateFormat("[MM-dd HH:mm:ss]");
19	21	public static final String LOG_DATE_FORMAT = "[MM-dd HH:mm:ss]";
	22	+ private final int NO_REPORT_SLEEP_TIME = 3000;
	23	+ private final int REPORT_SLEEP_TIME = 10;
20	24
21	25	private final ReportUserDto reportUserDto;
	26	+ private final ReportQueue reportQueue;
22	27	private final LogUtil logger;
	28	+
	29	+ private boolean isProcessRun = false;
23	30
24	31	public ReportQueueTask(ReportUserDto reportUserDto, LogUtil logger) {
25	32	this.reportUserDto = reportUserDto;
26	33	this.logger = logger;
	34	+
	35	+ /* 유효성 체크 */
	36	+ if (validQueueTask()) {
	37	+ this.reportQueue = reportUserDto.getReportQueue();
	38	+ /* 리포트큐에 최대 크기까지 쓰고 모두 리포트는 전송했는지 체크 : 테스트후 적용 */
	39	+ if (reportQueue.isWriteLimit(reportUserDto.getMaxQueueCount())) {
	40	+ /* Report Queue : Report를 전부 전송한 경우 Truncate & 리포트 큐에 저장 로직 수행, Report Queue 저장로직 수행 */
	41	+ if (reportQueue.isTruncateQueue(reportUserDto.getMaxQueueCount())) {
	42	+ try {
	43	+ reportQueue.truncateQueue();
	44	+ /* 정상적으로 Truncate가 완료된 경우만 isProcessRun : true */
	45	+ this.isProcessRun = true;
	46	+ } catch (Exception ignored) {}
	47	+ }
	48	+ } else {
	49	+ /* Report Queue가 최대 쓰기 크기에 도달하지 않은 경우 : Report Queue 저장로직 수행 */
	50	+ this.isProcessRun = true;
	51	+ }
	52	+ } else {
	53	+ this.reportQueue = null;
	54	+ }
	55	+
	56	+ /* Report Queue 모드로 설정 : Report Thread 제어용 */
	57	+ reportUserDto.setQueueMode(true); // Report Queue 모드로 설정
27	58	}
28	59
29		- @Override
30		- public void run() {
	60	+ private boolean validQueueTask() {
31	61	if (reportUserDto == null \|\| reportUserDto.getUserId() == null) {
32		- return;
	62	+ return false;
33	63	}
34	64	if (reportUserDto.getReportQueue() == null \|\| reportUserDto.getReportQueue().isOpen() == false) {
35		- return;
	65	+ return false;
36	66	}
37	67
38		- ReportQueue reportQueue = reportUserDto.getReportQueue();
39		- /* 리포트큐에 최대 크기까지 쓰고 모두 리포트는 전송했는지 체크 : 테스트후 적용 */
40		- if (reportQueue.isWriteLimit(reportUserDto.getMaxQueueCount())) {
41		- if (reportQueue.isTruncateQueue(reportUserDto.getMaxQueueCount())) {
42		- try {
43		- reportQueue.truncateQueue();
44		- } catch (Exception ignored) {}
45		- } else {
46		- return;
47		- }
48		- }
49		- ReportService reportService = (ReportService) CacheService.REPORT_SERVICE.getService();
50		- List<ReportDto> list = reportService.getReportListForUser(reportUserDto.getUserId());
51		- if (list == null \|\| list.isEmpty()) {
52		- return;
53		- }
	68	+ return true;
	69	+ }
54	70
55		- StringBuilder builder = new StringBuilder();
56		- for (ReportDto dto : list) {
57		- try {
58		- if (builder.isEmpty()) {
59		- builder.append(dto.getMsgId());
60		- } else {
61		- builder.append(",").append(dto.getMsgId());
	71	+ @SneakyThrows
	72	+ @Override
	73	+ public void run() {
	74	+ /* 정상적인 Report Queue 저장로직을 수행할 경우 */
	75	+ int cntReport = 0;
	76	+ if (isProcessRun) {
	77	+ ReportService reportService = (ReportService) CacheService.REPORT_SERVICE.getService();
	78	+ List<ReportDto> list = reportService.getReportListForUser(reportUserDto.getUserId());
	79	+
	80	+ /* Report Queue 저장 & DB 데이터 삭제 */
	81	+ if (list != null && list.size() > 0) {
	82	+ List<String> msgList = new ArrayList<>();
	83	+ /* Report Queue에 저장 */
	84	+ for (ReportDto dto : list) {
	85	+ boolean isError = false;
	86	+ try {
	87	+ reportQueue.pushReportToQueue(dto);
	88	+ msgList.add(dto.getMsgId());
	89	+ cntReport++; // Report를 처리한 카운트
	90	+ saveLog("reportDto : " + dto.toString());
	91	+ } catch (Exception e) {
	92	+ saveLog("ReportQueueTask ERROR");
	93	+ saveLog(e);
	94	+ isError = true;
	95	+ }
	96	+ /* 에러인 경우 */
	97	+ if (isError) {
	98	+ break;
	99	+ }
62	100	}
63		- saveLog("reportDto : " + dto.toString());
64		- reportQueue.pushReportToQueue(dto);
65		- } catch (Exception e) {
66		- throw new RuntimeException(e);
	101	+
	102	+ /* DB에서 데이터 삭제 */
	103	+ if (msgList.size() > 0) {
	104	+ Map<String, Object> reqMap = new HashMap<>();
	105	+ reqMap.put("userId", reportUserDto.getUserId());
	106	+ reqMap.put("msgList", msgList);
	107	+ reportService.deleteBulkReport(reqMap);
	108	+ }
67	109	}
68	110	}
69	111
70		- Map<String, String> reqMap = new HashMap<>();
71		- reqMap.put("userId", reportUserDto.getUserId());
72		- reqMap.put("msgId", builder.toString());
73		- reportService.deleteBulkReport(reqMap);
	112	+ /* Thread Sleep */
	113	+ if (cntReport > 0) {
	114	+ saveLog("ReportQueueTask is Run ... ... [ID : " + reportUserDto.getUserId() + "] [REPORT COUNT : " + cntReport + "]");
	115	+ Thread.sleep(REPORT_SLEEP_TIME);
	116	+ } else {
	117	+ Thread.sleep(NO_REPORT_SLEEP_TIME);
	118	+ }
	119	+
	120	+ /* Queue 실해모드 해제 */
	121	+ reportUserDto.setQueueMode(false); // Report Queue 모드로 설정
74	122	}
75	123
76	124	private void saveSystemLog(Object obj) {

...	...	@@ -38,14 +38,24 @@
38	38	</delete>
39	39
40	40	<delete id="deleteBulkReport">
	41	+ DELETE FROM mj_msg_report
	42	+ WHERE USER_ID = #{userId}
	43	+ AND MSG_ID IN
	44	+ <foreach item="item" index="index" collection="msgList"
	45	+ open="(" separator="," close=")">
	46	+ #{item}
	47	+ </foreach>
	48	+ </delete>
	49	+
	50	+ <delete id="deleteBulkReport_bak">
41	51	DELETE SRC FROM mj_msg_report SRC
42	52	INNER JOIN (
43	53	with recursive
44	54	T as ( select #{msgId} as items),
45		- N as ( select 1 as n union select n + 1 from N, T
46		- where n <![CDATA[<=]]> length(items) - length(replace(items, ',', '')))
47		- select distinct substring_index(substring_index(items, ',', n), ',', -1)
48		- MSG_ID from N, T
	55	+ N as ( select 1 as n union select n + 1 from N, T
	56	+ where n <![CDATA[<=]]> length(items) - length(replace(items, ',', '')))
	57	+ select distinct substring_index(substring_index(items, ',', n), ',', -1)
	58	+ MSG_ID from N, T
49	59	) DEST ON SRC.MSG_ID = DEST.MSG_ID
50	60	WHERE SRC.USER_ID = #{userId}
51	61	</delete>

Delete comment