diff --git a/.idea/vcs.xml b/.idea/vcs.xml new file mode 100644 index 0000000..35eb1dd --- /dev/null +++ b/.idea/vcs.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/pom.xml b/pom.xml index 73176fc..27db912 100644 --- a/pom.xml +++ b/pom.xml @@ -113,6 +113,8 @@ lombok + 11 + 11 diff --git a/src/main/java/com/startbot/StartBotApplication.java b/src/main/java/com/startbot/StartBotApplication.java index 5f4b198..a0c5a89 100644 --- a/src/main/java/com/startbot/StartBotApplication.java +++ b/src/main/java/com/startbot/StartBotApplication.java @@ -10,5 +10,4 @@ public class StartBotApplication { public static void main(String[] args) { SpringApplication.run(StartBotApplication.class, args); } - } diff --git a/src/main/java/com/startbot/config/QueueConfig.java b/src/main/java/com/startbot/config/QueueConfig.java new file mode 100644 index 0000000..886a5e7 --- /dev/null +++ b/src/main/java/com/startbot/config/QueueConfig.java @@ -0,0 +1,17 @@ +package com.startbot.config; + +import com.startbot.entity.dto.PrivateMessageDTO; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import java.util.concurrent.ArrayBlockingQueue; + +@Configuration +public class QueueConfig { + + @Bean + public ArrayBlockingQueue privateMessageQueue() { + // 设置队列容量，根据你的需求调整 + return new ArrayBlockingQueue<>(1000); + } +} \ No newline at end of file diff --git a/src/main/java/com/startbot/config/ThreadConfig.java b/src/main/java/com/startbot/config/ThreadConfig.java new file mode 100644 index 0000000..16dabed --- /dev/null +++ b/src/main/java/com/startbot/config/ThreadConfig.java @@ -0,0 +1,25 @@ +package com.startbot.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.task.TaskExecutor; +import org.springframework.scheduling.annotation.EnableAsync; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; + +import java.util.concurrent.Executor; + +// 配置线程池 +@Configuration +@EnableAsync +public class ThreadConfig { + @Bean(name = "privateMessageTaskExecutor") + public Executor privateMessageTaskExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(1);// 核心线程数 + executor.setMaxPoolSize(2);// 最大线程数 + executor.setQueueCapacity(100);// 队列容量 + executor.setThreadNamePrefix("private-message-"); + executor.initialize(); + return executor; + } +} \ No newline at end of file diff --git a/src/main/java/com/startbot/controller/StarBotController.java b/src/main/java/com/startbot/controller/StarBotController.java index 135b642..06892f7 100644 --- a/src/main/java/com/startbot/controller/StarBotController.java +++ b/src/main/java/com/startbot/controller/StarBotController.java @@ -1,31 +1,45 @@ package com.startbot.controller; +import com.startbot.entity.dto.PrivateMessageDTO; import com.startbot.entity.dto.ReqDTO; import com.startbot.service.PrivateMessageService; +import com.startbot.utils.PrivateMessageConsumer; import lombok.RequiredArgsConstructor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.core.task.TaskExecutor; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RestController; +import java.util.concurrent.ArrayBlockingQueue; + @RestController public class StarBotController { private static final Logger logger = LoggerFactory.getLogger(StarBotController.class); private final PrivateMessageService privateMessageService; + ArrayBlockingQueue privateMessageQueue; - public StarBotController(PrivateMessageService privateMessageService) { + public StarBotController(PrivateMessageService privateMessageService, + ArrayBlockingQueue privateMessageQueue, + PrivateMessageConsumer privateMessageConsumer + ) { this.privateMessageService = privateMessageService; + this.privateMessageQueue = privateMessageQueue; + new Thread(privateMessageConsumer).start(); + } @PostMapping("/") - public void demo1(@RequestBody ReqDTO req) { + public String PrivateMessageHandler(@RequestBody ReqDTO req) { switch (req.getEvent()) { //私聊消息事件 case "10002": - privateMessageService.getMessage(req); + return privateMessageService.getMessage(req, this.privateMessageQueue); } + return null; } } diff --git a/src/main/java/com/startbot/entity/dto/PrivateMessageDTO.java b/src/main/java/com/startbot/entity/dto/PrivateMessageDTO.java index 4de23cc..3f06180 100644 --- a/src/main/java/com/startbot/entity/dto/PrivateMessageDTO.java +++ b/src/main/java/com/startbot/entity/dto/PrivateMessageDTO.java @@ -1,8 +1,5 @@ package com.startbot.entity.dto; -import com.baomidou.mybatisplus.annotation.IdType; -import com.baomidou.mybatisplus.annotation.TableId; -import com.baomidou.mybatisplus.annotation.TableName; import lombok.Data; @Data diff --git a/src/main/java/com/startbot/service/PrivateMessageService.java b/src/main/java/com/startbot/service/PrivateMessageService.java index fe57836..4032dde 100644 --- a/src/main/java/com/startbot/service/PrivateMessageService.java +++ b/src/main/java/com/startbot/service/PrivateMessageService.java @@ -2,9 +2,12 @@ package com.startbot.service; import com.baomidou.mybatisplus.extension.service.IService; import com.startbot.entity.dataobj.PrivateMessageDO; +import com.startbot.entity.dto.PrivateMessageDTO; import com.startbot.entity.dto.ReqDTO; +import java.util.concurrent.ArrayBlockingQueue; + public interface PrivateMessageService extends IService { - public void getMessage(ReqDTO reqDTO); + public String getMessage(ReqDTO reqDTO, ArrayBlockingQueue privateMessageQueue); } diff --git a/src/main/java/com/startbot/service/impl/PrivateMessageServiceImpl.java b/src/main/java/com/startbot/service/impl/PrivateMessageServiceImpl.java index f781f37..f15c8df 100644 --- a/src/main/java/com/startbot/service/impl/PrivateMessageServiceImpl.java +++ b/src/main/java/com/startbot/service/impl/PrivateMessageServiceImpl.java @@ -11,14 +11,22 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Service; +import java.util.concurrent.ArrayBlockingQueue; + @Service public class PrivateMessageServiceImpl extends ServiceImpl implements PrivateMessageService { private static final Logger logger = LoggerFactory.getLogger(PrivateMessageServiceImpl.class); @Override - public void getMessage(ReqDTO reqDTO) { - ObjectMapper jsonMapper = new ObjectMapper(); - PrivateMessageDTO message = jsonMapper.convertValue(reqDTO.getData(),PrivateMessageDTO.class); - logger.info("收到来自：{} 的消息：{}", message.getFromNickName(), message.getMessage()); + public String getMessage(ReqDTO reqDTO, ArrayBlockingQueue privateMessageQueue) { + try { + ObjectMapper jsonMapper = new ObjectMapper(); + PrivateMessageDTO message = jsonMapper.convertValue(reqDTO.getData(), PrivateMessageDTO.class); + privateMessageQueue.put(message); + logger.info("收到消息，已放入队列"); + return "success"; + } catch (InterruptedException e) { + throw new RuntimeException(e); + } } } diff --git a/src/main/java/com/startbot/utils/HttpUtils.java b/src/main/java/com/startbot/utils/HttpUtils.java new file mode 100644 index 0000000..7a05269 --- /dev/null +++ b/src/main/java/com/startbot/utils/HttpUtils.java @@ -0,0 +1,245 @@ +package com.startbot.utils; + +import java.io.IOException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.net.URLEncoder; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.stream.Collectors; + +/** + * HTTP 请求工具类，基于 Java 11 HttpClient 实现 + */ +public class HttpUtils { + + private static final HttpClient httpClient; + private static final int DEFAULT_TIMEOUT = 10; // 默认超时时间（秒） + + static { + // 初始化 HttpClient，使用单例模式 + httpClient = HttpClient.newBuilder() + .version(HttpClient.Version.HTTP_1_1) + .followRedirects(HttpClient.Redirect.NORMAL) + .connectTimeout(Duration.ofSeconds(DEFAULT_TIMEOUT)) + .build(); + } + + // 私有构造函数，防止实例化 + private HttpUtils() {} + + // ==================== GET 请求 ==================== + + /** + * 发送不带参数的 GET 请求 + * @param url 请求 URL + * @return 响应结果 + * @throws IOException 网络异常 + * @throws InterruptedException 线程中断异常 + */ + public static String get(String url) throws IOException, InterruptedException { + return get(url, Map.of()); + } + + /** + * 发送带参数的 GET 请求 + * @param url 请求 URL + * @param params 请求参数 + * @return 响应结果 + * @throws IOException 网络异常 + * @throws InterruptedException 线程中断异常 + */ + public static String get(String url, Map params) throws IOException, InterruptedException { + return get(url, params, Map.of()); + } + + /** + * 发送带参数和请求头的 GET 请求 + * @param url 请求 URL + * @param params 请求参数 + * @param headers 请求头 + * @return 响应结果 + * @throws IOException 网络异常 + * @throws InterruptedException 线程中断异常 + */ + public static String get(String url, Map params, Map headers) throws IOException, InterruptedException { + String fullUrl = buildUrlWithParams(url, params); + HttpRequest.Builder requestBuilder = HttpRequest.newBuilder() + .uri(URI.create(fullUrl)) + .GET(); + + addHeaders(requestBuilder, headers); + + HttpRequest request = requestBuilder.build(); + HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()); + return response.body(); + } + + // ==================== POST 请求 ==================== + + /** + * 发送不带请求体的 POST 请求 + * @param url 请求 URL + * @return 响应结果 + * @throws IOException 网络异常 + * @throws InterruptedException 线程中断异常 + */ + public static String post(String url) throws IOException, InterruptedException { + return post(url, Map.of()); + } + + /** + * 发送带请求头的 POST 请求 + * @param url 请求 URL + * @param headers 请求头 + * @return 响应结果 + * @throws IOException 网络异常 + * @throws InterruptedException 线程中断异常 + */ + public static String post(String url, Map headers) throws IOException, InterruptedException { + return post(url, headers, ""); + } + + /** + * 发送带 JSON 请求体的 POST 请求 + * @param url 请求 URL + * @param jsonBody JSON 请求体 + * @return 响应结果 + * @throws IOException 网络异常 + * @throws InterruptedException 线程中断异常 + */ + public static String postJson(String url, String jsonBody) throws IOException, InterruptedException { + Map headers = new HashMap<>(); + headers.put("Content-Type", "application/json"); + return post(url, headers, jsonBody); + } + + /** + * 发送带表单数据的 POST 请求 + * @param url 请求 URL + * @param params 表单参数 + * @return 响应结果 + * @throws IOException 网络异常 + * @throws InterruptedException 线程中断异常 + */ + public static String postForm(String url, Map params) throws IOException, InterruptedException { + Map headers = new HashMap<>(); + headers.put("Content-Type", "application/x-www-form-urlencoded"); + String formBody = buildFormData(params); + return post(url, headers, formBody); + } + + /** + * 发送带请求头和请求体的 POST 请求 + * @param url 请求 URL + * @param headers 请求头 + * @param requestBody 请求体 + * @return 响应结果 + * @throws IOException 网络异常 + * @throws InterruptedException 线程中断异常 + */ + public static String post(String url, Map headers, String requestBody) throws IOException, InterruptedException { + HttpRequest.Builder requestBuilder = HttpRequest.newBuilder() + .uri(URI.create(url)) + .POST(HttpRequest.BodyPublishers.ofString(requestBody)); + + addHeaders(requestBuilder, headers); + + HttpRequest request = requestBuilder.build(); + HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()); + return response.body(); + } + + // ==================== 异步请求 ==================== + + /** + * 异步发送 GET 请求 + * @param url 请求 URL + * @return 包含响应结果的 CompletableFuture + */ + public static CompletableFuture getAsync(String url) { + return getAsync(url, Map.of()); + } + + /** + * 异步发送带参数的 GET 请求 + * @param url 请求 URL + * @param params 请求参数 + * @return 包含响应结果的 CompletableFuture + */ + public static CompletableFuture getAsync(String url, Map params) { + return getAsync(url, params, Map.of()); + } + + /** + * 异步发送带参数和请求头的 GET 请求 + * @param url 请求 URL + * @param params 请求参数 + * @param headers 请求头 + * @return 包含响应结果的 CompletableFuture + */ + public static CompletableFuture getAsync(String url, Map params, Map headers) { + String fullUrl = buildUrlWithParams(url, params); + HttpRequest.Builder requestBuilder = HttpRequest.newBuilder() + .uri(URI.create(fullUrl)) + .GET(); + + addHeaders(requestBuilder, headers); + + HttpRequest request = requestBuilder.build(); + return httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString()) + .thenApply(HttpResponse::body); + } + + // ==================== 辅助方法 ==================== + + /** + * 构建带参数的 URL + */ + private static String buildUrlWithParams(String url, Map params) { + if (params == null || params.isEmpty()) { + return url; + } + + String queryString = params.entrySet().stream() + .map(entry -> encodeParam(entry.getKey()) + "=" + encodeParam(entry.getValue())) + .collect(Collectors.joining("&")); + + return url + (url.contains("?") ? "&" : "?") + queryString; + } + + /** + * 构建表单数据 + */ + private static String buildFormData(Map params) { + if (params == null || params.isEmpty()) { + return ""; + } + + return params.entrySet().stream() + .map(entry -> encodeParam(entry.getKey()) + "=" + encodeParam(entry.getValue())) + .collect(Collectors.joining("&")); + } + + /** + * 添加请求头 + */ + private static void addHeaders(HttpRequest.Builder requestBuilder, Map headers) { + headers.forEach(requestBuilder::header); + } + + /** + * 编码 URL 参数 + */ + private static String encodeParam(String param) { + return Optional.ofNullable(param) + .map(p -> URLEncoder.encode(p, StandardCharsets.UTF_8)) + .orElse(""); + } +} \ No newline at end of file diff --git a/src/main/java/com/startbot/utils/PrivateMessageConsumer.java b/src/main/java/com/startbot/utils/PrivateMessageConsumer.java new file mode 100644 index 0000000..2aa1489 --- /dev/null +++ b/src/main/java/com/startbot/utils/PrivateMessageConsumer.java @@ -0,0 +1,40 @@ +package com.startbot.utils; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.startbot.entity.dto.PrivateMessageDTO; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.scheduling.annotation.Async; +import org.springframework.stereotype.Component; + +import java.io.IOException; +import java.util.HashMap; +import java.util.concurrent.ArrayBlockingQueue; + +@Component + +public class PrivateMessageConsumer implements Runnable { + ArrayBlockingQueue privateMessageQueue; + private static final Logger logger = LoggerFactory.getLogger(PrivateMessageConsumer.class); + public PrivateMessageConsumer(ArrayBlockingQueue privateMessageQueue) { + this.privateMessageQueue = privateMessageQueue; + logger.info("私聊消息处理线程启动成功"); + } + public void getMessage() { + while (true) { + PrivateMessageDTO message; + try { + message = privateMessageQueue.take(); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + logger.info("收到来自：{} 的消息：{}", message.getFromNickName(), message.getMessage()); + } + + } + + @Override + public void run() { + this.getMessage(); + } +} diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 44b6b49..bcd695a 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -1,5 +1,5 @@ server: - port: 9999 # 应用服务端口号 + port: 19999 # 应用服务端口号 spring: application: