tianhao/fmt
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v2.1.0 更新：

Browse Source 
1、引入metona-mq-mini-pro消息队列，重构实时日志获取方式
This commit is contained in:
紫影233
2025-07-23 18:27:16 +08:00
parent 7d8cda234e
commit bb99e48275
5 changed files with 70 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
pom.xml
View File

@@ -10,7 +10,7 @@
</parent>
<groupId>cn.somkit</groupId>
<artifactId>fmt</artifactId>
<version>2.0.1</version>
<version>2.1.0</version>
<name>fmt</name>
<description>File Manage System for by SpringBoot</description>
<properties>
@@ -51,6 +51,12 @@
<artifactId>metona-cache-spring-boot-starter</artifactId>
<version>1.0.0</version>
</dependency>
<dependency>
<groupId>cn.metona</groupId>
<artifactId>metona-mq-mini-pro</artifactId>
<version>1.0.1</version>
</dependency>
</dependencies>
<build>

13
src/main/java/cn/somkit/fmt/FmtApplication.java
View File

@@ -1,13 +1,26 @@
package cn.somkit.fmt;
import cn.metona.mq.consumer.MessageListener;
import cn.metona.mq.core.Message;
import cn.metona.mq.exception.MessageConsumeException;
import cn.metona.mq.util.MetonaMQUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class FmtApplication {
private static final Logger logger = LoggerFactory.getLogger(FmtApplication.class);
public static void main(String[] args) {
SpringApplication.run(FmtApplication.class, args);
if(!MetonaMQUtil.isInitialized()){
logger.info("Metona MQ Mini Pro 初始化...");
MetonaMQUtil.init();
}
}
}

14
src/main/java/cn/somkit/fmt/action/LoggingAction.java
View File

@@ -1,6 +1,10 @@
package cn.somkit.fmt.action;
import cn.metona.cache.Cache;
import cn.metona.mq.util.MetonaMQUtil;
import cn.somkit.fmt.socket.WebSocketServerHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
@@ -12,8 +16,10 @@ import org.springframework.web.bind.annotation.ResponseBody;
@RequestMapping("/logging")
public class LoggingAction {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private Cache<String, Object> cache;
private Cache<String, String> cache;
@GetMapping("/index")
public String index() throws Exception{
@@ -23,6 +29,10 @@ public class LoggingAction {
@ResponseBody
@PostMapping("/close")
public void close(Boolean closed) throws Exception {
cache.put("closed", closed);
cache.put("closed", String.valueOf(closed));
logger.info("Metona MQ Mini Pro 停止消费者（顺序消息）...");
if(closed){
MetonaMQUtil.stopOrderedConsuming();
}
}
}

13
src/main/java/cn/somkit/fmt/filter/LogStashFilter.java
View File

@@ -5,8 +5,10 @@ import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.filter.Filter;
import ch.qos.logback.core.spi.FilterReply;
import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONUtil;
import cn.metona.mq.exception.MessageSendException;
import cn.metona.mq.util.MetonaMQUtil;
import cn.somkit.fmt.entity.LoggerMessage;
import cn.somkit.fmt.utils.LoggerQueue;
import java.time.Instant;
import java.time.ZoneId;
@@ -27,7 +29,14 @@ public class LogStashFilter extends Filter<ILoggingEvent> {
DateUtil.format(Instant.ofEpochMilli(e.getTimeStamp()).atZone(ZoneId.systemDefault()).toLocalDateTime(),
"yyyy-MM-dd HH:mm:ss.SSS")
);
LoggerQueue.getInstance().push(msg); // 单例阻塞队列
try {
System.out.println(MetonaMQUtil.getConsumerStatus());
if(MetonaMQUtil.isInitialized() && MetonaMQUtil.isOrderedConsumerRunning()){
MetonaMQUtil.send("log-topic", "log-monitor", JSONUtil.toJsonStr(msg));
}
} catch (MessageSendException ex) {
System.out.println("发送消息队列失败");
}
return FilterReply.NEUTRAL;
}

35
src/main/java/cn/somkit/fmt/socket/WebSocketServerHandler.java
View File

@@ -1,10 +1,10 @@
package cn.somkit.fmt.socket;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import cn.metona.cache.Cache;
import cn.somkit.fmt.entity.LoggerMessage;
import cn.somkit.fmt.utils.LoggerQueue;
import cn.metona.mq.consumer.MessageListener;
import cn.metona.mq.core.Message;
import cn.metona.mq.util.MetonaMQUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -19,7 +19,7 @@ public class WebSocketServerHandler implements WebSocketHandler {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private Cache<String, Object> cache;
private Cache<String, String> cache;
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
@@ -27,10 +27,29 @@ public class WebSocketServerHandler implements WebSocketHandler {
}
private void push(WebSocketSession session) throws IOException {
while (StrUtil.isBlankIfStr(cache.get("closed")) || !Boolean.parseBoolean(String.valueOf(cache.get("closed")))) {
LoggerMessage log = LoggerQueue.getInstance().poll();
if(log != null){
session.sendMessage(new TextMessage(JSONUtil.toJsonStr(log)));
boolean closed = StrUtil.isNotBlank(cache.get("closed")) && Boolean.parseBoolean(cache.get("closed"));
if(!closed){
try {
if(MetonaMQUtil.isInitialized()){
logger.info("Metona MQ Mini Pro 订阅主题（log-topic）（顺序消息）...");
MetonaMQUtil.subscribeOrdered("log-topic", new MessageListener() {
@Override
public void onMessage(Message message) {
try {
session.sendMessage(new TextMessage(message.getBody()));
} catch (IOException e) {
throw new RuntimeException(e);
}
}
});
if(!MetonaMQUtil.isOrderedConsumerRunning()){
logger.info("Metona MQ Mini Pro 启动消费者（顺序消息）...");
MetonaMQUtil.startOrderedConsuming();
}
}
} catch (Exception e) {
logger.error("Metona MQ Mini Pro 异常", e);
}
}
}