From 222a735fd6203b0ab0185d20cb11dfc2aee1180b Mon Sep 17 00:00:00 2001
From: jack ning <github@bytedesk.com>
Date: Sat, 17 May 2025 09:07:59 +0800
Subject: [PATCH] update modules/kbase: add        5 mod        7 files

---
 modules/core/pom.xml                          |   6 -
 .../core/jms/JmsArtemisConstants.java         |   2 +
 .../bytedesk/kbase/config/ArtemisConfig.java  |  79 +++++++++++
 .../bytedesk/kbase/faq/FaqEventListener.java  |  48 ++-----
 .../bytedesk/kbase/faq/FaqIndexConsumer.java  | 134 ++++++++++++++++++
 .../bytedesk/kbase/faq/FaqIndexMessage.java   |  54 +++++++
 .../bytedesk/kbase/faq/FaqMessageService.java |   0
 .../bytedesk/kbase/faq/FaqRestService.java    |  17 ++-
 .../kbase/faq/FaqMessageQueueTests.java       |  71 ++++++++++
 modules/pom.xml                               |  10 +-
 10 files changed, 372 insertions(+), 49 deletions(-)
 create mode 100644 modules/kbase/src/main/java/com/bytedesk/kbase/config/ArtemisConfig.java
 create mode 100644 modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqIndexConsumer.java
 create mode 100644 modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqIndexMessage.java
 create mode 100644 modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqMessageService.java
 create mode 100644 modules/kbase/src/test/java/com/bytedesk/kbase/faq/FaqMessageQueueTests.java

diff --git a/modules/core/pom.xml b/modules/core/pom.xml
index 2f296f7f92..089dc4c296 100644
--- a/modules/core/pom.xml
+++ b/modules/core/pom.xml
@@ -45,12 +45,6 @@
 			<scope>provided</scope>
 		</dependency>
 
-		<dependency>
-			<groupId>org.springframework.boot</groupId>
-			<artifactId>spring-boot-starter-artemis</artifactId>
-			<scope>provided</scope>
-		</dependency>
-
 		<!-- spring statemachine - 迁移到 cola statemachine -->
 		<!-- https://github.com/spring-projects/spring-statemachine -->
 		<!--
diff --git a/modules/core/src/main/java/com/bytedesk/core/jms/JmsArtemisConstants.java b/modules/core/src/main/java/com/bytedesk/core/jms/JmsArtemisConstants.java
index ff0549ca8e..3898bff83e 100644
--- a/modules/core/src/main/java/com/bytedesk/core/jms/JmsArtemisConstants.java
+++ b/modules/core/src/main/java/com/bytedesk/core/jms/JmsArtemisConstants.java
@@ -33,4 +33,6 @@ public class JmsArtemisConstants {
 
     public static final String TOPIC_TEST_NAME = TOPIC_PREFIX + "test";
 
+    // FAQ索引队列
+    public static final String QUEUE_FAQ_INDEX = QUEUE_PREFIX + "faq.index";
 }
diff --git a/modules/kbase/src/main/java/com/bytedesk/kbase/config/ArtemisConfig.java b/modules/kbase/src/main/java/com/bytedesk/kbase/config/ArtemisConfig.java
new file mode 100644
index 0000000000..f4f2cb3060
--- /dev/null
+++ b/modules/kbase/src/main/java/com/bytedesk/kbase/config/ArtemisConfig.java
@@ -0,0 +1,79 @@
+/*
+ * @Author: jackning 270580156@qq.com
+ * @Date: 2025-05-17 10:15:00
+ * @LastEditors: jackning 270580156@qq.com
+ * @LastEditTime: 2025-05-17 10:15:00
+ * @Description: bytedesk.com https://github.com/Bytedesk/bytedesk
+ *   Please be aware of the BSL license restrictions before installing Bytedesk IM – 
+ *  selling, reselling, or hosting Bytedesk IM as a service is a breach of the terms and automatically terminates your rights under the license.
+ *  Business Source License 1.1: https://github.com/Bytedesk/bytedesk/blob/main/LICENSE 
+ *  contact: 270580156@qq.com 
+ *  联系：270580156@qq.com
+ * Copyright (c) 2024 by bytedesk.com, All Rights Reserved. 
+ */
+package com.bytedesk.kbase.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jms.annotation.EnableJms;
+import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
+import org.springframework.jms.config.JmsListenerContainerFactory;
+import org.springframework.jms.core.JmsTemplate;
+import org.springframework.jms.support.converter.MappingJackson2MessageConverter;
+import org.springframework.jms.support.converter.MessageConverter;
+import org.springframework.jms.support.converter.MessageType;
+
+import jakarta.jms.ConnectionFactory;
+
+/**
+ * Artemis配置类
+ * 用于配置JMS相关的Bean
+ */
+@Configuration
+@EnableJms
+public class ArtemisConfig {
+
+    /**
+     * 配置消息转换器，使用Jackson将Java对象转换为JSON格式
+     */
+    @Bean
+    public MessageConverter jacksonJmsMessageConverter() {
+        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
+        converter.setTargetType(MessageType.TEXT);
+        converter.setTypeIdPropertyName("_type");
+        return converter;
+    }
+    
+    /**
+     * 配置JMS监听器容器工厂
+     */
+    @Bean
+    public JmsListenerContainerFactory<?> faqJmsListenerContainerFactory(ConnectionFactory connectionFactory) {
+        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
+        factory.setConnectionFactory(connectionFactory);
+        factory.setMessageConverter(jacksonJmsMessageConverter());
+        // 配置并发消费者数量，动态调整
+        factory.setConcurrency("3-10");
+        // 使用事务管理
+        factory.setSessionTransacted(true);
+        // 设置消息确认模式
+        factory.setSessionAcknowledgeMode(javax.jms.Session.CLIENT_ACKNOWLEDGE);
+        // 设置恢复间隔，默认5000ms
+        factory.setRecoveryInterval(5000L);
+        return factory;
+    }
+    
+    /**
+     * 配置JMS模板
+     */
+    @Bean
+    public JmsTemplate faqJmsTemplate(ConnectionFactory connectionFactory) {
+        JmsTemplate template = new JmsTemplate(connectionFactory);
+        template.setMessageConverter(jacksonJmsMessageConverter());
+        // 设置消息确认模式为客户端确认
+        template.setSessionAcknowledgeMode(javax.jms.Session.CLIENT_ACKNOWLEDGE);
+        // 设置默认目标名称
+        template.setDefaultDestinationName("bytedesk.faq.index");
+        return template;
+    }
+}
diff --git a/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqEventListener.java b/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqEventListener.java
index 69de7814ad..5a03f57b9a 100644
--- a/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqEventListener.java
+++ b/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqEventListener.java
@@ -2,7 +2,7 @@
  * @Author: jackning 270580156@qq.com
  * @Date: 2024-09-07 15:42:23
  * @LastEditors: jackning 270580156@qq.com
- * @LastEditTime: 2025-05-14 14:27:14
+ * @LastEditTime: 2025-05-17 10:20:00
  * @Description: bytedesk.com https://github.com/Bytedesk/bytedesk
  *   Please be aware of the BSL license restrictions before installing Bytedesk IM – 
  *  selling, reselling, or hosting Bytedesk IM as a service is a breach of the terms and automatically terminates your rights under the license.
@@ -43,6 +43,8 @@ public class FaqEventListener {
     private final FaqRestService faqRestService;
 
     private final UploadRestService uploadRestService;
+    
+    private final FaqMessageService faqMessageService;
 
     @EventListener
     public void onUploadCreateEvent(UploadCreateEvent event) {
@@ -77,41 +79,24 @@ public class FaqEventListener {
         }
     }
 
-    // Faq仅用于全文搜索
+    // Faq仅用于全文搜索，通过消息队列异步处理
     @EventListener
     public void onFaqCreateEvent(FaqCreateEvent event) {
         FaqEntity faq = event.getFaq();
         log.info("FaqEventListener onFaqCreateEvent: {}", faq.getQuestion());
         
-        // 全文索引
-        faqService.indexFaq(faq);
-        
-        // 向量索引
-        try {
-            faqVectorService.indexFaqVector(faq);
-        } catch (Exception e) {
-            log.error("FAQ向量索引创建失败: {}, 错误: {}", faq.getQuestion(), e.getMessage());
-        }
+        // 使用消息队列异步处理索引，避免乐观锁冲突
+        faqMessageService.sendToIndexQueue(faq.getUid());
     }
 
-    // Faq用于全文搜索和向量搜索
+    // Faq用于全文搜索和向量搜索，通过消息队列异步处理
     @EventListener
     public void onFaqUpdateDocEvent(FaqUpdateDocEvent event) {
         FaqEntity faq = event.getFaq();
         log.info("FaqEventListener FaqUpdateDocEvent: {}", faq.getQuestion());
         
-        // 更新全文索引
-        faqService.indexFaq(faq);
-        
-        // 更新向量索引
-        try {
-            // 先删除旧的向量索引
-            faqVectorService.deleteFaqVector(faq);
-            // 再创建新的向量索引
-            faqVectorService.indexFaqVector(faq);
-        } catch (Exception e) {
-            log.error("FAQ向量索引更新失败: {}, 错误: {}", faq.getQuestion(), e.getMessage());
-        }
+        // 使用消息队列异步处理索引更新
+        faqMessageService.sendToIndexQueue(faq.getUid());
     }
 
     @EventListener
@@ -119,18 +104,7 @@ public class FaqEventListener {
         FaqEntity faq = event.getFaq();
         log.info("FaqEventListener onFaqDeleteEvent: {}", faq.getQuestion());
         
-        // 从全文索引中删除
-        boolean deleted = faqService.deleteFaq(faq.getUid());
-        if (!deleted) {
-            log.warn("从Elasticsearch中删除FAQ全文索引失败: {}", faq.getUid());
-            // 可以考虑添加重试逻辑或者其他错误处理
-        }
-        
-        // 从向量索引中删除
-        boolean vectorDeleted = faqVectorService.deleteFaqVector(faq);
-        if (!vectorDeleted) {
-            log.warn("从向量存储中删除FAQ索引失败: {}", faq.getUid());
-            // 可以考虑添加重试逻辑或者其他错误处理
-        }
+        // 删除操作通过消息队列异步处理
+        faqMessageService.sendToDeleteQueue(faq.getUid());
     }
 }
diff --git a/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqIndexConsumer.java b/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqIndexConsumer.java
new file mode 100644
index 0000000000..fc3c7ffd08
--- /dev/null
+++ b/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqIndexConsumer.java
@@ -0,0 +1,134 @@
+/*
+ * @Author: jackning 270580156@qq.com
+ * @Date: 2025-05-17 10:10:00
+ * @LastEditors: jackning 270580156@qq.com
+ * @LastEditTime: 2025-05-17 10:10:00
+ * @Description: bytedesk.com https://github.com/Bytedesk/bytedesk
+ *   Please be aware of the BSL license restrictions before installing Bytedesk IM – 
+ *  selling, reselling, or hosting Bytedesk IM as a service is a breach of the terms and automatically terminates your rights under the license.
+ *  Business Source License 1.1: https://github.com/Bytedesk/bytedesk/blob/main/LICENSE 
+ *  contact: 270580156@qq.com 
+ *  联系：270580156@qq.com
+ * Copyright (c) 2024 by bytedesk.com, All Rights Reserved. 
+ */
+package com.bytedesk.kbase.faq;
+
+import java.util.Optional;
+import java.util.Random;
+
+import org.springframework.jms.annotation.JmsListener;
+import org.springframework.stereotype.Component;
+
+import com.bytedesk.core.jms.JmsArtemisConstants;
+
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * FAQ索引消费者
+ * 用于处理FAQ索引队列中的消息
+ */
+@Slf4j
+@Component
+@AllArgsConstructor
+public class FaqIndexConsumer {
+
+    private final FaqElasticService faqElasticService;
+    private final FaqVectorService faqVectorService;
+    private final FaqRestService faqRestService;
+    private final Random random = new Random();
+
+    /**
+     * 处理FAQ索引队列中的消息
+     * 
+     * @param message FAQ索引消息
+     */
+    @JmsListener(destination = "bytedesk.faq.index", containerFactory = "jmsArtemisQueueFactory", concurrency = "3-10")
+    public void processIndexMessage(FaqIndexMessage message) {
+        try {
+            log.debug("接收到FAQ索引请求: {}", message.getFaqUid());
+            
+            // 引入随机延迟，避免并发冲突
+            int delay = 50 + random.nextInt(200); // 50-250ms随机延迟
+            Thread.sleep(delay);
+            
+            // 获取FAQ实体
+            Optional<FaqEntity> optionalFaq = faqRestService.findByUid(message.getFaqUid());
+            if (!optionalFaq.isPresent()) {
+                log.warn("无法找到要索引的FAQ: {}", message.getFaqUid());
+                return;
+            }
+            
+            FaqEntity faq = optionalFaq.get();
+            
+            // 根据操作类型执行相应的操作
+            if ("delete".equals(message.getOperationType())) {
+                handleDeleteOperation(faq, message);
+            } else {
+                handleIndexOperation(faq, message);
+            }
+            
+        } catch (Exception e) {
+            log.error("处理FAQ索引消息时出错: {}", e.getMessage(), e);
+        }
+    }
+    
+    /**
+     * 处理索引操作
+     */
+    private void handleIndexOperation(FaqEntity faq, FaqIndexMessage message) {
+        // 执行全文索引
+        if (message.isUpdateElasticIndex()) {
+            try {
+                log.debug("为FAQ创建全文索引: {}", faq.getUid());
+                faqElasticService.indexFaq(faq);
+            } catch (Exception e) {
+                log.error("FAQ全文索引创建失败: {}, 错误: {}", faq.getUid(), e.getMessage(), e);
+            }
+        }
+        
+        // 执行向量索引
+        if (message.isUpdateVectorIndex()) {
+            try {
+                log.debug("为FAQ创建向量索引: {}", faq.getUid());
+                // 先尝试删除旧的向量索引
+                faqVectorService.deleteFaqVector(faq);
+                // 创建新的向量索引
+                faqVectorService.indexFaqVector(faq);
+            } catch (Exception e) {
+                log.error("FAQ向量索引创建失败: {}, 错误: {}", faq.getUid(), e.getMessage(), e);
+            }
+        }
+    }
+    
+    /**
+     * 处理删除操作
+     */
+    private void handleDeleteOperation(FaqEntity faq, FaqIndexMessage message) {
+        // 从全文索引中删除
+        if (message.isUpdateElasticIndex()) {
+            try {
+                log.debug("从全文索引中删除FAQ: {}", faq.getUid());
+                boolean deleted = faqElasticService.deleteFaq(faq.getUid());
+                if (!deleted) {
+                    log.warn("从Elasticsearch中删除FAQ全文索引失败: {}", faq.getUid());
+                }
+            } catch (Exception e) {
+                log.error("从全文索引中删除FAQ失败: {}, 错误: {}", faq.getUid(), e.getMessage(), e);
+            }
+        }
+        
+        // 从向量索引中删除
+        if (message.isUpdateVectorIndex()) {
+            try {
+                log.debug("从向量索引中删除FAQ: {}", faq.getUid());
+                boolean vectorDeleted = faqVectorService.deleteFaqVector(faq);
+                if (!vectorDeleted) {
+                    log.warn("从向量存储中删除FAQ索引失败: {}", faq.getUid());
+                }
+            } catch (Exception e) {
+                log.error("从向量索引中删除FAQ失败: {}, 错误: {}", faq.getUid(), e.getMessage(), e);
+            }
+        }
+    }
+}
diff --git a/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqIndexMessage.java b/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqIndexMessage.java
new file mode 100644
index 0000000000..467c7e1a26
--- /dev/null
+++ b/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqIndexMessage.java
@@ -0,0 +1,54 @@
+/*
+ * @Author: jackning 270580156@qq.com
+ * @Date: 2025-05-17 10:00:00
+ * @LastEditors: jackning 270580156@qq.com
+ * @LastEditTime: 2025-05-17 10:00:00
+ * @Description: bytedesk.com https://github.com/Bytedesk/bytedesk
+ *   Please be aware of the BSL license restrictions before installing Bytedesk IM – 
+ *  selling, reselling, or hosting Bytedesk IM as a service is a breach of the terms and automatically terminates your rights under the license.
+ *  Business Source License 1.1: https://github.com/Bytedesk/bytedesk/blob/main/LICENSE 
+ *  contact: 270580156@qq.com 
+ *  联系：270580156@qq.com
+ * Copyright (c) 2024 by bytedesk.com, All Rights Reserved. 
+ */
+package com.bytedesk.kbase.faq;
+
+import java.io.Serializable;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * FAQ索引消息
+ * 用于在Artemis队列中传递需要索引的FAQ信息
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class FaqIndexMessage implements Serializable {
+    
+    private static final long serialVersionUID = 1L;
+    
+    /**
+     * FAQ唯一标识
+     */
+    private String faqUid;
+    
+    /**
+     * 操作类型：create, update, delete
+     */
+    private String operationType;
+    
+    /**
+     * 是否需要更新全文索引
+     */
+    private boolean updateElasticIndex;
+    
+    /**
+     * 是否需要更新向量索引
+     */
+    private boolean updateVectorIndex;
+}
diff --git a/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqMessageService.java b/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqMessageService.java
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqRestService.java b/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqRestService.java
index c91500295a..1201c3d46b 100644
--- a/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqRestService.java
+++ b/modules/kbase/src/main/java/com/bytedesk/kbase/faq/FaqRestService.java
@@ -331,6 +331,14 @@ public class FaqRestService extends BaseRestServiceWithExcel<FaqEntity, FaqReque
             FaqEntity entity) {
         try {
             log.warn("处理乐观锁冲突: {}", entity.getUid());
+            
+            // 引入随机延迟，避免并发冲突
+            try {
+                Thread.sleep(100 + (long)(Math.random() * 400)); // 100-500ms随机延迟
+            } catch (InterruptedException ie) {
+                Thread.currentThread().interrupt();
+            }
+            
             // 重新获取最新版本的实体
             Optional<FaqEntity> latest = faqRepository.findByUid(entity.getUid());
             if (latest.isPresent()) {
@@ -359,7 +367,14 @@ public class FaqRestService extends BaseRestServiceWithExcel<FaqEntity, FaqReque
                 latestEntity.setDocIdList(entity.getDocIdList());
                 latestEntity.setStatus(entity.getStatus());
 
-                return faqRepository.save(latestEntity);
+                // 尝试保存，可能会再次失败
+                try {
+                    return faqRepository.save(latestEntity);
+                } catch (ObjectOptimisticLockingFailureException retryException) {
+                    // 如果再次失败，记录日志并抛出异常
+                    log.error("重试处理乐观锁冲突仍然失败: {}", entity.getUid());
+                    throw retryException;
+                }
             }
         } catch (Exception ex) {
             log.error("无法处理乐观锁冲突: {}", ex.getMessage(), ex);
diff --git a/modules/kbase/src/test/java/com/bytedesk/kbase/faq/FaqMessageQueueTests.java b/modules/kbase/src/test/java/com/bytedesk/kbase/faq/FaqMessageQueueTests.java
new file mode 100644
index 0000000000..0a74fd0034
--- /dev/null
+++ b/modules/kbase/src/test/java/com/bytedesk/kbase/faq/FaqMessageQueueTests.java
@@ -0,0 +1,71 @@
+package com.bytedesk.kbase.faq;
+
+import java.util.List;
+import java.util.Optional;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * FAQ消息队列测试类
+ */
+@Slf4j
+@SpringBootTest
+public class FaqMessageQueueTests {
+    
+    @Autowired
+    private FaqMessageService faqMessageService;
+    
+    @Autowired
+    private FaqRepository faqRepository;
+    
+    /**
+     * 测试发送单个FAQ到索引队列
+     */
+    @Test
+    public void testSendSingleFaqToQueue() {
+        // 查找第一个FAQ记录
+        Optional<FaqEntity> firstFaq = faqRepository.findAll().stream().findFirst();
+        if (firstFaq.isPresent()) {
+            String faqUid = firstFaq.get().getUid();
+            log.info("发送FAQ到索引队列: {}", faqUid);
+            faqMessageService.sendToIndexQueue(faqUid);
+            // 等待一段时间，让消息处理完成
+            try {
+                Thread.sleep(2000);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+            log.info("消息发送完成");
+        } else {
+            log.warn("没有找到FAQ记录");
+        }
+    }
+    
+    /**
+     * 测试批量发送FAQ到索引队列
+     */
+    @Test
+    public void testBatchSendFaqToQueue() {
+        // 查找前10个FAQ记录
+        List<FaqEntity> faqs = faqRepository.findAll().stream().limit(10).toList();
+        if (!faqs.isEmpty()) {
+            log.info("批量发送{}个FAQ到索引队列", faqs.size());
+            faqMessageService.batchSendToIndexQueue(
+                faqs.stream().map(FaqEntity::getUid).toList()
+            );
+            // 等待一段时间，让消息处理完成
+            try {
+                Thread.sleep(5000);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+            log.info("批量消息发送完成");
+        } else {
+            log.warn("没有找到FAQ记录");
+        }
+    }
+}
diff --git a/modules/pom.xml b/modules/pom.xml
index eeb4e8ee2f..de78c3a680 100644
--- a/modules/pom.xml
+++ b/modules/pom.xml
@@ -86,11 +86,11 @@
             <scope>provided</scope>
         </dependency>
 
-        <!-- <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-data-rest</artifactId>
-            <scope>provided</scope>
-        </dependency> -->
+        <dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-artemis</artifactId>
+			<scope>provided</scope>
+		</dependency>
 
         <dependency>
             <groupId>org.springframework.boot</groupId>