IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Java直播系统案例

wen java案例 3

本文目录导读：

Java直播系统案例

  1. 系统架构
  2. 核心技术栈
  3. 核心代码实现
  4. 数据库设计
  5. 使用说明
  6. 系统优化建议

我将为您提供一个完整的Java直播系统案例,包括核心功能模块和代码实现。

系统架构

直播系统
├── 用户模块（登录、注册、个人信息）
├── 直播模块（开播、观看、互动）
├── 礼物系统
├── 聊天系统
├── 支付系统
└── 管理后台

核心技术栈

  • 后端：Spring Boot + MyBatis Plus
  • 实时通信：Netty + WebSocket
  • 数据库：MySQL + Redis
  • 视频处理：FFmpeg + HLS
  • 前端：Vue.js + Element UI

核心代码实现

项目结构

com.live.broadcast/
├── controller/
│   ├── UserController.java
│   ├── LiveController.java
│   └── GiftController.java
├── service/
│   ├── UserService.java
│   ├── LiveService.java
│   └── MessageService.java
├── netty/
│   ├── NettyServer.java
│   └── WebSocketHandler.java
├── entity/
│   ├── User.java
│   ├── LiveRoom.java
│   └── GiftRecord.java
└── config/
    └── MybatisPlusConfig.java

用户模块

@RestController
@RequestMapping("/api/user")
public class UserController {
    @Autowired
    private UserService userService;
    @PostMapping("/login")
    public Result login(@RequestBody LoginRequest request) {
        String token = userService.login(request);
        return Result.success(token);
    }
    @PostMapping("/register")
    public Result register(@RequestBody User user) {
        userService.register(user);
        return Result.success();
    }
    @GetMapping("/info/{userId}")
    public Result getUserInfo(@PathVariable Long userId) {
        User user = userService.getUserInfo(userId);
        return Result.success(user);
    }
}
@Service
public class UserServiceImpl implements UserService {
    @Autowired
    private UserMapper userMapper;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Override
    public String login(LoginRequest request) {
        // 验证用户信息
        QueryWrapper<User> wrapper = new QueryWrapper<>();
        wrapper.eq("username", request.getUsername());
        wrapper.eq("password", MD5Util.encrypt(request.getPassword()));
        User user = userMapper.selectOne(wrapper);
        if (user == null) {
            throw new BusinessException("用户名或密码错误");
        }
        // 生成token并存储到Redis
        String token = UUID.randomUUID().toString().replace("-", "");
        redisTemplate.opsForValue().set("token:" + token, user.getId(), 24, TimeUnit.HOURS);
        return token;
    }
    @Override
    public void updateUserStatus(Long userId, boolean online) {
        // 更新用户在线状态
        User user = userMapper.selectById(userId);
        user.setOnline(online);
        userMapper.updateById(user);
    }
}

直播模块

@RestController
@RequestMapping("/api/live")
public class LiveController {
    @Autowired
    private LiveService liveService;
    @PostMapping("/start")
    public Result startLive(@RequestBody StartLiveRequest request) {
        LiveRoom room = liveService.startLive(request);
        return Result.success(room);
    }
    @PostMapping("/stop/{roomId}")
    public Result stopLive(@PathVariable Long roomId) {
        liveService.stopLive(roomId);
        return Result.success();
    }
    @GetMapping("/info/{roomId}")
    public Result getLiveInfo(@PathVariable Long roomId) {
        LiveRoom room = liveService.getLiveInfo(roomId);
        return Result.success(room);
    }
    @GetMapping("/hot")
    public Result getHotLives() {
        List<LiveRoom> rooms = liveService.getHotLives();
        return Result.success(rooms);
    }
}
@Service
public class LiveServiceImpl implements LiveService {
    @Autowired
    private LiveRoomMapper liveRoomMapper;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Override
    @Transactional
    public LiveRoom startLive(StartLiveRequest request) {
        // 生成推流地址
        String pushUrl = generatePushUrl(request.getUserId());
        // 创建直播房间
        LiveRoom room = new LiveRoom();
        room.setAnchorId(request.getUserId());
        room.setTitle(request.getTitle());
        room.setCoverUrl(request.getCoverUrl());
        room.setPushUrl(pushUrl);
        room.setStreamKey(UUID.randomUUID().toString().replace("-", ""));
        room.setStatus(1); // 1-直播中
        room.setCreateTime(LocalDateTime.now());
        liveRoomMapper.insert(room);
        // 缓存到Redis
        String roomKey = "live_room:" + room.getId();
        redisTemplate.opsForValue().set(roomKey, room, 4, TimeUnit.HOURS);
        // 异步启动FFmpeg推流
        startFFmpegProcess(room);
        return room;
    }
    @Override
    public void stopLive(Long roomId) {
        LiveRoom room = liveRoomMapper.selectById(roomId);
        room.setStatus(0);
        room.setEndTime(LocalDateTime.now());
        liveRoomMapper.updateById(room);
        // 清除Redis缓存
        redisTemplate.delete("live_room:" + roomId);
        // 停止FFmpeg进程
        stopFFmpegProcess(roomId);
    }
    private String generatePushUrl(Long userId) {
        String rtmpHost = "rtmp://live.example.com/live/";
        return rtmpHost + userId + "_" + System.currentTimeMillis();
    }
}

实时通信（Netty + WebSocket）

@Component
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    private static final ConcurrentHashMap<Long, Channel> USER_CHANNELS = new ConcurrentHashMap<>();
    @Autowired
    private MessageService messageService;
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) {
        String message = frame.text();
        JSONObject jsonObject = JSON.parseObject(message);
        switch (jsonObject.getInteger("type")) {
            case 1: // 聊天消息
                handleChatMessage(ctx, jsonObject);
                break;
            case 2: // 礼物消息
                handleGiftMessage(ctx, jsonObject);
                break;
            case 3: // 进入直播间
                handleEnterRoom(ctx, jsonObject);
                break;
            case 4: // 退出直播间
                handleExitRoom(ctx, jsonObject);
                break;
        }
    }
    private void handleChatMessage(ChannelHandlerContext ctx, JSONObject message) {
        Long roomId = message.getLong("roomId");
        Long userId = message.getLong("userId");
        String content = message.getString("content");
        // 存储消息到Redis
        ChatMessage chatMessage = new ChatMessage();
        chatMessage.setRoomId(roomId);
        chatMessage.setUserId(userId);
        chatMessage.setContent(content);
        chatMessage.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern("HH:mm:ss")));
        // 广播给直播间所有用户
        broadcastToRoom(roomId, JSON.toJSONString(chatMessage));
    }
    private void handleGiftMessage(ChannelHandlerContext ctx, JSONObject message) {
        Long roomId = message.getLong("roomId");
        Long userId = message.getLong("userId");
        String giftId = message.getString("giftId");
        // 验证用户余额和礼物信息
        boolean success = messageService.sendGift(roomId, userId, giftId);
        if (success) {
            GiftMessage giftMessage = new GiftMessage();
            giftMessage.setType("gift");
            giftMessage.setUserId(userId);
            giftMessage.setGiftId(giftId);
            giftMessage.setContent("赠送了礼物");
            // 广播礼物消息
            broadcastToRoom(roomId, JSON.toJSONString(giftMessage));
        }
    }
    private void broadcastToRoom(Long roomId, String message) {
        for (Map.Entry<Long, Channel> entry : USER_CHANNELS.entrySet()) {
            // 这里需要维护roomId与Channel的映射关系
            if (entry.getValue().isActive()) {
                entry.getValue().writeAndFlush(new TextWebSocketFrame(message));
            }
        }
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        ctx.close();
    }
}

礼物系统

@Service
public class GiftServiceImpl implements GiftService {
    @Autowired
    private GiftMapper giftMapper;
    @Autowired
    private UserService userService;
    @Override
    @Transactional
    public boolean sendGift(Long roomId, Long userId, String giftId) {
        // 获取礼物信息
        Gift gift = giftMapper.selectById(giftId);
        if (gift == null) {
            return false;
        }
        // 扣减用户余额
        User user = userService.getUserById(userId);
        if (user.getBalance() < gift.getPrice()) {
            throw new BusinessException("余额不足");
        }
        // 记录礼物消费记录
        GiftRecord record = new GiftRecord();
        record.setUserId(userId);
        record.setRoomId(roomId);
        record.setGiftId(giftId);
        record.setPrice(gift.getPrice());
        record.setTime(LocalDateTime.now());
        // 保存记录并扣款
        giftRecordMapper.insert(record);
        user.setBalance(user.getBalance() - gift.getPrice());
        userService.updateUser(user);
        // 更新主播收入
        LiveRoom room = liveRoomMapper.selectById(roomId);
        if (room != null) {
            User anchor = userService.getUserById(room.getAnchorId());
            anchor.setIncome(anchor.getIncome() + gift.getPrice() * 0.7);
            userService.updateUser(anchor);
        }
        return true;
    }
}
@Entity
@Table(name = "gift")
public class Gift {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String name;
    private BigDecimal price;
    private String iconUrl;
    private boolean isVip;
    // getter/setter...
}

配置类

@Configuration
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
    @Bean
    public NettyServer nettyServer() {
        return new NettyServer(8088);
    }
}
@Component
public class NettyServer {
    private final int port;
    private final ServerBootstrap bootstrap;
    public NettyServer(int port) {
        this.port = port;
        this.bootstrap = new ServerBootstrap();
    }
    @PostConstruct
    public void start() throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        bootstrap.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .option(ChannelOption.SO_BACKLOG, 128)
            .childOption(ChannelOption.SO_KEEPALIVE, true)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    ch.pipeline()
                        .addLast(new HttpServerCodec())
                        .addLast(new HttpObjectAggregator(65536))
                        .addLast(new WebSocketServerProtocolHandler("/ws"))
                        .addLast(new WebSocketHandler());
                }
            });
        bootstrap.bind(port).sync();
        log.info("Netty server started on port: {}", port);
    }
}

数据库设计

-- 用户表
CREATE TABLE `user` (
    `id` bigint NOT NULL AUTO_INCREMENT,
    `username` varchar(50) NOT NULL,
    `password` varchar(100) NOT NULL,
    `nickname` varchar(50) DEFAULT NULL,
    `avatar` varchar(200) DEFAULT NULL,
    `balance` decimal(10,2) DEFAULT '0.00',
    `income` decimal(10,2) DEFAULT '0.00',
    `level` int DEFAULT '1',
    `online` tinyint(1) DEFAULT '0',
    PRIMARY KEY (`id`)
);
-- 直播间表
CREATE TABLE `live_room` (
    `id` bigint NOT NULL AUTO_INCREMENT,
    `anchor_id` bigint NOT NULL, varchar(100) NOT NULL,
    `cover_url` varchar(200) DEFAULT NULL,
    `status` tinyint DEFAULT '0',
    `push_url` varchar(200) DEFAULT NULL,
    `stream_key` varchar(100) DEFAULT NULL,
    `viewer_count` int DEFAULT '0',
    `create_time` datetime DEFAULT NULL,
    `end_time` datetime DEFAULT NULL,
    PRIMARY KEY (`id`)
);
-- 礼物表
CREATE TABLE `gift` (
    `id` varchar(50) NOT NULL,
    `name` varchar(50) NOT NULL,
    `price` decimal(10,2) NOT NULL DEFAULT '0.00',
    `icon_url` varchar(200) DEFAULT NULL,
    `is_vip` tinyint(1) DEFAULT '0',
    PRIMARY KEY (`id`)
);
-- 礼物记录表
CREATE TABLE `gift_record` (
    `id` bigint NOT NULL AUTO_INCREMENT,
    `user_id` bigint NOT NULL,
    `room_id` bigint NOT NULL,
    `gift_id` varchar(50) NOT NULL,
    `price` decimal(10,2) NOT NULL,
    `time` datetime DEFAULT NULL,
    PRIMARY KEY (`id`)
);
-- 聊天记录表
CREATE TABLE `chat_record` (
    `id` bigint NOT NULL AUTO_INCREMENT,
    `room_id` bigint NOT NULL,
    `user_id` bigint NOT NULL,
    `content` varchar(500) DEFAULT NULL,
    `time` datetime DEFAULT NULL,
    PRIMARY KEY (`id`)
);

使用说明

启动系统

# 安装依赖
mvn clean install
# 启动应用
java -jar live-broadcast.jar

配置环境

  • 修改 application.yml 中的数据库连接
  • 配置Redis连接地址
  • 设置FFmpeg路径

系统优化建议

  1. 高并发处理：使用消息队列（RabbitMQ/Kafka）处理聊天和礼物消息
  2. CDN加速：集成CDN分发视频流
  3. 缓存优化：使用Redis缓存热点数据
  4. 监控告警：集成Prometheus监控系统状态

这个案例涵盖了直播系统的主要功能模块,您可以根据实际需求进行扩展和优化。

上一篇Java社交系统案例

下一篇Java在线教育系统案例

相关文章

抱歉，评论功能暂时关闭!