public TimedTask toLiveTask(LiveInfo liveInfo) {Jedis jedis = JedisUtils.getResource();jedis.lpush(Constants.WAIT_LIVE_TASK_QUEUE, GsonUtils.toJson(liveInfo));Timer timer = new Timer(true);boolean flag = true;TimedTask timedTask = new TimedTask(flag,liveInfo);timer.schedule(timedTask, Global.WAIT_TIME);jedis.close();return timedTask;}

@Transactional(readOnly = false)public String liveOnTask(LiveInfo liveInfo,TimedTask timedTask) {Jedis jedis = JedisUtils.getResource();boolean flag = true;Timer t = new Timer(true);timedTask = new TimedTask(flag,liveInfo);t.schedule(timedTask, Global.WAIT_TIME);//定时器开始while(flag && timedTask.flag) {//任务还在synchronized (this) {//并发处理//判断当前直播总数是否小于最大直播总数,如果小于则分配直播地址,大于则继续等待    if(Integer.valueOf(jedis.get(Constants.ON_LIVE_MAX_SUM))<Global.MAX_ON_LIVE_SUM) {//判断当前直播总数liveInfo.setBeginTime(new Date());//直播开始时间liveInfo.setLiveUrl(IdGen.uuid());//分发直播地址Transaction tx = jedis.multi();//redis事务liveInfoDao.insert(liveInfo);tx.sadd(Constants.ON_LIVE_TASK_QUEUE, GsonUtils.toJson(liveInfo));//加入直播任务queuetx.incr(Constants.ON_LIVE_MAX_SUM);//直播人数增加tx.exec();jedis.close();//System.out.println("返回直播地址："+liveInfo.getLiveUrl());flag = false;timedTask.cancel();//直播地址分配成功,定时任务取消return liveInfo.getLiveUrl();}else {flag = timedTask.flag;//继续等待标识//System.out.println("等待中，继续监控任务");}}}jedis.close();return null;}

@Transactional(readOnly = false)public void liveOffTask(LiveInfo liveInfo){Jedis jedis = JedisUtils.getResource();if (liveInfo == null) return;if(liveInfo.getLiveUrl().equals("") || liveInfo.getLiveUrl()==null){jedis.lrem(Constants.WAIT_LIVE_TASK_QUEUE, 1, GsonUtils.toJson(liveInfo));//取消任务jedis.close();return ;}Transaction tx = jedis.multi();liveInfo.setEndTime(new Date());liveInfoDao.updateByPrimaryKey(liveInfo);//同步到数据库tx.decr(Constants.ON_LIVE_MAX_SUM);//直播总数递减     tx.srem(Constants.ON_LIVE_TASK_QUEUE, GsonUtils.toJson(liveInfo));//清除直播任务tx.exec();jedis.close();}

4.定时器

public class TimedTask extends TimerTask{public boolean flag;private LiveInfo liveInfo;public TimedTask(boolean flag,LiveInfo liveInfo) {this.flag = flag;this.liveInfo = liveInfo;}@Overridepublic void run() {Jedis jedis = JedisUtils.getResource();jedis.lrem(Constants.WAIT_LIVE_TASK_QUEUE, 1, GsonUtils.toJson(liveInfo));//等待任务队列jedis.close();flag = false;//取消等待标记//System.out.println("等待超时了，任务结束"+ "flag="+flag);}}

@ResponseBody
@RequestMapping(value = "/getLiveUrl", method = RequestMethod.POST)
public String getLiveAddr(@RequestParam(value = "userId", required = false) String userId,@RequestParam(value = "title", required = false) String title,@RequestParam(value = "summary", required = false) String summary,@RequestParam(value = "location", required = false) String location){LiveInfo liveInfo = new LiveInfo();liveInfo.setUserId(userId);liveInfo.setTitle(title);liveInfo.setSummary(summary);liveInfo.setLocation(location);liveInfo.setBeginTime(new Date());liveInfo.setLiveUrl(IdGen.uuid());return liveTaskService.liveOnTask(liveInfo, liveTaskService.toLiveTask(liveInfo));}

@ResponseBody@RequestMapping(value = "/liveOut", method = RequestMethod.GET)public void liveOut(@RequestParam(value = "liveUrl", required = false) String liveUrl) {LiveInfo liveInfo = liveInfoService.getLiveInfoByLiveUrl(liveUrl);if(liveInfo==null) {throw new PhoneException(HttpStatus.BAD_REQUEST, new Error(400, "直播地址错误！"));}liveTaskService.liveOffTask(liveInfo);}