update 统一管理BiliLiveDatabase

This commit is contained in:
Yutousama 2024-11-28 23:18:24 +08:00
parent 3f10e19e50
commit 8afb8f8f0b
8 changed files with 79 additions and 48 deletions

View File

@ -17,13 +17,16 @@ import com.yutou.biliapi.bean.websocket.live.WSData;
import com.yutou.biliapi.databases.BiliBiliLoginDatabase; import com.yutou.biliapi.databases.BiliBiliLoginDatabase;
import com.yutou.biliapi.databases.BiliLiveDatabase; import com.yutou.biliapi.databases.BiliLiveDatabase;
import com.yutou.biliapi.utils.BytesUtils; import com.yutou.biliapi.utils.BytesUtils;
import com.yutou.bilibili.services.LiveDatabasesService;
import com.yutou.common.okhttp.HttpBody; import com.yutou.common.okhttp.HttpBody;
import com.yutou.common.okhttp.HttpCallback; import com.yutou.common.okhttp.HttpCallback;
import com.yutou.common.utils.ConfigTools; import com.yutou.common.utils.ConfigTools;
import com.yutou.common.utils.Log; import com.yutou.common.utils.Log;
import jakarta.annotation.Resource;
import okhttp3.Headers; import okhttp3.Headers;
import org.java_websocket.client.WebSocketClient; import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake; import org.java_websocket.handshake.ServerHandshake;
import org.springframework.stereotype.Service;
import retrofit2.Response; import retrofit2.Response;
import java.io.ByteArrayOutputStream; import java.io.ByteArrayOutputStream;
@ -38,24 +41,20 @@ import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
public class WebSocketClientManager { @Service
public class WebSocketServer {
ThreadPoolExecutor executor; ThreadPoolExecutor executor;
private static WebSocketClientManager instance; private static WebSocketServer instance;
Map<LiveRoomConfig, DanmuTask> roomMap; Map<LiveRoomConfig, DanmuTask> roomMap;
private final List<String> userStopList = new ArrayList<>();//手动停止列表 private final List<String> userStopList = new ArrayList<>();//手动停止列表
@Resource
LiveDatabasesService liveDatabasesService;
private WebSocketClientManager() { private WebSocketServer() {
roomMap = new HashMap<>(); roomMap = new HashMap<>();
executor = new ThreadPoolExecutor(2, 4, Long.MAX_VALUE, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(100)); executor = new ThreadPoolExecutor(2, 4, Long.MAX_VALUE, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(100));
} }
public static WebSocketClientManager getInstance() {
if (instance == null) {
instance = new WebSocketClientManager();
}
return instance;
}
public boolean checkRoom(LiveRoomConfig roomConfig) { public boolean checkRoom(LiveRoomConfig roomConfig) {
return roomMap.containsKey(roomConfig); return roomMap.containsKey(roomConfig);
} }
@ -98,13 +97,13 @@ public class WebSocketClientManager {
} }
} }
private static class DanmuTask implements Runnable { private class DanmuTask implements Runnable {
LiveRoomConfig roomConfig; LiveRoomConfig roomConfig;
WebSocketClientTh client; WebSocketClientTh client;
public DanmuTask(LiveRoomConfig config) { public DanmuTask(LiveRoomConfig config) {
this.roomConfig = config; this.roomConfig = config;
WebSocketClientManager.getInstance().roomMap.put(roomConfig, this); roomMap.put(roomConfig, this);
} }
@Override @Override
@ -117,7 +116,7 @@ public class WebSocketClientManager {
roomConfig.setRoomInfo(execute.body() != null ? execute.body().getData() : null); roomConfig.setRoomInfo(execute.body() != null ? execute.body().getData() : null);
} }
} catch (IOException e) { } catch (IOException e) {
WebSocketClientManager.getInstance().roomMap.remove(roomConfig); roomMap.remove(roomConfig);
throw new RuntimeException(e); throw new RuntimeException(e);
} }
api.getLiveRoomDanmuInfo(String.valueOf(roomConfig.getRoomId())).enqueue(new HttpCallback<LiveDanmuInfo>() { api.getLiveRoomDanmuInfo(String.valueOf(roomConfig.getRoomId())).enqueue(new HttpCallback<LiveDanmuInfo>() {
@ -131,7 +130,7 @@ public class WebSocketClientManager {
roomConfig.setLiveInfo(response); roomConfig.setLiveInfo(response);
client = new WebSocketClientTh(new URI(url), roomConfig); client = new WebSocketClientTh(new URI(url), roomConfig);
} catch (URISyntaxException e) { } catch (URISyntaxException e) {
WebSocketClientManager.getInstance().roomMap.remove(roomConfig); roomMap.remove(roomConfig);
throw new RuntimeException(e); throw new RuntimeException(e);
} }
} }
@ -139,7 +138,7 @@ public class WebSocketClientManager {
@Override @Override
public void onFailure(Throwable throwable) { public void onFailure(Throwable throwable) {
WebSocketClientManager.getInstance().roomMap.remove(roomConfig); roomMap.remove(roomConfig);
Log.e(throwable); Log.e(throwable);
} }
}); });
@ -150,7 +149,7 @@ public class WebSocketClientManager {
} }
} }
private static class WebSocketClientTh extends WebSocketClient { private class WebSocketClientTh extends WebSocketClient {
private final LiveRoomConfig roomConfig; private final LiveRoomConfig roomConfig;
private final HeartbeatTask heartbeatTask; private final HeartbeatTask heartbeatTask;
BiliLiveDatabase liveDatabase; BiliLiveDatabase liveDatabase;
@ -159,7 +158,7 @@ public class WebSocketClientManager {
super(serverUri); super(serverUri);
Log.i("WebSocketClientTh.WebSocketClientTh : " + serverUri); Log.i("WebSocketClientTh.WebSocketClientTh : " + serverUri);
this.roomConfig = roomId; this.roomConfig = roomId;
liveDatabase = new BiliLiveDatabase(roomConfig); liveDatabase = liveDatabasesService.getLiveDatabase(roomConfig.getRoomId());
Brotli4jLoader.ensureAvailability(); Brotli4jLoader.ensureAvailability();
heartbeatTask = new HeartbeatTask(); heartbeatTask = new HeartbeatTask();
addHeader("User-Agent", ConfigTools.getUserAgent()); addHeader("User-Agent", ConfigTools.getUserAgent());
@ -190,8 +189,7 @@ public class WebSocketClientManager {
@Override @Override
public void onClose(int i, String s, boolean b) { public void onClose(int i, String s, boolean b) {
Log.e("WebSocketClientTh.onClose", "i = " + i + ", s = " + s + ", b = " + b, roomConfig.getRoomId(), heartbeatTask.socket.isOpen()); Log.e("WebSocketClientTh.onClose", "i = " + i + ", s = " + s + ", b = " + b, roomConfig.getRoomId(), heartbeatTask.socket.isOpen());
WebSocketClientManager.getInstance().roomMap.remove(roomConfig); roomMap.remove(roomConfig);
liveDatabase.close();
heartbeatTask.cancel(); heartbeatTask.cancel();
} }
@ -199,8 +197,7 @@ public class WebSocketClientManager {
public void onError(Exception e) { public void onError(Exception e) {
Log.i("WebSocketClientTh.onError", roomConfig.getRoomId()); Log.i("WebSocketClientTh.onError", roomConfig.getRoomId());
Log.e(e); Log.e(e);
WebSocketClientManager.getInstance().roomMap.remove(roomConfig); roomMap.remove(roomConfig);
liveDatabase.close();
heartbeatTask.cancel(); heartbeatTask.cancel();
} }

View File

@ -10,11 +10,12 @@ import com.yutou.biliapi.bean.live.database.LiveVideoDatabaseBean;
import com.yutou.biliapi.bean.websocket.live.WSData; import com.yutou.biliapi.bean.websocket.live.WSData;
import com.yutou.biliapi.databases.BiliLiveConfigDatabase; import com.yutou.biliapi.databases.BiliLiveConfigDatabase;
import com.yutou.biliapi.databases.BiliLiveDatabase; import com.yutou.biliapi.databases.BiliLiveDatabase;
import com.yutou.biliapi.net.WebSocketClientManager; import com.yutou.biliapi.net.WebSocketServer;
import com.yutou.bilibili.Tools.AssTools; import com.yutou.bilibili.Tools.AssTools;
import com.yutou.bilibili.Tools.Tools; import com.yutou.bilibili.Tools.Tools;
import com.yutou.bilibili.datas.web.LiveVideoDanmu; import com.yutou.bilibili.datas.web.LiveVideoDanmu;
import com.yutou.common.utils.Log; import com.yutou.common.utils.Log;
import jakarta.annotation.Resource;
import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
@ -23,31 +24,35 @@ import java.util.List;
@Service @Service
public class LiveDanmuService { public class LiveDanmuService {
@Resource
LiveDatabasesService liveDatabasesService;
@Resource
WebSocketServer webSocketServer;
public void start(String roomId, boolean isUser) { public void start(String roomId, boolean isUser) {
WebSocketClientManager.getInstance().addRoom(LiveRoomConfig.buildConfig(roomId), isUser); webSocketServer.addRoom(LiveRoomConfig.buildConfig(roomId), isUser);
} }
public void start(LiveConfigDatabaseBean roomId, boolean isUser) { public void start(LiveConfigDatabaseBean roomId, boolean isUser) {
WebSocketClientManager.getInstance().addRoom(LiveRoomConfig.buildConfig(roomId.getRoomId()), isUser); webSocketServer.addRoom(LiveRoomConfig.buildConfig(roomId.getRoomId()), isUser);
} }
public boolean check(String roomId) { public boolean check(String roomId) {
LiveRoomConfig roomConfig = new LiveRoomConfig(); LiveRoomConfig roomConfig = new LiveRoomConfig();
roomConfig.setRoomId(roomId); roomConfig.setRoomId(roomId);
return WebSocketClientManager.getInstance().checkRoom(roomConfig); return webSocketServer.checkRoom(roomConfig);
} }
public void stop(String roomId, boolean isUser) { public void stop(String roomId, boolean isUser) {
WebSocketClientManager.getInstance().stopRoom(roomId, isUser); webSocketServer.stopRoom(roomId, isUser);
} }
public JSONArray getLiveRoomList() { public JSONArray getLiveRoomList() {
return WebSocketClientManager.getInstance().getLiveRoomList(); return webSocketServer.getLiveRoomList();
} }
public void clearUserList() { public void clearUserList() {
WebSocketClientManager.getInstance().clearUserStopList(); webSocketServer.clearUserStopList();
} }
public List<File> getDanmuFileList(String roomId) { public List<File> getDanmuFileList(String roomId) {
@ -88,7 +93,7 @@ public class LiveDanmuService {
public LiveVideoDanmu getDanmu(String roomId, String videoId, int page) { public LiveVideoDanmu getDanmu(String roomId, String videoId, int page) {
LiveVideoDanmu danmus = new LiveVideoDanmu(); LiveVideoDanmu danmus = new LiveVideoDanmu();
BiliLiveDatabase liveDatabase = new BiliLiveDatabase(LiveRoomConfig.buildConfig(roomId)); BiliLiveDatabase liveDatabase = liveDatabasesService.getLiveDatabase(roomId);
try { try {
LiveVideoDatabaseBean videoBean = liveDatabase.getVideo(videoId); LiveVideoDatabaseBean videoBean = liveDatabase.getVideo(videoId);
if (videoBean == null) { if (videoBean == null) {
@ -121,8 +126,6 @@ public class LiveDanmuService {
} catch (Exception e) { } catch (Exception e) {
Log.e(e); Log.e(e);
} finally {
liveDatabase.close();
} }
return danmus; return danmus;

View File

@ -0,0 +1,21 @@
package com.yutou.bilibili.services;
import com.yutou.biliapi.bean.live.LiveRoomConfig;
import com.yutou.biliapi.databases.BiliLiveDatabase;
import org.springframework.stereotype.Service;
import java.util.HashMap;
import java.util.Map;
@Service
public class LiveDatabasesService {
private final Map<String, BiliLiveDatabase> liveDatabases=new HashMap<>();
public BiliLiveDatabase getLiveDatabase(String roomId) {
if(liveDatabases.containsKey(roomId)) {
return liveDatabases.get(roomId);
}
BiliLiveDatabase liveDatabase = new BiliLiveDatabase(LiveRoomConfig.buildConfig(roomId));
liveDatabases.put(roomId, liveDatabase);
return liveDatabase;
}
}

View File

@ -33,8 +33,11 @@ public class LiveService {
LiveVideoDownloadService videoDownloadService; LiveVideoDownloadService videoDownloadService;
@Resource @Resource
LiveDanmuService danmuService; LiveDanmuService danmuService;
@Resource
LiveDatabasesService databasesService;
LiveApi api; LiveApi api;
public LiveService() { public LiveService() {
liveConfigDatabase = new BiliLiveConfigDatabase(); liveConfigDatabase = new BiliLiveConfigDatabase();
api = BiliLiveNetApiManager.getInstance().getApi(null); api = BiliLiveNetApiManager.getInstance().getApi(null);
@ -108,7 +111,7 @@ public class LiveService {
} }
public JSONObject getGiftInfo(String roomId, String videoId) { public JSONObject getGiftInfo(String roomId, String videoId) {
BiliLiveDatabase database = new BiliLiveDatabase(LiveRoomConfig.buildConfig(roomId)); BiliLiveDatabase database = databasesService.getLiveDatabase(roomId);
LiveVideoDatabaseBean videoBean = database.getVideo(videoId); LiveVideoDatabaseBean videoBean = database.getVideo(videoId);
if (videoBean == null) { if (videoBean == null) {
return null; return null;
@ -124,4 +127,6 @@ public class LiveService {
List<LiveData> data = service.getLiveList(1, 16); List<LiveData> data = service.getLiveList(1, 16);
System.out.println(data.size()); System.out.println(data.size());
} }
} }

View File

@ -18,7 +18,7 @@ import com.yutou.biliapi.enums.LiveVideoCodec;
import com.yutou.biliapi.enums.LiveVideoDefinition; import com.yutou.biliapi.enums.LiveVideoDefinition;
import com.yutou.biliapi.enums.LiveVideoFormat; import com.yutou.biliapi.enums.LiveVideoFormat;
import com.yutou.biliapi.net.BiliLiveNetApiManager; import com.yutou.biliapi.net.BiliLiveNetApiManager;
import com.yutou.biliapi.net.WebSocketClientManager; import com.yutou.biliapi.net.WebSocketServer;
import com.yutou.bilibili.Tools.DateFormatUtils; import com.yutou.bilibili.Tools.DateFormatUtils;
import com.yutou.bilibili.Tools.FileServerUtils; import com.yutou.bilibili.Tools.FileServerUtils;
import com.yutou.bilibili.datas.VideoFilePath; import com.yutou.bilibili.datas.VideoFilePath;
@ -29,6 +29,7 @@ import com.yutou.common.record.AbsVideoRecord;
import com.yutou.common.utils.ConfigTools; import com.yutou.common.utils.ConfigTools;
import com.yutou.common.utils.FFmpegUtils; import com.yutou.common.utils.FFmpegUtils;
import com.yutou.common.utils.Log; import com.yutou.common.utils.Log;
import jakarta.annotation.Resource;
import okhttp3.Headers; import okhttp3.Headers;
import org.apache.commons.io.FileUtils; import org.apache.commons.io.FileUtils;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
@ -48,6 +49,10 @@ public class LiveVideoDownloadService {
private final ThreadPoolExecutor executor; private final ThreadPoolExecutor executor;
private final List<String> userStopList = new ArrayList<>();//手动停止列表 private final List<String> userStopList = new ArrayList<>();//手动停止列表
private final AbsVideoRecord videoRecord; private final AbsVideoRecord videoRecord;
@Resource
LiveDatabasesService liveDatabasesService;
@Resource
WebSocketServer webSocketServer;
public LiveVideoDownloadService() { public LiveVideoDownloadService() {
Log.i("初始化下载服务"); Log.i("初始化下载服务");
@ -176,7 +181,7 @@ public class LiveVideoDownloadService {
config.setLogin(StringUtils.hasText(bean.getRecordUid())); config.setLogin(StringUtils.hasText(bean.getRecordUid()));
config.setRoomInfo(roomInfo); config.setRoomInfo(roomInfo);
config.setRootPath(bean.getRecordPath()); config.setRootPath(bean.getRecordPath());
database = new BiliLiveDatabase(config); database = liveDatabasesService.getLiveDatabase(bean.getRoomId());
saveLiveInfo(roomInfo); saveLiveInfo(roomInfo);
api.getLiveRoomPlayInfo( api.getLiveRoomPlayInfo(
bean.getRoomId(), bean.getRoomId(),
@ -355,14 +360,14 @@ public class LiveVideoDownloadService {
//录制弹幕 //录制弹幕
private void recordDanmu() { private void recordDanmu() {
if (bean.isSyncDanmuForLive() && !WebSocketClientManager.getInstance().checkRoom(LiveRoomConfig.buildConfig(bean.getRoomId()))) { if (bean.isSyncDanmuForLive() && !webSocketServer.checkRoom(LiveRoomConfig.buildConfig(bean.getRoomId()))) {
WebSocketClientManager.getInstance().addRoom(LiveRoomConfig.buildConfig(bean.getRoomId()), true); webSocketServer.addRoom(LiveRoomConfig.buildConfig(bean.getRoomId()), true);
} }
} }
private void stopRecordDanmu() { private void stopRecordDanmu() {
if (bean.isSyncDanmuForLive() && WebSocketClientManager.getInstance().checkRoom(LiveRoomConfig.buildConfig(bean.getRoomId()))) { if (bean.isSyncDanmuForLive() && webSocketServer.checkRoom(LiveRoomConfig.buildConfig(bean.getRoomId()))) {
WebSocketClientManager.getInstance().stopRoom(bean.getRoomId(), false); webSocketServer.stopRoom(bean.getRoomId(), false);
} }
} }
} }
@ -381,13 +386,11 @@ public class LiveVideoDownloadService {
private VideoFilePath getVideoFilePath(LiveConfigDatabaseBean configBean) { private VideoFilePath getVideoFilePath(LiveConfigDatabaseBean configBean) {
String recordPath = configBean.getRecordPath() + File.separator + configBean.getAnchorName(); String recordPath = configBean.getRecordPath() + File.separator + configBean.getAnchorName();
File recordDir = new File(recordPath); File recordDir = new File(recordPath);
BiliLiveDatabase database = new BiliLiveDatabase(LiveRoomConfig.buildConfig(configBean.getRoomId()), recordDir.getAbsolutePath());
VideoFilePath path = createVideoRootFilePath(configBean, recordDir); VideoFilePath path = createVideoRootFilePath(configBean, recordDir);
if (recordDir.exists()) { if (recordDir.exists()) {
List<LiveVideoDatabaseBean> infos = database.getLiveInfos(); List<LiveVideoDatabaseBean> infos = liveDatabasesService.getLiveDatabase(configBean.getRoomId()).getLiveInfos();
path.setChildren(getVideoInfo(infos)); path.setChildren(getVideoInfo(infos));
} }
database.close();
return path; return path;
} }
@ -430,14 +433,12 @@ public class LiveVideoDownloadService {
String recordPath = config.getRecordPath() + File.separator + config.getAnchorName(); String recordPath = config.getRecordPath() + File.separator + config.getAnchorName();
configDatabase.close(); configDatabase.close();
LiveVideoDatabaseBean videoInfo = null; LiveVideoDatabaseBean videoInfo = null;
BiliLiveDatabase liveDatabase = new BiliLiveDatabase(LiveRoomConfig.buildConfig(roomId), new File(recordPath).getAbsolutePath()); for (LiveVideoDatabaseBean info : liveDatabasesService.getLiveDatabase(roomId).getLiveInfos()) {
for (LiveVideoDatabaseBean info : liveDatabase.getLiveInfos()) {
if (videoId.trim().equals(String.valueOf(info.getSql_time().getTime()))) { if (videoId.trim().equals(String.valueOf(info.getSql_time().getTime()))) {
videoInfo = info; videoInfo = info;
break; break;
} }
} }
liveDatabase.close();
if (videoInfo != null) { if (videoInfo != null) {
File videoFile = new File(videoInfo.getPath().replace("-%04d.ts", ".m3u8")); File videoFile = new File(videoInfo.getPath().replace("-%04d.ts", ".m3u8"));
if (!videoFile.exists()) { if (!videoFile.exists()) {

View File

@ -22,6 +22,7 @@ public class SystemService {
LiveVideoDownloadService videoService; LiveVideoDownloadService videoService;
@Resource @Resource
LiveDanmuService danmuService; LiveDanmuService danmuService;
SystemConfigDatabases databases = new SystemConfigDatabases(); SystemConfigDatabases databases = new SystemConfigDatabases();
private ScheduledExecutorService timer; private ScheduledExecutorService timer;
private ScheduledFuture<?> scheduled; private ScheduledFuture<?> scheduled;

View File

@ -27,7 +27,7 @@ public class RedisTools {
//Properties properties = PropertyUtil.loadProperties("jedis.properties"); //Properties properties = PropertyUtil.loadProperties("jedis.properties");
//host = properties.getProperty("redis.host"); //host = properties.getProperty("redis.host");
//port = Integer.valueOf(properties.getProperty("redis.port")); //port = Integer.valueOf(properties.getProperty("redis.port"));
host = "172.21.35.118"; host = "192.168.31.148";
port = 6379; port = 6379;
} }
@ -41,7 +41,6 @@ public class RedisTools {
String ret = jedis.set(key, value); String ret = jedis.set(key, value);
jedis.close(); jedis.close();
} catch (Exception e) { } catch (Exception e) {
// TODO: handle exception
Log.e(e); Log.e(e);
return false; return false;
} }
@ -71,7 +70,6 @@ public class RedisTools {
} }
jedis.close(); jedis.close();
} catch (Exception e) { } catch (Exception e) {
// TODO: handle exception
Log.e(e); Log.e(e);
return false; return false;
} }

View File

@ -1,3 +1,8 @@
server.port=8080 server.port=8880
logging.file.path=./logs logging.file.path=./logs
logging.level.com.log.controller = trace logging.level.com.log.controller = trace
# 启用Hibernate SQL日志
spring.jpa.show-sql=true
spring.jpa.properties.hibernate.format_sql=true
logging.level.org.hibernate.SQL=DEBUG
logging.level.org.hibernate.type.descriptor.sql.BasicBinder=TRACE