diff --git a/conf/srs.conf b/conf/srs.conf new file mode 100644 index 0000000..1311672 --- /dev/null +++ b/conf/srs.conf @@ -0,0 +1,28 @@ +listen 1935; +max_connections 1000; +daemon off; +http_api { + enabled on; + listen 1985; +} +http_server { + enabled on; + listen 8080; +} +vhost __defaultVhost__ { + hls { + enabled on; + hls_path ./objs/nginx/html; + hls_fragment 10; + hls_window 60; + } + http_remux { + enabled on; + mount [vhost]/[app]/[stream].flv; + } + dvr { + enabled on; + dvr_path ./recordings/[app]/[stream].flv; + dvr_plan session; + } +} \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..12f22de --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,11 @@ +version: '3' +services: + srs: + image: ossrs/srs:4 + ports: + - "1935:1935" # RTMP + - "8080:8080" # HTTP-FLV + - "8088:8088" # HLS + volumes: + - ./conf/srs.conf:/usr/local/srs/conf/srs.conf + - ./recordings:/usr/local/srs/recordings \ No newline at end of file diff --git a/pom.xml b/pom.xml index de918dd..b981831 100644 --- a/pom.xml +++ b/pom.xml @@ -223,6 +223,26 @@ + + + org.springframework.boot + spring-boot-starter-websocket + + + + + com.github.ossrs + srs-sdk + 1.0.0 + + + + + io.netty + netty-all + 4.1.94.Final + + diff --git a/src/main/java/com/guwan/backend/controller/LiveController.java b/src/main/java/com/guwan/backend/controller/LiveController.java new file mode 100644 index 0000000..85ae99c --- /dev/null +++ b/src/main/java/com/guwan/backend/controller/LiveController.java @@ -0,0 +1,50 @@ +package com.guwan.backend.controller; + +import com.guwan.backend.entity.LiveRoom; +import com.guwan.backend.entity.LiveRoomDTO; +import com.guwan.backend.service.LiveService; +import com.guwan.backend.util.Result; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.web.bind.annotation.*; + +@Slf4j +@RestController +@RequestMapping("/api/live") +@RequiredArgsConstructor +public class LiveController { + + private final LiveService liveService; + + @PostMapping("/room") + public Result createLiveRoom(@RequestBody LiveRoomDTO dto) { + try { + return Result.success(liveService.createLiveRoom(dto)); + } catch (Exception e) { + log.error("创建直播间失败", e); + return Result.error(e.getMessage()); + } + } + + @PostMapping("/room/{id}/start") + public Result startLive(@PathVariable Long id) { + try { + liveService.startLive(id); + return Result.success(); + } catch (Exception e) { + log.error("开始直播失败", e); + return Result.error(e.getMessage()); + } + } + + @PostMapping("/room/{id}/end") + public Result endLive(@PathVariable Long id) { + try { + liveService.endLive(id); + return Result.success(); + } catch (Exception e) { + log.error("结束直播失败", e); + return Result.error(e.getMessage()); + } + } +} \ No newline at end of file diff --git a/src/main/java/com/guwan/backend/entity/LiveMessage.java b/src/main/java/com/guwan/backend/entity/LiveMessage.java new file mode 100644 index 0000000..9475b7c --- /dev/null +++ b/src/main/java/com/guwan/backend/entity/LiveMessage.java @@ -0,0 +1,10 @@ +@Data +public class LiveMessage { + private String type; // 消息类型：CHAT-聊天，GIFT-礼物，LIKE-点赞，ENTER-进入，LEAVE-离开 + private Long userId; // 用户ID + private String username; // 用户名 + private String content; // 消息内容 + private String giftType; // 礼物类型 + private Integer giftCount; // 礼物数量 + private LocalDateTime time; // 时间 +} \ No newline at end of file diff --git a/src/main/java/com/guwan/backend/entity/LiveRoom.java b/src/main/java/com/guwan/backend/entity/LiveRoom.java new file mode 100644 index 0000000..02c319b --- /dev/null +++ b/src/main/java/com/guwan/backend/entity/LiveRoom.java @@ -0,0 +1,24 @@ +@Data +@TableName("live_room") +public class LiveRoom { + @TableId(type = IdType.AUTO) + private Long id; + + private String title; // 直播间标题 + private String description; // 直播间描述 + private String coverUrl; // 封面图 + private Long userId; // 主播ID + private String username; // 主播名称 + private String streamKey; // 推流密钥 + private String rtmpUrl; // RTMP推流地址 + private String hlsUrl; // HLS播放地址 + private String status; // 状态：PREPARING-准备中，LIVING-直播中，ENDED-已结束 + private Integer onlineCount; // 在线人数 + private Integer likeCount; // 点赞数 + + @TableField(fill = FieldFill.INSERT) + private LocalDateTime createdTime; + + @TableField(fill = FieldFill.INSERT_UPDATE) + private LocalDateTime updatedTime; +} \ No newline at end of file diff --git a/src/main/java/com/guwan/backend/service/LiveService.java b/src/main/java/com/guwan/backend/service/LiveService.java new file mode 100644 index 0000000..07526fc --- /dev/null +++ b/src/main/java/com/guwan/backend/service/LiveService.java @@ -0,0 +1,80 @@ +@Slf4j +@Service +@RequiredArgsConstructor +public class LiveService { + + private final LiveRoomMapper liveRoomMapper; + private final SrsClient srsClient; + + /** + * 创建直播间 + */ + public LiveRoom createLiveRoom(LiveRoomDTO dto) { + LiveRoom room = new LiveRoom(); + BeanUtils.copyProperties(dto, room); + + // 生成推流密钥 + String streamKey = generateStreamKey(); + room.setStreamKey(streamKey); + + // 生成推流地址 + String rtmpUrl = generateRtmpUrl(streamKey); + room.setRtmpUrl(rtmpUrl); + + // 生成播放地址 + String hlsUrl = generateHlsUrl(streamKey); + room.setHlsUrl(hlsUrl); + + room.setStatus("PREPARING"); + room.setOnlineCount(0); + room.setLikeCount(0); + + liveRoomMapper.insert(room); + return room; + } + + /** + * 开始直播 + */ + public void startLive(Long roomId) { + LiveRoom room = liveRoomMapper.selectById(roomId); + room.setStatus("LIVING"); + liveRoomMapper.updateById(room); + + // 开始录制 + startRecording(room.getStreamKey()); + } + + /** + * 结束直播 + */ + public void endLive(Long roomId) { + LiveRoom room = liveRoomMapper.selectById(roomId); + room.setStatus("ENDED"); + liveRoomMapper.updateById(room); + + // 停止录制 + stopRecording(room.getStreamKey()); + + // 生成回放地址 + String replayUrl = generateReplayUrl(room.getStreamKey()); + room.setReplayUrl(replayUrl); + liveRoomMapper.updateById(room); + } + + /** + * 开始录制 + */ + private void startRecording(String streamKey) { + srsClient.startRecord(streamKey, new RecordConfig() + .setFormat("flv") + .setFilePath("/recordings/" + streamKey + ".flv")); + } + + /** + * 停止录制 + */ + private void stopRecording(String streamKey) { + srsClient.stopRecord(streamKey); + } +} \ No newline at end of file diff --git a/src/main/java/com/guwan/backend/websocket/LiveWebSocketHandler.java b/src/main/java/com/guwan/backend/websocket/LiveWebSocketHandler.java new file mode 100644 index 0000000..f29a10c --- /dev/null +++ b/src/main/java/com/guwan/backend/websocket/LiveWebSocketHandler.java @@ -0,0 +1,82 @@ +package com.guwan.backend.websocket; + +import com.alibaba.fastjson.JSON; +import com.guwan.backend.entity.LiveMessage; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.springframework.web.socket.CloseStatus; +import org.springframework.web.socket.TextMessage; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.handler.TextWebSocketHandler; + +import java.io.IOException; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentHashSet; + +@Slf4j +@Component +public class LiveWebSocketHandler extends TextWebSocketHandler { + + private static final Map> roomSessions = new ConcurrentHashMap<>(); + private static final Map userSessions = new ConcurrentHashMap<>(); + + @Override + public void afterConnectionEstablished(WebSocketSession session) { + String roomId = getRoomId(session); + String userId = getUserId(session); + + // 加入房间 + roomSessions.computeIfAbsent(roomId, k -> new ConcurrentHashSet<>()).add(session); + userSessions.put(userId, session); + + // 广播进入消息 + broadcastMessage(roomId, createEnterMessage(userId)); + } + + @Override + public void handleTextMessage(WebSocketSession session, TextMessage message) { + String roomId = getRoomId(session); + LiveMessage liveMessage = JSON.parseObject(message.getPayload(), LiveMessage.class); + + // 处理不同类型的消息 + switch (liveMessage.getType()) { + case "CHAT": + broadcastMessage(roomId, message); + break; + case "GIFT": + handleGiftMessage(roomId, liveMessage); + break; + case "LIKE": + handleLikeMessage(roomId, liveMessage); + break; + } + } + + @Override + public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { + String roomId = getRoomId(session); + String userId = getUserId(session); + + // 离开房间 + roomSessions.get(roomId).remove(session); + userSessions.remove(userId); + + // 广播离开消息 + broadcastMessage(roomId, createLeaveMessage(userId)); + } + + private void broadcastMessage(String roomId, TextMessage message) { + Set sessions = roomSessions.get(roomId); + if (sessions != null) { + sessions.forEach(session -> { + try { + session.sendMessage(message); + } catch (IOException e) { + log.error("发送消息失败", e); + } + }); + } + } +} \ No newline at end of file