直播模块

slive/pom.xml

@@ -76,6 +76,11 @@
             <artifactId>httpclient</artifactId>
             <version>4.5.9</version>
         </dependency>
+        <!--amqp的起步依赖-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+        </dependency>
     </dependencies>
     <dependencyManagement>
         <dependencies>

slive/src/main/java/com/xhkjedu/slive/config/RabbitConfig.java

@@ -0,0 +1,23 @@
+package com.xhkjedu.slive.config;
+
+import org.springframework.amqp.core.Queue;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author ywx
+ * @classname RabbitConfig
+ * @description
+ * @date 2021/6/18 12:51
+ **/
+@Configuration
+public class RabbitConfig {
+    @Value("${rabbitmq.queue}")
+    private String rabbitmqQueue;
+
+    @Bean
+    public Queue queueA() {
+        return new Queue(rabbitmqQueue); //队列持久
+    }
+}

slive/src/main/java/com/xhkjedu/slive/config/ScheduledConfig.java

@@ -0,0 +1,34 @@
+package com.xhkjedu.slive.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Condition;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+import org.springframework.scheduling.annotation.ScheduledAnnotationBeanPostProcessor;
+
+/**
+ * @author ywx
+ * @className ScheduledConfig
+ * @description @Scheduled 注解，是被一个叫做 ScheduledAnnotationBeanPostProcessor 的类所拦截的，
+ * 所以我们可以根据配置，决定是否创建这个 bean，如果没有这个 bean，@Scheduled 就不会被拦截，那么定时任务肯定不会执行了，
+ * 需要注意的是：这种方式，启动类上面的 @EnableScheduling 需要去掉。
+ * @date 2021/5/18 9:15
+ **/
+@Configuration
+public class ScheduledConfig {
+    @Conditional(ScheduledCondition.class)
+    @Bean
+    public ScheduledAnnotationBeanPostProcessor processor(){
+        return new ScheduledAnnotationBeanPostProcessor();
+    }
+}
+
+class ScheduledCondition implements Condition{
+
+    @Override
+    public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
+        return Boolean.valueOf(context.getEnvironment().getProperty("enable.scheduled"));
+    }
+}

slive/src/main/java/com/xhkjedu/slive/listener/MessageConsumer.java

@@ -0,0 +1,34 @@
+package com.xhkjedu.slive.listener;
+
+import com.alibaba.fastjson.JSON;
+import com.rabbitmq.client.Channel;
+import com.xhkjedu.slive.model.liveplay.LCourseSectionPlayback;
+import com.xhkjedu.slive.service.liveplay.CourseSectionPlaybackService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.annotation.RabbitHandler;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.time.LocalDateTime;
+
+/**
+ * 消息到达消费者监听类
+ */
+@Component
+@Slf4j
+public class MessageConsumer {
+    @Resource
+    private CourseSectionPlaybackService courseSectionPlaybackService;
+
+    //处理收到的rabbit消息的回调方法
+    @RabbitHandler
+    @RabbitListener(queues = "${rabbitmq.queue}")
+    public void onMessage(Message message, Channel channel) throws Exception {
+        String msg = new String(message.getBody(), "UTF-8");
+        System.out.println(LocalDateTime.now()+" 接收消息内容："+msg);
+        LCourseSectionPlayback playback = JSON.parseObject(msg,LCourseSectionPlayback.class);
+        courseSectionPlaybackService.updatePlayback(playback);
+    }
+}

slive/src/main/java/com/xhkjedu/slive/task/LivePlayTask.java

@@ -0,0 +1,59 @@
+package com.xhkjedu.slive.task;
+
+import com.xhkjedu.slive.service.agora.AgoralicenseService;
+import com.xhkjedu.slive.service.liveplay.CourseSectionService;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.util.Date;
+
+/**
+ * @author ywx
+ * @classname LivePlayTask
+ * @description
+ * @date 2021/1/16 13:30
+ **/
+@Component
+public class LivePlayTask {
+
+    @Resource
+    private CourseSectionService courseSectionService;
+    @Resource
+    private AgoralicenseService agoralicenseService;
+
+    /**
+     * 定时任务修改课节状态
+     * @Param []
+     * @Author ywx
+     * @Date 2021/1/16 13:34
+     * @return void
+     **/
+    @Scheduled(cron = "${cron.updateSectionState}")
+    public void updateSectionState(){
+        try {
+            System.out.println(new Date());
+            courseSectionService.updateSectionState2();
+        } catch (Exception e) {
+            System.out.println("定时任务修改课节状态异常："+e.getCause().getMessage());
+        }
+    }
+
+    /**
+     * 定时任务修改license状态
+     * @Param []
+     * @Author ywx
+     * @Date 2021/4/7 10:10
+     * @return void
+     **/
+    @Scheduled(cron = "${cron.updateLicenseState}")
+    public void updateLicenseState() {
+        agoralicenseService.updateLicenseState();
+    }
+
+    //处理未释放的授权
+    @Scheduled(cron = "${cron.updateLicense}")
+    public void updateLicense() {
+        agoralicenseService.updateLicense();
+    }
+}

slive/src/main/resources/application.properties

@@ -52,10 +52,12 @@ spring.redis.jedis.pool.max-active=60000
 redisdatatime=604800
 #secretKey必须为24位
 secretKey=nanhuakaizhangjianwangni
+
 #ws地址
 ws.address=ws://39.104.164.162:8917/ws,ws://39.104.164.162:8918/ws
 #云平台接口地址
 cloudapi=http://scapitest.xhkjedu.com/
+
 #声网
 #APP ID
 agora.appId=b1b07f9eeb2444d0b7ff2205e072b973
@@ -67,3 +69,30 @@ agora.expirationTimeInSeconds=86400
 agora.license.key=2cb21d4aadb44dbca60263df0ab9213d
 #license 密钥
 agora.license.secret=e2439c74719940649aa754ef8c79369d
+
+rabbitmq.queue=xhkjedu.xhschool.livequeue_dev_bd
+spring.rabbitmq.host=116.63.199.166
+spring.rabbitmq.port=5672
+spring.rabbitmq.username=xhkjedu
+spring.rabbitmq.password=xhkjedud07
+spring.rabbitmq.connection-timeout=15000
+#rabbitmq自动签收
+spring.rabbitmq.listener.simple.acknowledge-mode=auto
+# 是否开启消费者重试（为false时关闭消费者重试）
+spring.rabbitmq.listener.simple.retry.enabled=true
+# 最大重试重新投递消息次数
+spring.rabbitmq.listener.simple.retry.max-attempts=3
+# 重试重新投递消息的间隔时间（单位毫秒）
+spring.rabbitmq.listener.simple.retry.initial-interval=30000ms
+#重试次数超过上面的设置之后，是否丢弃(消费者listener抛出异常，是否重回队列（默认true：重回队列，false：不重回队列（可结合死信交换机））
+spring.rabbitmq.listener.simple.default-requeue-rejected=false
+
+#定时任务
+#每隔1分钟修改课节状态
+cron.updateSectionState=0 0/1 * * * ?
+#每天凌晨处理未释放的授权
+cron.updateLicense=0 0 0 * * ?
+#每隔1小时修改license状态
+cron.updateLicenseState=0 0 0/1 * * ?
+#默认为false，不开启定时任务
+enable.scheduled=true

slive/src/main/resources/mapper/agora/AgoralicenseMapper.xml

@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.xhkjedu.slive.mapper.agora.AgoralicenseMapper">
+    <!--列表-->
+    <select id="listAll" resultType="com.xhkjedu.slive.vo.agora.AgoralicenseVo">
+        select l.agid, l.custom, l.cert, l.credential, l.activated, l.used, l.userid, l.createid, l.createtime
+        ,min(lr.activatedtime)starttime,(max(lr.activatedtime)+lr.validityperiod*24*60*60)endtime
+        ,l.username,l.loginname,l.cloudcode,c.cloudname
+        from t_agoralicense l left join t_agoralicense_record lr on l.agid = lr.agid
+        left join t_cloud c on l.cloudcode=c.cloudcode
+        <where>
+            <if test="license.activated!=null">and l.activated=#{license.activated}</if>
+            <if test="license.used!=null">and l.used=#{license.used}</if>
+            <if test="license.cloudcode!=null">and l.cloudcode=#{license.cloudcode}</if>
+        </where>
+        group by l.agid
+    </select>
+    <!--进入直播绑定license-->
+    <select id="inLive" resultType="com.xhkjedu.slive.vo.agora.AgoralicenseVo">
+        select agid, custom, cert, credential, activated, used, cloudcode, userid, createid, createtime
+        from t_agoralicense where userid=#{userid} and activated=1 limit 1
+    </select>
+    <!--获取已激活license-->
+    <select id="getActivatedLicense" resultType="com.xhkjedu.slive.vo.agora.AgoralicenseVo">
+        select agid, custom, cert, credential, activated, used, cloudcode, userid, createid, createtime
+        from t_agoralicense where activated=1 and used=0 order by rand() limit 1
+    </select>
+    <!--获取未激活license-->
+    <select id="getLicense" resultType="com.xhkjedu.slive.vo.agora.AgoralicenseVo">
+        select agid, custom, cert, credential, activated, used, cloudcode, userid, createid, createtime
+        from t_agoralicense where activated=0 order by rand() limit 1
+    </select>
+    <!--退出直播解绑license-->
+    <update id="outLive">
+        update t_agoralicense set used=0,cloudcode=null,schoolid=null,userid=0,username=null,loginname=null
+        where userid=#{userid};
+    </update>
+    <!--处理未释放的授权-->
+    <update id="updateLicense">
+        update t_agoralicense set used=0,cloudcode=null,schoolid=null,userid=0,username=null,loginname=null where used=1
+    </update>
+    <!--更新失效的license的激活状态-->
+    <update id="updateActivate">
+        update (SELECT agid,(max(activatedtime)+validityperiod*24*60*60)activatedtime
+        FROM t_agoralicense_record GROUP BY agid)r LEFT JOIN t_agoralicense l ON r.agid=l.agid
+        set l.activated=2
+        WHERE r.activatedtime<![CDATA[ < ]]>UNIX_TIMESTAMP(NOW());
+    </update>
+    <update id="endLive2">
+        update t_agoralicense l set
+        l.used=0,l.cloudcode=null,l.schoolid=null,l.userid=0,l.username=null,l.loginname=null
+        where (
+        <foreach collection="license.userids" item="userid" index="index" separator="or">
+            l.userid=#{userid}
+        </foreach>)
+    </update>
+    <!--获取学校正在使用直播人数-->
+    <select id="getUseNum" resultType="java.lang.Integer">
+        select count(*) from t_agoralicense where schoolid=#{schoolid} and used=1 and userid!=#{userid}
+    </select>
+</mapper>