ai对话---多线程并发处理问题-Toy模板网

这篇具有很好参考价值的文章主要介绍了ai对话---多线程并发处理问题。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

ai对话—多线程并发处理问题

先简单回顾一下旧版本的对话接口如何实现

其实这里更多是涉及到多线程工作上的学习问题

在初代版本中我自己以为的搞了一个线程池就能完成多线程的任务了

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卡住了就没有办法得知发生了什么事情

于是这里就有了下面的重写了的语句

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说完

所以这里就遇到了几个问题：

主线程没办法精确的知道副线程当中进行到什么地步了容易没把话说完就回复给客户端了
如果进行了线程复用的话很可能会串不同用户之间的对话历史记录
超时等待的时候没有跳出会直接让一个线程死在里面如果并发线程量够大足够造成死锁

下面就是解决的办法

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的原理）

Java
// 构造函数
public BigModelNew(@org.springframework.beans.factory.annotation.Value("

{userId}") String userId<br>` `,@Value("

{wsCloseFlag}") Boolean wsCloseFlag
,@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进行查询历史记录如果有就填充到其中的历史记录消息数组当中

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);
}
//…
}

这样我们就能异步的处理这些对话消息并且把他们放在对应的缓存空间当中

这个是获取历史记录的方法文章来源地址https://www.toymoban.com/news/detail-828658.html