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

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 {
    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 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 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;
        }
 
    }
 
 
}