代码提交

liusheng

2024-05-12 f243bb8af00a2464d16536d42150a4107f87b4ae

 ruoyi-admin/src/main/java/com/ruoyi/web/component/RabbitMqReceiver.java

@@ -2,16 +2,14 @@

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.web.task.PhoneTask;
import com.ruoyi.web.task.PhoneTask2;
import com.smartor.common.SendService;
import com.smartor.config.RabbitMqConfig;
import com.smartor.domain.IvrLibaTemplateVO;
import com.smartor.domain.IvrTask;
import com.smartor.domain.IvrTaskSingle;
import com.smartor.domain.IvrTaskcallMQ;
import com.smartor.config.PhoneUtils;
import com.smartor.config.RabbitMqCallPhoneConfig;
import com.smartor.domain.*;
import com.smartor.mapper.IvrTaskMapper;
import com.smartor.mapper.IvrTaskSingleMapper;
import com.smartor.service.IIvrLibaTemplateService;
@@ -19,12 +17,11 @@
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;

import java.io.IOException;
import java.util.List;
@@ -35,6 +32,7 @@
@Component//监听此队列
public class RabbitMqReceiver {


    private static IIvrLibaTemplateService ivrLibaTemplateService;

    private static IvrTaskSingleMapper ivrTaskcallMapper;
@@ -44,6 +42,8 @@
    private static SendService sendService;

    private static RedisCache redisCache;

    private static RabbitMqCallPhoneConfig rabbitMqCallPhoneConfig;

    // 创建固定大小的线程池
    private static final ExecutorService executorService = Executors.newFixedThreadPool(10);
@@ -57,6 +57,11 @@
    @Autowired
    public void setSendService(SendService sendService) {
        RabbitMqReceiver.sendService = sendService;
    }

    @Autowired
    public void setRabbitMqCallPhoneConfig(RabbitMqCallPhoneConfig rabbitMqCallPhoneConfig) {
        RabbitMqReceiver.rabbitMqCallPhoneConfig = rabbitMqCallPhoneConfig;
    }

    @Autowired
@@ -74,37 +79,22 @@
        RabbitMqReceiver.redisCache = redisCache;
    }

//    @RabbitListener(queues = RabbitMqConfig.delay_queue)
//    public void consultReceiveDealy(String content, Message message, Channel channel) throws IOException {
//        log.info("----------------接收延迟队列消息--------------------");
//        //通知 MQ 消息已被接收,可以ACK(从队列中删除)了
//        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//        try {
//            log.info("=============Do Something==============");
//        } catch (Exception e) {
//            log.error("============消费失败,尝试消息补发再次消费!==============");
//            log.error(e.getMessage());
//            /**
//             * basicRecover方法是进行补发操作，
//             * 其中的参数如果为true是把消息退回到queue但是有可能被其它的consumer(集群)接收到，
//             * 设置为false是只补发给当前的consumer
//             */
//            channel.basicRecover(false);
//        }
//    }

    /**
     * 电话随访
     * concurrency = "50"  并发数为50
     */
    @RabbitListener(queues = RabbitMqConfig.phone_queue, concurrency = "50")
    @RabbitListener(queues = "phone_queue", concurrency = "50")
    public void phoneVisit(String content, Message message, Channel channel) throws IOException {
        try {
            IvrTaskcallMQ ivrTaskcallMQ = null;

            ObjectMapper mapper = new ObjectMapper();
            try {
                ivrTaskcallMQ = mapper.readValue("{" + content + "}", IvrTaskcallMQ.class);
                if (!content.contains("{")) {
                    ivrTaskcallMQ = mapper.readValue("{" + content + "}", IvrTaskcallMQ.class);
                } else {
                    ivrTaskcallMQ = mapper.readValue(content, IvrTaskcallMQ.class);
                }
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
@@ -125,22 +115,50 @@
                    break;
                }
//                //通过多线程的方式去打电话
//                executorService.submit(new PhoneTask(ivrTaskcall1, ivrLibaTemplateVO1, redisCache, channel, message, sendService));
                new PhoneTask2().runPhone(ivrTaskcall1, ivrLibaTemplateVO1, redisCache, channel, message, sendService);
                executorService.submit(new PhoneTask(ivrTaskcall1, ivrLibaTemplateVO1, redisCache, rabbitMqCallPhoneConfig, message, sendService));
            }
            //通知 MQ 消息已被接收,可以ACK(从队列中删除)了   （这个需要根据业务再去处理ACK）
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

            log.info("=============Do Something==============");
        } catch (Exception e) {
            log.error("============消费失败,尝试消息补发再次消费!==============");
            log.error(e.getMessage());
            /**
             * basicRecover方法是进行补发操作，
             * 其中的参数如果为true是把消息退回到queue但是有可能被其它的consumer(集群)接收到，
             * 设置为false是只补发给当前的consumer
             */
            channel.basicRecover(false);
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
        }
    }

    /**
     * 机器人第一句话语音
     * concurrency = "50"  并发数为50
     */
    @RabbitListener(queues = "ob_queue", concurrency = "50")
    public void obVisit(String content, Message message, Channel channel) throws IOException {
        try {
            IvrTaskcallPhoneMQ ivrTaskcallMQ = null;
            ObjectMapper mapper = new ObjectMapper();
            ivrTaskcallMQ = mapper.readValue(content, IvrTaskcallPhoneMQ.class);
            //判断患者是否已经接电话
            String str = redisCache.getCacheObject(ivrTaskcallMQ.getUuid() + "state_id");
            System.out.println("-----------------" + str);
            if (StringUtils.isNotEmpty(str) && str.equals("0")) {
                //患者已经接听了电话
                new PhoneUtils().ttsPlayback(ivrTaskcallMQ.getScript(), ivrTaskcallMQ.getUuid());
            } else if (StringUtils.isNotEmpty(str) && str.equals("-10")) {
                //这个说明,已经打了指定遍数还是没有人接，可以将消息从队列中移除了
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                return;
            } else {
                //进行队列尾部，进行下一次校验
//                channel.basicRecover(false);
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
                System.out.println("++++++++++++++++++++++" + str);
                return;
            }
            //通知 MQ 消息已被接收,可以ACK(从队列中删除)了   （这个需要根据业务再去处理ACK）
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            log.error("============消费失败,尝试消息补发再次消费!==============");
            log.error(e.getMessage());
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
        }
    }