初始化项目

2026-08-06 15:19:45 +08:00
commit 63c8d8b757
229 changed files with 12653 additions and 0 deletions
@@ -0,0 +1,35 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.xujun</groupId>
+        <artifactId>xtools-boot-mq</artifactId>
+        <version>5.1.0</version>
+    </parent>
+    <artifactId>xtools-boot-mq-kafka</artifactId>
+
+    <!-- 依赖 -->
+    <dependencies>
+        <!-- xtools-boot begin -->
+        <!-- xtools-boot-core -->
+        <dependency>
+            <groupId>org.xujun</groupId>
+            <artifactId>xtools-boot-core</artifactId>
+        </dependency>
+        <!-- xtools-boot-mq-base -->
+        <dependency>
+            <groupId>org.xujun</groupId>
+            <artifactId>xtools-boot-mq-base</artifactId>
+        </dependency>
+        <!-- xtools-boot end -->
+
+        <!-- kafka -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-kafka</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>
@@ -0,0 +1,28 @@
+package xtools.boot.mq.kafka;
+
+import org.springframework.context.annotation.Import;
+import xtools.boot.core.utils.ModuleLoadUtils;
+import xtools.boot.mq.kafka.selector.BootKafkaImportSelector;
+
+/**
+ * <p>Title : BootKafkaConfiguration</p>
+ * <p>Description : BootKafkaConfiguration</p>
+ * <p>DevelopTools : Idea_x64_v2026.1</p>
+ * <p>DevelopSystem : macOS Sequoia 15.7.5</p>
+ * <p>Company : org.xujun</p>
+ *
+ * @author : XuJun
+ * @version : 5.0.0
+ * @date : 2026/01/01 09:30
+ */
+@Import(BootKafkaImportSelector.class)
+public class BootKafkaConfiguration {
+
+    /**
+     * 构造方法
+     */
+    public BootKafkaConfiguration() {
+        ModuleLoadUtils.loadSuccess(BootKafkaConfiguration.class);
+    }
+
+}
@@ -0,0 +1,38 @@
+package xtools.boot.mq.kafka.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * <p>Title : BootKafkaConfig</p>
+ * <p>Description : BootKafkaConfig</p>
+ * <p>DevelopTools : Idea_x64_v2026.1</p>
+ * <p>DevelopSystem : macOS Sequoia 15.7.5</p>
+ * <p>Company : org.xujun</p>
+ *
+ * @author : XuJun
+ * @version : 1.0.0
+ * @date : 2026/7/10 16:01
+ */
+@Data
+@Configuration
+@ConfigurationProperties(prefix = "xtools.kafka")
+public class BootKafkaConfig {
+
+    /**
+     * 分区数
+     */
+    private int partitions = 1;
+
+    /**
+     * 副本数
+     */
+    private int replicaCount = 1;
+
+    /**
+     * 消息保留时长
+     */
+    private String retentionMs = "604800000";
+
+}
@@ -0,0 +1,107 @@
+package xtools.boot.mq.kafka.handle;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.common.config.TopicConfig;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.config.TopicBuilder;
+import org.springframework.kafka.core.KafkaAdmin;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
+import org.springframework.kafka.listener.MessageListener;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ReflectionUtils;
+import xtools.boot.api.exection.BizError;
+import xtools.boot.mq.base.handle.BaseMqHandle;
+import xtools.boot.mq.kafka.config.BootKafkaConfig;
+
+import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
+import java.util.Objects;
+
+/**
+ * <p>Title : KafkaHandle</p>
+ * <p>Description : KafkaHandle</p>
+ * <p>DevelopTools : Idea_x64_v2026.1</p>
+ * <p>DevelopSystem : macOS Sequoia 15.7.5</p>
+ * <p>Company : org.xujun</p>
+ *
+ * @author : XuJun
+ * @version : 1.0.0
+ * @date : 2026/7/10 14:56
+ */
+@Component
+@RequiredArgsConstructor
+public class KafkaHandle implements BaseMqHandle {
+
+    /**
+     * 默认监听方法
+     */
+    private final static String DEFAULT_LISTENER_METHOD = "baseHandleMessage";
+
+    private final BootKafkaConfig bootKafkaConfig;
+
+    private final ConcurrentKafkaListenerContainerFactory<?, ?> containerFactory;
+
+    private final KafkaAdmin kafkaAdmin;
+
+    private final KafkaTemplate<String, Object> kafkaTemplate;
+
+    /**
+     * 初始化消息队列
+     *
+     * @param queueName 消息队列名称
+     */
+    @Override
+    public void initQueue(String queueName) {
+        NewTopic topic = TopicBuilder
+                // topic名称
+                .name(queueName)
+                // 分区数
+                .partitions(bootKafkaConfig.getPartitions())
+                // 副本数
+                .replicas(bootKafkaConfig.getReplicaCount())
+                // 消息保留时长
+                .config(TopicConfig.RETENTION_MS_CONFIG, bootKafkaConfig.getRetentionMs())
+                .build();
+        kafkaAdmin.createOrModifyTopics(topic);
+    }
+
+    /**
+     * 添加消息队列监听
+     *
+     * @param queueName      消息队列
+     * @param delegate       监听类
+     * @param listenerMethod 监听方法
+     * @param params         参数
+     */
+    @Override
+    public void addListen(String queueName, Object delegate, String listenerMethod, Object params) {
+        if (Objects.isNull(listenerMethod)) {
+            listenerMethod = DEFAULT_LISTENER_METHOD;
+        }
+        // 获取方法
+        Method method = ReflectionUtils.findMethod(delegate.getClass(), listenerMethod, String.class);
+        if (Objects.isNull(method)) {
+            throw new BizError("在" + delegate.getClass().getName() + "上找不到[" + listenerMethod + "]的方法");
+        }
+        ConcurrentMessageListenerContainer<?, ?> container = containerFactory.createContainer(queueName);
+        container.getContainerProperties().setMessageListener((MessageListener<Object, Object>) record -> {
+            Object value = record.value();
+            String message = value instanceof byte[] bytes ? new String(bytes, StandardCharsets.UTF_8) : String.valueOf(value);
+            ReflectionUtils.invokeMethod(method, delegate, message);
+        });
+        container.start();
+    }
+
+    /**
+     * 推送消息
+     *
+     * @param routingKey 路由 key
+     * @param data       消息数据
+     */
+    @Override
+    public void push(String routingKey, Object data) {
+        kafkaTemplate.send(routingKey, data);
+    }
+}
@@ -0,0 +1,36 @@
+package xtools.boot.mq.kafka.selector;
+
+import org.jspecify.annotations.NonNull;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.context.annotation.ClassPathBeanDefinitionScanner;
+import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
+import org.springframework.core.type.AnnotationMetadata;
+
+/**
+ * <p>Title : BootKafkaImportSelector</p>
+ * <p>Description : BootKafkaImportSelector</p>
+ * <p>DevelopTools : Idea_x64_v2026.1</p>
+ * <p>DevelopSystem : macOS Sequoia 15.7.5</p>
+ * <p>Company : org.xujun</p>
+ *
+ * @author : XuJun
+ * @version : 5.0.0
+ * @date : 2026/01/01 09:30
+ */
+public class BootKafkaImportSelector implements ImportBeanDefinitionRegistrar {
+
+    /**
+     * 根据给定的注释元数据,根据需要注册bean
+     *
+     * @param importingClassMetadata AnnotationMetadata
+     * @param registry               BeanDefinitionRegistry
+     */
+    @Override
+    public void registerBeanDefinitions(@NonNull AnnotationMetadata importingClassMetadata, @NonNull BeanDefinitionRegistry registry) {
+        // 构建扫描对象
+        ClassPathBeanDefinitionScanner scanner = new ClassPathBeanDefinitionScanner(registry, true);
+        // 扫描包下路径
+        scanner.scan("xtools.boot.mq.kafka");
+    }
+
+}
@@ -0,0 +1 @@
+xtools.boot.mq.kafka.BootKafkaConfiguration
				`@@ -0,0 +1 @@`
				`xtools.boot.mq.kafka.BootKafkaConfiguration`