开源模型应用落地-业务优化篇（八）

一、前言

在之前的学习中，我相信您已经学会了一些优化技巧，比如分布式锁、线程池优化、请求排队、服务实例扩容和消息解耦等等。现在，我要给您介绍最后一篇业务优化的内容了。这个优化方法是通过定时统计问题的请求频率，然后将一些经常被请求的问题缓存起来，以提高系统的响应速度。

二、术语

2.1、任务调度框架

是一种用于管理和执行任务的软件工具或平台。它提供了一种结构和机制，使用户能够以自动化的方式安排、调度和执行任务，以满足特定的需求和要求。

2.2、分布式任务调度框架

是一种用于管理和调度分布式环境中任务的软件工具或平台。它专注于在分布式系统中协调和执行任务，以提高整体性能、可伸缩性和容错性。

分布式任务调度框架通常用于处理大规模任务和作业，并利用集群、云计算或容器化环境中的多个计算节点来并行执行任务。它们提供了一种分布式任务调度器，可以协调和分配任务到可用的计算节点，并监控任务的执行状态和进度。

2.3、XXL-JOB

是一个开源的分布式任务调度平台，用于解决大规模任务调度和分布式定时任务管理的需求。它提供了一个可视化的任务调度中心，可以集中管理和调度各种类型的任务，包括定时任务、流程任务和API任务等。

2.4、Milvus

是一个开源的向量数据库引擎，专门用于存储和处理大规模高维向量数据。它提供了高效的向量索引和相似性搜索功能，使用户能够快速地进行向量数据的存储、查询和分析。

Milvus的设计目标是为了满足现代应用中对大规模向量数据的需求，例如人脸识别、图像搜索、推荐系统等。它采用了向量空间模型和多种索引算法，包括倒排索引、近似最近邻（Approximate Nearest Neighbor，ANN）等，以支持高效的相似性搜索。

Milvus提供了易于使用的编程接口和丰富的功能，使用户可以方便地插入、查询和分析向量数据。它支持多种数据类型的向量，包括浮点型、整型等，也支持多种距离度量方法，如欧氏距离、余弦相似度等。

Milvus还提供了分布式部署和横向扩展的能力，可以在多台机器上构建高可用性和高性能的向量数据库集群。它支持数据的分片和负载均衡，可以处理大规模数据集和高并发查询。

三、前置条件

3.1、已经根据前面的“开源模型应用落地”的学习搭建起完整AI流程

1) 如何部署AI服务

2) 如何使用向量数据库

3) 如何使用RocketMQ

......

本篇将通过定时任务周期性的统计问题请求的频次，并从向量数据库中，将热点问题同步至Redis，实现缓存前置，提升访问性能。

四、技术实现

4.1、新增定时任务处理类

import cn.hutool.core.map.MapUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RMap;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Map;

@Slf4j
@Component
public class HotTopicStatistics {
    private static final Long DEFAULT_HOTSPOT_THRESHOLD = 10L;

    @Autowired
    private RedisUtils redisUtils;
    @Autowired
    private ContentCacheUtils contentCacheUtils;

    @Scheduled(cron = "*/30 * * * * ?")
    public void statistics() {
        RMap<String, String> rmap = redisUtils.hincget("CONTENT_COUNTER");

        if (MapUtil.isNotEmpty(rmap)) {

            for (Map.Entry<String, String> entry : rmap.entrySet()) {
                String keyword = entry.getKey();
                Long count = Long.parseLong(entry.getValue());
                // 计数器统计数值 > 热度阈值
                if(count.compareTo(DEFAULT_HOTSPOT_THRESHOLD) > 0){
                    // 从向量数据库中拉取数据
                    log.info("从向量数据库中拉取数据");

                    String cacheContent = contentCacheUtils.cacheFromMilvus(keyword);

                    if(StringUtils.isNotEmpty(cacheContent) && StringUtils.isNotBlank(cacheContent)){
                        log.info("将热点内容缓存至redis中，过期时间设置为3600秒，内容为：{}",cacheContent);

                        // 将热点内容缓存至redis中，过期时间设置为3600秒
                        redisUtils.buckSet(keyword,cacheContent,60*60);
                    }

                }
            }

        }
    }

}

4.2、新增内容缓存公共类

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.lang.Console;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Slf4j
@Component
public class ContentCacheUtils {
    private static final int DIM = 256;

    @Autowired
    private AIChatUtils aiChatUtils;
    @Autowired
    private MilvusUtils milvusUtils;


    public String cacheFromMilvus(String keyWord){
        if (StringUtils.isEmpty(keyWord) || StringUtils.isBlank(keyWord)){
            return null;
        }

        float[] vector = aiChatUtils.getVector("", keyWord);
        double[] double_arr = milvusUtils.pretreatment(vector, DIM);
        Float[] float_arr = Convert.toFloatArray(double_arr);
        List<Float> vectorList = CollUtil.list(false, float_arr);

        List search_vectors = new ArrayList(1);
//        打印日志
        Console.log(search_vectors);
        search_vectors.add(vectorList);
        Map<String, String> resultMap = milvusUtils.search_data_vector("tb_content", "keyword",
                search_vectors, null, 1, CollUtil.list(false, "content"));


        String status = resultMap.get("status");
        String cacheContent = null;
        if (StringUtils.equals(status, "0")) {
            cacheContent = resultMap.get("content");

        }

        return cacheContent;
    }

}

4.3、修改Redis公共类

增加以下方法

public  RMap<String, String> hincget(String key){
	RMap<String, String> rmap = null;

	if (StringUtils.isNotEmpty(key) && StringUtils.isNoneBlank(key) ) {
		rmap = redissonClient.getMap(key);
	}
	return rmap;
}

public void buckSet(String key, String value,long second) {
	if (StringUtils.isNotEmpty(key) && StringUtils.isNoneBlank(key) && StringUtils.isNotEmpty(value) && StringUtils.isNoneBlank(value)) {
		RBucket<String> bucket =  redissonClient.getBucket(key);
		bucket.set(value,second, TimeUnit.SECONDS);
	}
}

4.4、修改业务处理类

使用上面内容缓存公共类替换早前未封装的代码

五、测试

5.1、启动Redis

启动windows版本的redis服务，redis-server.exe redis.windows.conf

5.2、将CONTENT_COUNTER的值设置为11

下面使用Redis Desktop Manager工具编辑CONTENT_COUNTER的值

5.3、启动Milvus Server，并初始化数据

5.4、启动SpringBoot项目

（一）运行Application

（二）Redis当前只有一个Key，热点内容未缓存

（三）定时任务触发

（四）Redis缓存热点内容

六、附带说明

6.1、Spring Boot开启定时任务

启用类增加@EnableScheduling注解

需要将具体任务类加入到Spring管理，例如：增加@Component注解

6.2、实际项目中，应用使用分布式任务调度平台去替换本示例中SpringBoot内置的任务调度功能

6.3、Milvus Server启动超时

直接编辑milvus下面的__init__.py文件，将timeout设置大一些

6.4、本章BusinessHandler完整代码

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.lang.Console;
import com.alibaba.fastjson.JSON;
import io.netty.channel.ChannelHandler;
import lombok.extern.slf4j.Slf4j;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;


/**
 * @Description: 处理消息的handler
 */
@Slf4j
@ChannelHandler.Sharable
@Component
public class BusinessHandler extends AbstractBusinessLogicHandler<TextWebSocketFrame> {
    public static final String LINE_UP_QUEUE_NAME = "AI-REQ-QUEUE";
    private static final String LINE_UP_LOCK_NAME = "AI-REQ-LOCK";

    private static final int MAX_QUEUE_SIZE = 100;

//    @Autowired
//    private TaskUtils taskExecuteUtils;
//    @Autowired
//    private AIChatUtils aiChatUtils;
//    @Autowired
//    private MilvusUtils milvusUtils;

    @Autowired
    private RedisUtils redisUtils;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private NettyConfig nettyConfig;
    @Autowired
    private RocketMQProducer rocketMQProducer;
    @Autowired
    private ContentCacheUtils contentCacheUtils;


    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        String channelId = ctx.channel().id().asShortText();
        log.info("add client,channelId：{}", channelId);
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        String channelId = ctx.channel().id().asShortText();
        log.info("remove client,channelId：{}", channelId);
    }


    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame)
            throws Exception {
        // 获取客户端传输过来的消息
        String content = textWebSocketFrame.text();
        // 兼容在线测试
        if (StringUtils.equals(content, "PING")) {
            buildResponse(channelHandlerContext, ApiRespMessage.builder().code(String.valueOf(StatusCode.SUCCESS.getCode()))
                    .respTime(String.valueOf(System.currentTimeMillis()))
                    .msgType(String.valueOf(MsgType.HEARTBEAT.getCode()))
                    .contents("心跳测试，很高兴收到你的心跳包")
                    .build());

            return;
        }
        log.info("接收到客户端发送的信息: {}", content);

        Long userIdForReq;
        String msgType = "";
        String contents = "";

        try {
            ApiReqMessage apiReqMessage = JSON.parseObject(content, ApiReqMessage.class);
            msgType = apiReqMessage.getMsgType();
            contents = apiReqMessage.getContents();


            userIdForReq = apiReqMessage.getUserId();
            // 用户身份标识校验
            if (null == userIdForReq || (long) userIdForReq <= 10000) {
                ApiRespMessage apiRespMessage = ApiRespMessage.builder().code(String.valueOf(StatusCode.SYSTEM_ERROR.getCode()))
                        .respTime(String.valueOf(System.currentTimeMillis()))
                        .contents("用户身份标识有误!")
                        .msgType(String.valueOf(MsgType.SYSTEM.getCode()))
                        .build();
                buildResponseAndClose(channelHandlerContext, apiRespMessage);
                return;
            }


            if (StringUtils.equals(msgType, String.valueOf(MsgType.CHAT.getCode()))) {
                // 对用户输入的内容进行自定义违规词检测
                // 对用户输入的内容进行第三方在线违规词检测
                // 对用户输入的内容进行组装成Prompt
                // 对Prompt根据业务进行增强（完善prompt的内容）
                // 对history进行裁剪或总结（检测history是否操作模型支持的上下文长度，例如qwen-7b支持的上下文长度为8192）
                // ...

//                通过线程池来处理
//                String messageId = apiReqMessage.getMessageId();
//                List<ChatContext> history = apiReqMessage.getHistory();
//                AITaskReqMessage aiTaskReqMessage = AITaskReqMessage.builder().messageId(messageId).userId(userIdForReq).contents(contents).history(history).build();
//                taskExecuteUtils.execute(aiTaskReqMessage);


                // 违规词检测
                log.info("contents: {}",contents);
                if (WordDetection.contains_illegal_word(contents)) {
                    log.warn("the content sent contains illegal words");
                    ApiRespMessage apiRespMessage = ApiRespMessage.builder().code(String.valueOf(StatusCode.ILLEGAL_WORDS_FAILURE_731.getCode()))
                            .respTime(String.valueOf(System.currentTimeMillis()))
                            .contents("内容不合规!")
                            .msgType(String.valueOf(MsgType.SYSTEM.getCode()))
                            .build();
                    buildResponseAndClose(channelHandlerContext, apiRespMessage);
                    return;
                }

                String[] filterWords = new String[]{"一", "语文", "老师"};
                List<String> keyWordsList = KeyWordsUtils.extractKeywords(contents, Arrays.asList(filterWords));
                String keyWord = CollUtil.join(keyWordsList, "");
                log.info("keyWord: {}", keyWord);
                String cacheContent = redisUtils.buckGet(keyWord);
                // 返回redis中的缓存数据
                if (StringUtils.isNotEmpty(cacheContent) && StringUtils.isNoneBlank(cacheContent)) {
                    buildResponse(channelHandlerContext, ApiRespMessage.builder().code(String.valueOf(StatusCode.SUCCESS.getCode()))
                            .respTime(String.valueOf(System.currentTimeMillis()))
                            .msgType(String.valueOf(MsgType.CHAT.getCode()))
                            .contents(cacheContent)
                            .build());
                    return;
                } else {
                    // 从milvus中检索数据
                    cacheContent = contentCacheUtils.cacheFromMilvus(keyWord);

                    if (StringUtils.isNotEmpty(cacheContent) && StringUtils.isNoneBlank(cacheContent)) {
                        buildResponse(channelHandlerContext, ApiRespMessage.builder().code(String.valueOf(StatusCode.SUCCESS.getCode()))
                                .respTime(String.valueOf(System.currentTimeMillis()))
                                .msgType(String.valueOf(MsgType.CHAT.getCode()))
                                .contents(cacheContent)
                                .build());
                        return;
                    }

                    //投递消息
                    String msg = "{\"msg\":\""+keyWord+"\"}";
                    rocketMQProducer.send("ai-topic",msg);
                }
//                通过队列来缓冲
                boolean flag = true;

                RLock lock = redissonClient.getLock(LINE_UP_LOCK_NAME);
                String queueName = LINE_UP_QUEUE_NAME + "-" + nettyConfig.getNode();

                //尝试获取锁，最多等待3秒，锁的自动释放时间为10秒
                if (lock.tryLock(3, 10, TimeUnit.SECONDS)) {
                    try {
                        if (redisUtils.queueSize(queueName) < MAX_QUEUE_SIZE) {
                            redisUtils.queueAdd(queueName, content);
                            log.info("当前线程为：{}， 添加请求至redis队列", Thread.currentThread().getName());
                        } else {
                            flag = false;
                        }
                    } catch (Throwable e) {
                        log.error("系统处理异常", e);
                    } finally {
                        lock.unlock();
                    }
                } else {
                    flag = false;
                }

                if (!flag) {
                    buildResponse(channelHandlerContext, ApiRespMessage.builder().code(String.valueOf(StatusCode.SUCCESS.getCode()))
                            .respTime(String.valueOf(System.currentTimeMillis()))
                            .msgType(String.valueOf(MsgType.SYSTEM.getCode()))
                            .contents("当前排队人数较多，请稍后再重试！")
                            .build());
                }


            } else if (StringUtils.equals(msgType, String.valueOf(MsgType.INIT.getCode()))) {
                //一、业务黑名单检测（多次违规，永久锁定）

                //二、账户锁定检测（临时锁定）

                //三、多设备登录检测

                //四、剩余对话次数检测

                //检测通过，绑定用户与channel之间关系
                addChannel(channelHandlerContext, userIdForReq);
                String respMessage = "用户标识: " + userIdForReq + " 登录成功";

                buildResponse(channelHandlerContext, ApiRespMessage.builder().code(String.valueOf(StatusCode.SUCCESS.getCode()))
                        .respTime(String.valueOf(System.currentTimeMillis()))
                        .msgType(String.valueOf(MsgType.INIT.getCode()))
                        .contents(respMessage)
                        .build());

            } else if (StringUtils.equals(msgType, String.valueOf(MsgType.HEARTBEAT.getCode()))) {

                buildResponse(channelHandlerContext, ApiRespMessage.builder().code(String.valueOf(StatusCode.SUCCESS.getCode()))
                        .respTime(String.valueOf(System.currentTimeMillis()))
                        .msgType(String.valueOf(MsgType.HEARTBEAT.getCode()))
                        .contents("心跳测试，很高兴收到你的心跳包")
                        .build());
            } else {
                log.info("用户标识: {}, 消息类型有误，不支持类型: {}", userIdForReq, msgType);
            }


        } catch (Exception e) {
            log.warn("【BusinessHandler】接收到请求内容：{}，异常信息：{}", content, e.getMessage(), e);
            // 异常返回
            return;
        }

    }


}