ai对话—多线程并发处理问题
先简单回顾一下旧版本的对话接口如何实现
其实这里更多是涉及到多线程工作上的学习问题
在初代版本中 我自己以为的搞了一个线程池就能完成多线程的任务了
代码语言：text复制Java
public ThreadPoolExecutor pool=new ThreadPoolExecutor(13,13,8,
TimeUnit.MINUTES,new ArrayBlockingQueue<>(4),
Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());
@RequestMapping("/get")
public Result get(@RequestParam(“question”) String question,@RequestParam(“id”) int id) throws Exception {
Future f1 = pool.submit(new callable(question,id));
String answer =f1.get();
return Result.ok(answer);
}工作的代码就不放了 这里是重点 重点是我的接口中 每一次任务的执行都是new一个新的线程出来 去执行任务 但并没有主动的写关闭线程的语句 这就导致了 线程很容易堆满 每次执行完应该释放一个线程 而且这里并没有加多对异常的处理 如果对端那边的ai卡住了 就没有办法得知发生了什么事情
于是这里就有了下面的重写了的语句
代码语言：text复制Java
public ThreadPoolExecutor pool = new ThreadPoolExecutor(13, 13, 1,
TimeUnit.MINUTES, new ArrayBlockingQueue<>(6),
Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
@RequestMapping("/get")
public DeferredResult get(@RequestParam(“question”) String question, @RequestParam(“id”) String id) {
//创建了一个DeferredResult对象，并将其返回给前端。在异步任务执行完毕后，
// 通过调用deferredResult.setResult(result)方法将结果设置到DeferredResult对象中，从而实现异步返回结果给前端。
DeferredResult deferredResult = new DeferredResult<>();
CompletableFuture.supplyAsync(() -> {
try {
callable callable = new callable(question, id);
String answer = callable.call();
Result result = Result.ok(answer);
System.out.println(“接口：” answer);
return result;
} catch (Exception e) {
Result result = Result.fail(e.getMessage());
return result;
}
}, pool).whenComplete((result, throwable) -> {
if (throwable != null) {
result = Result.fail(throwable.getMessage());
deferredResult.setResult(result);
} else {
deferredResult.setResult(result);
}
});
return deferredResult;
}
public class callable implements Callable{
private String id;
private String question;
public callable(String question,String id) {
//这里处理一下userId的长度 因为讯飞那边限制了
if (id.length() >= 30) {
id= id.substring(0, 30);
}
this.question = question;
this.id=id;
}
@Override
public String call() throws Exception {
String answer =main(question,id);
//System.out.println(answer);
answer = JSONUtil.toJsonStr(answer);
botText.content="";//清空
return answer;
}
}再来讲这个和ai之间的对话的接口原理
实际上在每个main函数当中会构建一个WebSocket的服务区跟他进行对话 而当 每一个对话结束 实际上是没有把话说完的 是要进行n次回复 ai说的话才能被拼接好 这个过程就跟 ai一次性说完有比较大的区别 在于他的WebSocket每次都要新建这样的一个对象出来 来和对端的ai进行对话 并且要“等”ai说完
所以这里就遇到了几个问题：
- 主线程没办法精确的知道副线程当中 进行到什么地步了 容易没把话说完就回复给客户端了
- 如果进行了线程复用的话 很可能会串不同用户之间的对话历史记录
- 超时等待的时候 没有跳出 会直接让一个线程死在里面 如果并发线程量够大 足够造成死锁
下面就是解决的办法
代码语言：text复制Java
public String main(String newQuestion,String userid) throws Exception {
if(totalFlag){
totalFlag=false;
NewQuestion=newQuestion;
// 构建鉴权url
String authUrl = getAuthUrl(hostUrl, apiKey, apiSecret);
OkHttpClient client = new OkHttpClient.Builder().build();
String url = authUrl.toString().replace(“http://”, “ws://”).replace(“https://”, “wss://”);
Request request = new Request.Builder().url(url).build();
totalAnswer="";
//这里创建了大模型的新对象 实际上那些发送请求获取答案的操作都是在这个线程中做的
BigModelNew bigModelNew = null;
if (getHistory(userid)!=null){
bigModelNew=new BigModelNew(userid, false,getHistory(userid),stringRedisTemplate);
}
else {
bigModelNew=new BigModelNew(userid, false,historyList,stringRedisTemplate);
}
// 等待 WebSocket 的 run() 方法执行完毕
int maxWaitTime = 10000; // 最大等待时间，单位：毫秒
int currentWaitTime = 0; // 当前已等待的时间，单位：毫秒
int waitInterval = 1000;// 每次等待的时间间隔，单位：毫秒
WebSocket webSocket = client.newWebSocket(request, bigModelNew);
System.out.println(maxWaitTime);
while (currentWaitTime < maxWaitTime) {
if (bigModelNew.getBotContent().equals("")) {
// run() 方法还未执行完毕，可以进行一些其他操作或等待一段时间
Thread.sleep(waitInterval);
System.out.println(“正在执行线程” Thread.currentThread().getName() "…等待时间还剩：" (maxWaitTime-currentWaitTime));
currentWaitTime = waitInterval;
} else {
return bigModelNew.getBotContent();
}
}
}
totalFlag=true;
return “网络开了点小差 试试重新发送你的消息吧”;
}代码解释：
这是Spring提供的一种支持异步处理结果的类。在接口处理过程中，它会先返回一个空的DeferredResult对象给前端，然后在异步任务执行完毕后，通过调用deferredResult.setResult(result)方法将最终的结果设置到DeferredResult对象中，实现异步返回结果给前端。
在异步任务的实现中，使用CompletableFuture.supplyAsync()方法创建一个异步任务，并在其中执行具体的业务逻辑。这里使用了一个callable对象来处理问题和ID，并返回一个回答。
上方的代码解决了1和3 我们打印出来他的执行时间以及线程的名字 以便我们能够追踪到他
而超过了一定的时长 线程就会自动跳出 并且返回报错信息让用户重新发送
而线程2当中我们发现需要缓存历史记录 并且要对用户进行区分 所以在构造大模型对象的时候 我写了一个特殊的构造参数 （这里一定要记得 把Redis给注入进去 否则会爆空指针的错 大概原理是因为新的对象里面并没有被注入Redis 他作为一个新的Bean没有与这个Bean产生绑定的关系 这里涉及到Spring容器构成Bean的原理）
代码语言：text复制Java
// 构造函数
public BigModelNew(@org.springframework.beans.factory.annotation.Value("u s e r I d " ) S t r i n g u s e r I d < b r > ‘ ‘ , @ V a l u e ( " {userId}") String userId<br>` `,@Value("userId")StringuserId<br>‘‘,@Value("{wsCloseFlag}") Boolean wsCloseFlag
,@Value("H i s t o r y L i s t " ) L i s t < R o l e C o n t e n t > H i s t o r y L i s t < b r > ‘ ‘ , @ V a l u e ( " {HistoryList}")List<RoleContent> HistoryList<br>` `,@Value("HistoryList")List<RoleContent>HistoryList<br>‘‘,@Value("{stringRedisTemplate}") StringRedisTemplate stringRedisTemplate) {
this.userId = userId;
this.wsCloseFlag = wsCloseFlag;
this.historyList=HistoryList;
this.stringRedisTemplate = stringRedisTemplate;
}而后 我们使用userId来进行区分 在每一个大模型对象当中 的静态变量中的userId给写死了，并且在初始化的时候 还要根据userId进行查询历史记录 如果有 就填充到其中的历史记录消息数组当中
代码语言：text复制Java
// 从 Redis 中获取对话历史
public List getHistory(String userId) {
String historyStr = stringRedisTemplate.opsForValue().get(“id:” userId “:history”);
if (historyStr==null){
return null;
}
return JSONUtil.toList(JSONUtil.parseArray(historyStr), RoleContent.class);
}
public String main(String newQuestion,String userid) throws Exception {
//…
//这里创建了大模型的新对象 实际上那些发送请求获取答案的操作都是在这个线程中做的
BigModelNew bigModelNew = null;
if (getHistory(userid)!=null){//这里进行了判断 这个用户有没有历史记录
bigModelNew=new BigModelNew(userid, false,getHistory(userid),stringRedisTemplate);
}
else {
bigModelNew=new BigModelNew(userid, false,historyList,stringRedisTemplate);
}
//…
}这样我们就能异步的处理这些对话消息 并且把他们放在对应的缓存空间当中
这个是获取历史记录的方法
代码语言：text复制Java
@RequestMapping("/history")
public Result history(@RequestParam(“id”) String id) {
String history = stringRedisTemplate.opsForValue().get(“id:” id “:history”);
if (history == null) {
return Result.fail(“没有找到历史记录”);
}
JSONArray jsonObject = JSON.parseArray(history);
// String jsonString = JSON.toJSONString(jsonObject);
return Result.ok(jsonObject);
}
