2210088963/AiJiNanProject
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1.修改定时人开关播sql语句逻辑问题

Browse Source 
2.修复了因为定时任务未绑定租户ID而报错的情况
3.完成对轮播排序的支持
4.优化了点赞功能导致客户端未响应的问题
5.剪辑后建立SSE连接，后端不实现断开功能，直到阿里云回调后给前端发送剪辑完成指令
6.修改APP搜索接口的最近一天，一周，一个月的逻辑
7.重新封装栏目信息，弃用/platform/live/channel/getComponent和/platform/live/channel/getChannelComponent接口
This commit is contained in:
2210088963 2025-03-19 15:52:42 +08:00
parent 9bf5c6972b
commit 9246ec9327
20 changed files with 423 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/com/platform/config/CorsConfig.java
View File

@ -36,6 +36,4 @@ public class CorsConfig extends WebMvcConfigurerAdapter {
return new CorsFilter(urlBasedCorsConfigurationSource);
}
}

13
src/main/java/com/platform/config/MybatisPlusConfig.java
View File

@ -69,6 +69,10 @@ public class MybatisPlusConfig implements ApplicationRunner {
return new LongValue(tenantId.toString());
}
} catch (Exception e) {//处理定时任务
// 如果是定时任务，手动设置租户 ID
if (isScheduledTaskContext()) { // <-- 自定义判断逻辑
return new LongValue(1L); // <-- 返回系统默认租户 ID
}
logger.warn("未登录获取不到 SecurityManager {}", e.getMessage());
}
return null;
@ -116,4 +120,13 @@ public class MybatisPlusConfig implements ApplicationRunner {
List<String> tableNames = sysSchemaService.queryTableNames(properties.getColumnName());
tables.addAll(tableNames);
}
/**
* 判断当前是否在定时任务上下文中
*/
private boolean isScheduledTaskContext() {
// 通过线程名称或自定义标记判断（根据实际任务线程池配置调整）
return Thread.currentThread().getName().startsWith("pool-")
|| Thread.currentThread().getName().contains("scheduler");
}
}

5
src/main/java/com/platform/config/ShiroConfig.java
View File

@ -103,6 +103,11 @@ public class ShiroConfig {
filterMap.put("/trtc/**", "anon");
filterMap.put("/websocket/**", "anon");
filterMap.put("/**", "oauth2");
//APP详细
filterMap.put("/live/merge/detail/", "anon");
shiroFilter.setFilterChainDefinitionMap(filterMap);
return shiroFilter;

2
src/main/java/com/platform/modules/foreign/dto/AppSearchDto.java
View File

@ -11,13 +11,13 @@ import java.util.List;
@Data
public class AppSearchDto {
private String liveDesc;
private Integer timeState;
private Date startTime;
private Date endTime;
private String channelName;
private List<String> channelNameList;
private Integer page;
private Integer size;
public List<String> getChannelNameList() {
if(StringUtils.isBlank(channelName)) {
return Collections.emptyList();

2
src/main/java/com/platform/modules/job/config/ScheduleConfig.java
View File

@ -33,7 +33,7 @@ public class ScheduleConfig {
prop.put("org.quartz.jobStore.clusterCheckinInterval", "15000");
prop.put("org.quartz.jobStore.maxMisfiresToHandleAtATime", "1");
prop.put("org.quartz.jobStore.misfireThreshold", "12000");
prop.put("org.quartz.jobStore.misfireThreshold", "60000");
prop.put("org.quartz.jobStore.tablePrefix", "QRTZ_");
factory.setQuartzProperties(prop);

71
src/main/java/com/platform/modules/live/controller/ClipController.java Normal file
View File

@ -0,0 +1,71 @@
package com.platform.modules.live.controller;
import io.swagger.models.auth.In;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
// ① 控制器类
@Slf4j
@RestController
public class ClipController {
private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
@GetMapping(value = "/status-stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter streamTranscodingStatus(@RequestParam String sceneId) {
SseEmitter emitter = new SseEmitter(3L * 60 * 60 * 1000);
// 直接发送单次事件（关键修改点）
try {
emitter.send(
SseEmitter.event()
// .id("CONNECT_EVENT") // 可选：给事件添加ID
// 事件名称
.name("Connect")
// 核心数据（会按SSE规范自动包装）
.data("建立连接成功！")
);
} catch (IOException e) {
emitter.completeWithError(e);
}
// 维护连接池（根据业务需求调整）
emitters.put(sceneId, emitter);
// 清理逻辑
emitter.onCompletion(() -> emitters.remove(sceneId, emitter));
emitter.onTimeout(() -> {
emitters.remove(sceneId, emitter);
emitter.complete();
});
return emitter;
}
// ④ 推送剪辑完成事件（在其他服务中调用）
public void sendClipFinishedEvent(Integer videoId) {
emitters.forEach((clientId, emitter) -> {
try {
// 直接构建 String 消息（示例使用 JSON 格式）
String message = String.format("{\"videoId\":\"%s\"} \"message\":\"已经完成剪辑\"", videoId);
emitter.send(SseEmitter.event()
.name("clip_finished") // 事件名称
.data(message) // 直接发送 String
);
} catch (IOException e) {
log.error("剪辑回调失败");
}
});
}
}

7
src/main/java/com/platform/modules/live/controller/LiveChannelController.java
View File

@ -1,10 +1,8 @@
package com.platform.modules.live.controller;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import com.baomidou.mybatisplus.mapper.EntityWrapper;
import com.platform.modules.live.entity.LiveChannelComponentEntity;
import com.platform.modules.live.service.LiveChannelTenantService;
import com.platform.modules.live.service.impl.LiveChannelComponentServiceImpl;
@ -192,11 +190,12 @@ public class LiveChannelController extends AbstractController{
/**
* 查询现场大屏和栏目封装
*
* @return 集合
*/
@PostMapping("/appView")
public HashMap<Object, Object> appView() {
return liveChannelService.appView();
public R appView() {
return R.ok().put("appView",liveChannelService.appView());
}
/**

26
src/main/java/com/platform/modules/live/controller/LiveMultimediaController.java
View File

@ -36,9 +36,9 @@ import springfox.documentation.annotations.ApiIgnore;
/**
* 素材信息
*
*
* @author weiyan
* @email
* @email
* @date 2019-03-12 10:09:03
*/
@RestController
@ -68,7 +68,7 @@ public class LiveMultimediaController extends AbstractController{
PageUtils page = liveMultimediaService.queryPageByParams(params);
return R.ok().put("page", page);
}
/**
* 列表
*/
@ -87,7 +87,7 @@ public class LiveMultimediaController extends AbstractController{
PageUtils page = liveMultimediaService.queryPageByParams(params);
return R.ok().put("page", page);
}
@GetMapping("/common/saveToLocal")
@ApiOperation(value = "保存到本地")
public R saveToLocal(String videoUrl,String title, HttpServletResponse response) {
@ -97,7 +97,7 @@ public class LiveMultimediaController extends AbstractController{
liveMultimediaService.saveToLocal(videoUrl,title,response);
return R.ok();
}
/**
* 信息
*/
@ -107,7 +107,7 @@ public class LiveMultimediaController extends AbstractController{
public R info(@RequestParam("id") Integer id){
return R.ok().put("info", liveMultimediaService.info(id));
}
/**
* 根据mediaId查询素材信息
*/
@ -117,7 +117,7 @@ public class LiveMultimediaController extends AbstractController{
public R getInfoByMediaId(@RequestParam("mediaId") String mediaId){
return R.ok().put("info", liveMultimediaService.getInfoByMediaId(mediaId));
}
/**
* 保存
*/
@ -125,8 +125,12 @@ public class LiveMultimediaController extends AbstractController{
@PostMapping("/save")
@RequiresPermissions("live:multimedia:save")
@ApiOperation("保存素材")
public R save(@ModelAttribute LiveMultimediaEntity liveMultimedia, MultipartFile mediaFile){
return R.ok().put("info", liveMultimediaService.save(liveMultimedia, mediaFile, getUserId()));
public R save(@ModelAttribute LiveMultimediaEntity liveMultimedia, MultipartFile mediaFile,@RequestParam(value = "limitSize", required = false, defaultValue = "500")Long limitSize){
if(limitSize * 1024 <= mediaFile.getSize()){
return R.error("图片大小必须控制在" + limitSize + "之内！");
}
return R.ok().put("info", liveMultimediaService.save(liveMultimedia, mediaFile, getUserId()));
}
/**
@ -148,7 +152,7 @@ public class LiveMultimediaController extends AbstractController{
// 线程处理
liveMultimediaSyncService.uploadMediaSync(identifier, filepath, (Integer) data.get("id"));
return R.ok().put("info", data);
}
/**
@ -162,7 +166,7 @@ public class LiveMultimediaController extends AbstractController{
liveMultimediaService.update(liveMultimedia, getUserId());
return R.ok();
}
/**
* 删除
*/

2
src/main/java/com/platform/modules/live/controller/MediaUploadController.java
View File

@ -114,7 +114,7 @@ public class MediaUploadController {
@PostMapping("/common/upload_media/chunk")
@RequiresPermissions(value = {"live:multimedia:save", "live:multimedia:update"}, logical = Logical.OR)
public R uploadChunk(PlmChunkEntity chunk,@RequestParam(value = "limitSize", required = false, defaultValue = "500")Long limitSize) {
public R uploadChunk(PlmChunkEntity chunk,@RequestParam(value = "limitSize", required = true)Long limitSize) {
if (null == chunk || null == chunk.getFile() || StringUtils.isBlank(chunk.getIdentifier())
|| null == chunk.getChunkNumber()) {
return R.error("缺少必要参数");

5
src/main/java/com/platform/modules/live/dto/ChannelDTO.java
View File

@ -1,5 +1,6 @@
package com.platform.modules.live.dto;
import com.platform.modules.live.entity.LiveSceneEntity;
import lombok.Data;
import java.util.List;
@ -10,6 +11,10 @@ public class ChannelDTO {
private String componentType; // 组件类型
private Integer componentCount; // 组件数量
private String titleSetting;
private Integer priority;
private Integer isDisplay;
private String componentLayout;
private String identify;
private List<LiveSceneDTO> liveScenes;
}

22
src/main/java/com/platform/modules/live/service/impl/LiveChannelServiceImpl.java
View File

@ -4,6 +4,7 @@ import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;
import com.baomidou.mybatisplus.mapper.Wrapper;
import com.platform.modules.live.constants.SceneConstants;
import com.platform.modules.live.dto.ChannelDTO;
import com.platform.modules.live.dto.LiveSceneDTO;
@ -11,6 +12,8 @@ import com.platform.modules.live.entity.LiveChannelComponentEntity;
import com.platform.modules.live.entity.LiveSceneChannelEntity;
import com.platform.modules.live.entity.LiveSceneEntity;
import com.platform.modules.live.service.LiveSceneService;
import com.platform.modules.live.utils.ForeignProperties;
import com.platform.modules.plm.util.MaterialUtils;
import com.platform.modules.sys.dao.SysConfigDao;
import com.platform.modules.sys.entity.SysConfigEntity;
import com.platform.modules.sys.service.impl.SysConfigServiceImpl;
@ -20,6 +23,7 @@ import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanInfoFactory;
import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import com.baomidou.mybatisplus.mapper.EntityWrapper;
@ -55,6 +59,9 @@ public class LiveChannelServiceImpl extends ServiceImpl<LiveChannelDao, LiveChan
@Autowired
private SysConfigDao sysConfigDao;
@Autowired
private ForeignProperties foreignProperties;
@Override
@Transactional(readOnly = true)
public PageUtils page(Map<String, Object> params){
@ -365,6 +372,8 @@ private LiveChannelDao liveChannelDao;
@Override
public HashMap<Object, Object> appView() {
HashMap<Object, Object> map = new HashMap<>();
//1.查询轮播大图
@ -384,7 +393,13 @@ private LiveChannelDao liveChannelDao;
map.put("carouselScene", liveScenes);
// 2.按照栏目封装现场
List<LiveChannelEntity> channelEntities = liveChannelService.selectList(null);
Wrapper<LiveChannelEntity> wrapper = new EntityWrapper<LiveChannelEntity>()
.eq("is_delete", 0)
.eq("is_display", 1)
.orderBy("priority", false);
List<LiveChannelEntity> channelEntities = liveChannelService.selectList(wrapper);
List<ChannelDTO> collect = channelEntities.stream().map(this::convertToB).collect(Collectors.toList());
// 为每个 ChannelDTO 获取 liveScenes
if(CollectionUtils.isNotEmpty(collect)){
@ -421,6 +436,7 @@ private LiveChannelDao liveChannelDao;
//2.将所有查询到的现场的channel_display字段设置为true
liveSceneEntities.forEach(liveSceneEntity -> {
liveChannelDao.updateChannelDisplay(liveSceneEntity.getId());
liveSceneEntity.setTitleImage( MaterialUtils.getMaterialUrl(liveSceneEntity.getTitleImage(), foreignProperties.getMediaurl()));
});
}
@ -490,6 +506,10 @@ private LiveChannelDao liveChannelDao;
channelDTO.setComponentCount(a.getComponentCount());
channelDTO.setComponentType(a.getComponentType());
channelDTO.setTitleSetting(a.getTitleSetting());
channelDTO.setPriority(a.getPriority());
channelDTO.setIsDisplay(a.getIsDisplay());
channelDTO.setComponentLayout(a.getComponentLayout());
channelDTO.setIdentify(a.getIdentify());
return channelDTO;
}
}

6
src/main/java/com/platform/modules/live/service/impl/LiveMultimediaServiceImpl.java
View File

@ -15,6 +15,7 @@ import java.util.*;
import javax.servlet.http.HttpServletResponse;
import com.aliyuncs.vod.model.v20170321.*;
import com.platform.modules.live.controller.ClipController;
import com.platform.plugs.live.handlers.IVideoClipsHandler;
import io.swagger.models.auth.In;
import org.apache.commons.lang3.StringUtils;
@ -105,6 +106,9 @@ public class LiveMultimediaServiceImpl extends ServiceImpl<LiveMultimediaDao, Li
@Autowired
private LiveSceneRecordService liveSceneRecordService;
@Autowired
private ClipController clipController;
UTCDateFormat df = new UTCDateFormat();
@Override
@ -932,7 +936,7 @@ public class LiveMultimediaServiceImpl extends ServiceImpl<LiveMultimediaDao, Li
sceneRecordEntity.setRecordType(5);
}
liveSceneRecordService.updateById(sceneRecordEntity);
clipController.sendClipFinishedEvent(sceneRecordEntity.getId());
}

2
src/main/java/com/platform/modules/live/service/impl/LiveReportServiceImpl.java
View File

@ -17,6 +17,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;
@ -698,6 +699,7 @@ public class LiveReportServiceImpl extends ServiceImpl<LiveReportDao, LiveReport
}
@Override
@Transactional(propagation = Propagation.REQUIRES_NEW, timeout = 5) // 每个更新独立事务
public void addLiveCountToDb(Integer reportId, long totalLikes) {
liveReportDao.addLiveCountToDb(reportId,totalLikes);
}

2
src/main/java/com/platform/modules/live/service/impl/LiveSceneCountServiceImpl.java
View File

@ -26,6 +26,7 @@ import com.platform.modules.live.utils.CalculateSceneCountUtils;
import com.platform.modules.live.vo.SceneCountVo;
import com.platform.modules.sys.entity.SysConfigEntity;
import com.platform.modules.sys.service.SysConfigService;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
@Service("liveSceneCountService")
@ -258,6 +259,7 @@ public class LiveSceneCountServiceImpl extends ServiceImpl<LiveSceneCountDao, Li
}
@Override
@Transactional(propagation = Propagation.REQUIRES_NEW, timeout = 5) // 每个更新独立事务
public void addLiveCountToDb(Integer sceneId, Long liveCount) {
liveSceneCountDao.addLiveCount(sceneId,liveCount);
}

2
src/main/java/com/platform/modules/live/service/impl/LiveSceneRecordServiceImpl.java
View File

@ -94,7 +94,7 @@ public class LiveSceneRecordServiceImpl extends ServiceImpl<LiveSceneRecordDao,
wrapper.orderBy("priority", false);
wrapper.orderBy("id", false);
// 20200306_Mxy 不显示m3u8类型的回顾
// wrapper.ne("record_type", LiveConstants.RecordType.M3U8_RECORD);
wrapper.ne("record_type", LiveConstants.RecordType.MP4_RECORD);
Page<LiveSceneRecordEntity> page = this.selectPage(
new Query<LiveSceneRecordEntity>(params).getPage(),
wrapper

30
src/main/java/com/platform/modules/live/service/impl/LiveSceneServiceImpl.java
View File

@ -2446,7 +2446,8 @@ public class LiveSceneServiceImpl extends ServiceImpl<LiveSceneDao, LiveSceneEnt
public R dragSort(Integer cid, Integer[] sceneIds, long[] dragSorts) {
List<LiveSceneChannelEntity> entityList = liveSceneChannelService.selectList(
new EntityWrapper<LiveSceneChannelEntity>().eq("channel_id", cid).in("scene_id", sceneIds)
new EntityWrapper<LiveSceneChannelEntity>()
.in("scene_id", sceneIds)
);
if (null == entityList || entityList.size() == 0)
@ -2767,14 +2768,17 @@ public class LiveSceneServiceImpl extends ServiceImpl<LiveSceneDao, LiveSceneEnt
LocalDateTime localDateTime = LocalDateTime.now();
LocalDateTime localDateTimePlusOne = localDateTime.plusMinutes(1);
EntityWrapper<LiveSceneEntity> wrapper = new EntityWrapper<>();
wrapper.eq("is_auto_live", SceneConstants.IsAutoLive.OPEN)
.eq("delete_flag", SceneConstants.IsDelete.NOTISDELETE)
.eq("scene_state", SceneConstants.SCENE_PROD)
.andNew("(live_state = {0} AND live_start <= {1}) OR (live_state = {2} AND live_end <= {3})",
SceneConstants.LIVE_STATE_NOTICE, new Date(),
SceneConstants.LIVE_STATE_DIRECT, new Date()
);
//改为数据库查询
List<LiveSceneEntity> liveSceneEntityList = this.baseMapper.selectList(new EntityWrapper<LiveSceneEntity>()
.eq("is_auto_live", SceneConstants.IsAutoLive.OPEN)
.eq("delete_flag",SceneConstants.IsDelete.NOTISDELETE).eq("scene_state", SceneConstants.SCENE_PROD)
.eq("live_state", SceneConstants.LIVE_STATE_NOTICE).le("live_start", localDateTimePlusOne)
.or().eq("is_auto_live", SceneConstants.IsAutoLive.OPEN)
.eq("delete_flag",SceneConstants.IsDelete.NOTISDELETE).eq("scene_state", SceneConstants.SCENE_PROD)
.eq("live_state", SceneConstants.LIVE_STATE_DIRECT).le("live_end", localDateTime));
List<LiveSceneEntity> liveSceneEntityList = this.baseMapper.selectList(wrapper);
List<LiveSceneEntity> result = new ArrayList<>();
if (CollectionUtil.isNotEmpty(liveSceneEntityList)) {
@ -2933,6 +2937,16 @@ public class LiveSceneServiceImpl extends ServiceImpl<LiveSceneDao, LiveSceneEnt
@Override
public List<LiveSceneDTO> appSearch(AppSearchDto appSearchDto) {
Page<LiveSceneDTO> page = new Page<>(appSearchDto.getPage(), appSearchDto.getSize());
// 筛选一天（例如，获取昨天的日期）
Calendar calendar = Calendar.getInstance();
calendar.setTime(new Date());
if(appSearchDto.getTimeState() ==1 || appSearchDto.getTimeState() ==7 || appSearchDto.getTimeState() ==30) {
calendar.add(Calendar.DAY_OF_MONTH, -1 * appSearchDto.getTimeState());
appSearchDto.setStartTime(calendar.getTime());
}
if(appSearchDto.getStartTime() != null && appSearchDto.getEndTime() == null) {
appSearchDto.setEndTime(new Date());
}

21
src/main/java/com/platform/modules/live/task/AutoLiveTask.java
View File

@ -5,6 +5,8 @@ import com.platform.common.utils.RedisUtils;
import com.platform.modules.live.constants.SceneConstants;
import com.platform.modules.live.entity.LiveSceneEntity;
import com.platform.modules.live.service.LiveSceneService;
import org.apache.shiro.SecurityUtils;
import org.quartz.DisallowConcurrentExecution;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@ -17,6 +19,7 @@ import java.util.List;
* @author Matt
* 2021/4/19 11:30
*/
@DisallowConcurrentExecution
@Component("autoLiveTask")
public class AutoLiveTask {
@ -28,17 +31,19 @@ public class AutoLiveTask {
@Autowired
private RedisUtils redisUtils;
public void AutoLiveTask(){
public void AutoLiveTask() {
synchronized (this.getClass()) {
logger.info("现场自动开关播定时刷新任务，刷新时间为：" + DateUtils.format(new Date(), DateUtils.DATE_TIME_PATTERN));
logger.info("现场自动开关播定时刷新任务，刷新时间为：" + DateUtils.format(new Date(), DateUtils.DATE_TIME_PATTERN));
List<LiveSceneEntity> dataList = liveSceneService.freshLiveStateByAutoLive();
List<LiveSceneEntity> dataList = liveSceneService.freshLiveStateByAutoLive();
if (dataList != null && dataList.size() > 0) {
dataList.forEach(e -> {
liveSceneService.createSceneIndex(e, null, null, null, null, null);
redisUtils.delete(SceneConstants.SCENE_REDIS_PREFIX + e.getId());
});
if (dataList != null && dataList.size() > 0) {
dataList.forEach(e -> {
liveSceneService.createSceneIndex(e, null, null, null, null, null);
redisUtils.delete(SceneConstants.SCENE_REDIS_PREFIX + e.getId());
});
}
}
}
}

83
src/main/java/com/platform/modules/live/task/LikeSyncTask.java
View File

@ -2,7 +2,9 @@ package com.platform.modules.live.task;
import com.platform.modules.live.service.LiveReportService;
import com.platform.modules.live.service.impl.LiveSceneCountServiceImpl;
import org.quartz.DisallowConcurrentExecution;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
@ -16,6 +18,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@Component
@DisallowConcurrentExecution
public class LikeSyncTask {
private static final Logger logger = LoggerFactory.getLogger(LikeSyncTask.class);
@ -29,58 +32,66 @@ public class LikeSyncTask {
private LiveReportService liveReportService;
public void syncLikesToDatabase() {
ScanOptions options = ScanOptions.scanOptions().match("scene:liveCount:*").build();
Cursor<byte[]> cursor = redisTemplate.getConnectionFactory()
.getConnection()
.scan(options);
RedisConnection connection = null;
try {
connection = redisTemplate.getConnectionFactory().getConnection();
ScanOptions options = ScanOptions.scanOptions().match("scene:liveCount:*").build();
Cursor<byte[]> cursor = connection.scan(options);
RedisSerializer<String> serializer = new StringRedisSerializer();
RedisSerializer<String> serializer = new StringRedisSerializer();
while (cursor.hasNext()) {
byte[] keyBytes = cursor.next();
String key = serializer.deserialize(keyBytes);
while (cursor.hasNext()) {
byte[] keyBytes = cursor.next();
String key = serializer.deserialize(keyBytes);
if (key != null) {
Integer sceneId = Integer.valueOf(key.split(":")[2]);
Object liveCount = redisTemplate.opsForValue().get(key);
if (liveCount != null) {
long totalLikes = Long.parseLong(liveCount.toString());
try {
liveSceneCountService.addLiveCountToDb(sceneId, totalLikes);
redisTemplate.delete(key);
} catch (Exception e) {
logger.error("Failed to sync likes for key: {}", key, e);
if (key != null) {
Integer sceneId = Integer.valueOf(key.split(":")[2]);
Object liveCount = redisTemplate.opsForValue().get(key);
if (liveCount != null) {
long totalLikes = Long.parseLong(liveCount.toString());
try {
liveSceneCountService.addLiveCountToDb(sceneId, totalLikes);
redisTemplate.delete(key);
} catch (Exception e) {
logger.error("Failed to sync likes for key: {}", key, e);
}
}
}
}
}finally {
if (connection != null) connection.close();
}
}
public void syncReportLikesToDatabase() {
ScanOptions options = ScanOptions.scanOptions().match("report:liveCount:*").build();
Cursor<byte[]> cursor = redisTemplate.getConnectionFactory()
.getConnection()
.scan(options);
RedisConnection connection = null;
try {
connection = redisTemplate.getConnectionFactory().getConnection();
ScanOptions options = ScanOptions.scanOptions().match("scene:liveCount:*").build();
Cursor<byte[]> cursor = connection.scan(options);
RedisSerializer<String> serializer = new StringRedisSerializer();
RedisSerializer<String> serializer = new StringRedisSerializer();
while (cursor.hasNext()) {
byte[] keyBytes = cursor.next();
String key = serializer.deserialize(keyBytes);
while (cursor.hasNext()) {
byte[] keyBytes = cursor.next();
String key = serializer.deserialize(keyBytes);
if (key != null) {
Integer reportId = Integer.valueOf(key.split(":")[2]);
Object liveCount = redisTemplate.opsForValue().get(key);
if (liveCount != null) {
long totalLikes = Long.parseLong(liveCount.toString());
try {
liveReportService.addLiveCountToDb(reportId,totalLikes);
redisTemplate.delete(key);
} catch (Exception e) {
logger.error("Failed to sync likes for key: {}", key, e);
if (key != null) {
Integer reportId = Integer.valueOf(key.split(":")[2]);
Object liveCount = redisTemplate.opsForValue().get(key);
if (liveCount != null) {
long totalLikes = Long.parseLong(liveCount.toString());
try {
liveReportService.addLiveCountToDb(reportId, totalLikes);
redisTemplate.delete(key);
} catch (Exception e) {
logger.error("Failed to sync likes for key: {}", key, e);
}
}
}
}
}finally {
if (connection != null) connection.close(); // 归还连接
}
}

19
src/main/java/com/platform/modules/sys/service/impl/SysConfigServiceImpl.java
View File

@ -13,6 +13,7 @@ import com.platform.modules.sys.redis.SysConfigRedis;
import com.platform.modules.sys.service.SysConfigService;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
@ -52,13 +53,25 @@ public class SysConfigServiceImpl extends ServiceImpl<SysConfigDao, SysConfigEnt
sysConfigRedis.saveOrUpdate(config);
}
//发送锁超时情况
@Override
@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
public void updateValueByKey(String key, String value) {
baseMapper.updateValueByKey(key, value);
sysConfigRedis.delete(key);
updateValueInTransaction(key, value);
asyncDeleteRedisKey(key);
}
// 独立事务仅用于数据库更新
@Transactional(rollbackFor = Exception.class)
public void updateValueInTransaction(String key, String value) {
baseMapper.updateValueByKey(key, value);
}
// 异步删除缓存（使用 Spring @Async）
@Async
public void asyncDeleteRedisKey(String key) {
sysConfigRedis.delete(key);
}
@Override
@Transactional(rollbackFor = Exception.class)
public void deleteBatch(Long[] ids) {

179
src/test/java/com/platform/service/sse_test.html Normal file
View File

@ -0,0 +1,179 @@
<!DOCTYPE html>
<html lang="zh">
<head>
<meta charset="UTF-8">
<title>SSE 测试页面</title>
<style>
body {
font-family: Arial, sans-serif;
margin: 20px;
}
.container {
max-width: 800px;
margin: 0 auto;
}
#messages {
border: 1px solid #ccc;
padding: 10px;
margin: 10px 0;
height: 400px;
overflow-y: auto;
background-color: #f9f9f9;
}
.message {
margin: 5px 0;
padding: 8px;
border-radius: 4px;
background-color: #fff;
border: 1px solid #e0e0e0;
}
.error {
color: #721c24;
background-color: #f8d7da;
border-color: #f5c6cb;
}
.success {
color: #155724;
background-color: #d4edda;
border-color: #c3e6cb;
}
.control-panel {
margin: 20px 0;
}
button {
padding: 8px 15px;
margin-right: 10px;
border: none;
border-radius: 4px;
cursor: pointer;
font-size: 14px;
}
button:hover {
opacity: 0.8;
}
#connect {
background-color: #28a745;
color: white;
}
#disconnect {
background-color: #dc3545;
color: white;
}
#clear {
background-color: #6c757d;
color: white;
}
#status {
padding: 10px;
margin: 10px 0;
border-radius: 4px;
background-color: #e9ecef;
}
</style>
</head>
<body>
<div class="container">
<h2>SSE 连接测试</h2>
<div class="control-panel">
<button id="connect" onclick="connectSSE()">连接 SSE</button>
<button id="disconnect" onclick="disconnectSSE()">断开连接</button>
<button id="clear" onclick="clearMessages()">清空消息</button>
</div>
<div id="status">状态: 未连接</div>
<div id="messages"></div>
</div>
<script>
let controller = null;
async function connectSSE() {
if (controller) {
console.log('已经存在连接，请先断开');
return;
}
const token = '9263b6dbab0ede5c12b2e8b414f82dd5';
const url = `http://10.0.0.25:8081/platform/status-stream?sceneId=123&token=${token}`;
controller = new AbortController();
try {
const response = await fetch(url, {
method: 'GET',
headers: {
'Accept': 'text/event-stream',
'Cache-Control': 'no-cache',
},
mode: 'cors',
signal: controller.signal
});
if (!response.ok) {
throw new Error(`HTTP error! status: ${response.status}`);
}
const reader = response.body.getReader();
document.getElementById('status').textContent = '状态: 已连接';
addMessage('连接已建立', 'success');
while (true) {
const {value, done} = await reader.read();
if (done) break;
const text = new TextDecoder().decode(value);
try {
const jsonData = JSON.parse(text);
for (const item of jsonData) {
if (item.data && item.data.startsWith('data:')) {
const messageData = item.data.substring(6).trim();
try {
const parsedMessage = JSON.parse(messageData);
const messageText = `类型: ${parsedMessage.recordType}, 消息: ${parsedMessage.message}`;
addMessage(messageText);
} catch (e) {
if (messageData) {
addMessage(messageData);
}
}
}
}
} catch (e) {
if (text.trim()) {
addMessage('解析错误: ' + text, 'error');
}
}
}
} catch (error) {
if (error.name === 'AbortError') {
addMessage('连接已手动断开');
} else {
addMessage('连接错误: ' + error.message, 'error');
document.getElementById('status').textContent = '状态: 连接错误';
}
} finally {
controller = null;
}
}
function disconnectSSE() {
if (controller) {
controller.abort();
controller = null;
document.getElementById('status').textContent = '状态: 已断开连接';
addMessage('连接已断开');
}
}
function addMessage(message, type = 'info') {
const messagesDiv = document.getElementById('messages');
const messageElement = document.createElement('div');
messageElement.className = `message ${type}`;
messageElement.textContent = `[${new Date().toLocaleTimeString()}] ${message}`;
messagesDiv.appendChild(messageElement);
messagesDiv.scrollTop = messagesDiv.scrollHeight;
}
function clearMessages() {
document.getElementById('messages').innerHTML = '';
}
</script>
</body>
</html>