以下代码实现了，按F1录屏+推送到kafka F2停止录屏+停止推送

import asyncio
import base64
import json
import threading
import time

import cv2
import numpy as np
from direct.actor.Actor import Actor
from direct.showbase.ShowBase import ShowBase
from fastapi import websockets
from kafka import KafkaProducer
from panda3d.core import CollisionNode, CollisionHandlerEvent, CollisionBox, CollisionSphere, CardMaker, \
    DirectionalLight, AmbientLight, WindowProperties


def add_lighting(self):
    """添加基础光照"""
    # 环境光
    alight = AmbientLight("ambient")
    alight.setColor((0.5, 0.5, 0.5, 1))
    alnp = self.render.attachNewNode(alight)
    self.render.setLight(alnp)

    # 方向光
    dlight = DirectionalLight("directional")
    dlight.setColor((0.8, 0.8, 0.8, 1))
    dlnp = self.render.attachNewNode(dlight)
    dlnp.setHpr(45, -45, 0)
    self.render.setLight(dlnp)

def create_cube(self):
    """创建简单正方体"""
    # 使用 CardMaker 创建立方体的各个面
    cm = CardMaker("cube_face")
    cm.setFrame(-1, 1, -1, 1)

    # 创建立方体节点
    self.cube = self.render.attachNewNode("cube")
    # 为正方体设置双面渲染
    self.cube.setTwoSided(True)
    # 创建6个面
    faces = []
    # 前面
    face = self.cube.attachNewNode(cm.generate())
    face.setPos(0, 1, 0)

    # 后面
    face = self.cube.attachNewNode(cm.generate())
    face.setPos(0, -1, 0)
    face.setHpr(180, 0, 0)

    # 左面
    face = self.cube.attachNewNode(cm.generate())
    face.setPos(-1, 0, 0)
    face.setHpr(90, 0, 0)

    # 右面
    face = self.cube.attachNewNode(cm.generate())
    face.setPos(1, 0, 0)
    face.setHpr(-90, 0, 0)

    # 上面
    face = self.cube.attachNewNode(cm.generate())
    face.setPos(0, 0, 1)
    face.setHpr(0, -90, 0)

    # 下面
    face = self.cube.attachNewNode(cm.generate())
    face.setPos(0, 0, -1)
    face.setHpr(0, 90, 0)

    # 设置颜色
    self.cube.setColor(0.2, 0.6, 1.0, 1.0)

# 加载模型时忽略动画数据
def load_glb_without_animation(self, file_path):
    try:
        # 尝试加载模型但不处理动画
        self.model = self.loader.loadModel(file_path)
        if self.model:
            # 移除骨骼相关节点
            self.remove_skeleton_nodes(self.model)
            self.model.reparentTo(self.render)
            return self.model
    except Exception as e:
        print(f"加载模型时出错: {e}")
        return None

def remove_skeleton_nodes(self, node_path):
    """移除骨骼相关节点"""
    # 查找并移除骨骼节点
    skeleton_nodes = node_path.findAllMatches("**/+SkeletonNode")
    for node in skeleton_nodes:
        node.removeNode()


class PandaRenderer(ShowBase):
    def __init__(self):
        ShowBase.__init__(self)
        # 或者使用更完整的方式
        from panda3d.core import WindowProperties
        wp = WindowProperties()
        wp.setTitle("实时推流")
        self.win.requestProperties(wp)


        # Kafka 配置
        self.kafka_producer = None
        self.kafka_topic = "panda3d_frames"
        self.setup_kafka()


        # 加载环境模型
        # self.environ = self.loader.loadModel("models/environment")
        # self.environ.reparentTo(self.render)
        # self.environ.setScale(0.25, 0.25, 0.25)
        # self.environ.setPos(-8, 42, 0)

        # 加载 GLB 模型
        # self.model = self.loader.loadModel("walking.glb")
        # if self.model:
        #     self.model.reparentTo(self.render)
        #     self.model.setPos(0, -15, 0)
        #     self.model.setScale(10.0)
        # else:
        #     print("模型加载失败")

        create_cube(self)
        add_lighting(self)


        self.camera.setPos(0, -15, 0)
        self.camera.lookAt(0, 0, 0)

        # 加载场景角色(一只熊猫)
        self.pandaActor = Actor(models="models/panda-model",
                                anims={"walk": "models/panda-walk4"})
        # 设置熊猫大小
        self.pandaActor.setScale(0.05, 0.05, 0.05)
        # 将熊猫加入渲染列表
        self.pandaActor.reparentTo(self.render)
        # 加入熊猫的循环动画
        self.pandaActor.loop("walk")










        #
        # # 创建一个碰撞检测节点，使用包围球
        # self.cNode = self.environ.attachNewNode(CollisionNode('player'))
        # # 创建包围球碰撞体（中心在原点，半径为1）
        # collision_sphere = CollisionSphere(0, 0, 0, 1)
        # self.cNode.node().addSolid(collision_sphere)
        #
        # # 创建碰撞检测处理器
        # self.cHandler = CollisionHandlerEvent()
        # self.cHandler.addInPattern('player-into-environment')
        #
        # # 配置碰撞节点
        # self.cNode.show()
        # self.cNode.node().setIntoCollideMask(1)
        # self.cNode.node().setFromCollideMask(128)
        # self.accept('player-into-environment', self.intoEnvironment)

        # 视频录制相关属性
        self.is_recording = False
        self.video_writer = None
        self.recorded_frames = []

        # 添加录制控制键
        self.accept('f1', self.start_recording)
        self.accept('f2', self.stop_recording)
        # 添加持续帧捕获任务
        self.frame_capture_task = None


        # 添加鼠标中键缩放功能
        self.accept('wheel_up', self.zoom_in)
        self.accept('wheel_down', self.zoom_out)

        # 初始化相机距离
        self.camera_distance = 8.0

        self.accept('arrow_left', self.rotateViewLeft)
        self.accept('arrow_right', self.rotateViewRight)
        self.accept("w", self.moveForward)


    def moveForward(self):
        print("Moving forward!")
    def intoEnvironment(self, entry):
        # 当玩家进入环境时执行的操作
        print("You have entered the environment!")
    def rotateViewLeft(self):
        self.camera.setHpr(self.camera.getHpr() + (0, -10, 0))

    def rotateViewRight(self):
        self.camera.setHpr(self.camera.getHpr() + (0, 10, 0))

    def zoom_in(self):
        """鼠标滚轮向上滚动时拉近相机"""
        self.camera_distance = max(5.0, self.camera_distance - 1.0)
        self.update_camera_position()

    def zoom_out(self):
        """鼠标滚轮向下滚动时推远相机"""
        self.camera_distance = min(50.0, self.camera_distance + 1.0)
        self.update_camera_position()

    def update_camera_position(self):
        """更新相机位置"""
        # 基于当前相机朝向和距离更新位置
        # 这里假设相机始终看向原点 (0, 0, 0)
        # from panda3d.core import Vec3
        # direction = self.camera.getPos() - Vec3(0, 0, 0)
        # direction.normalize()
        # new_pos = direction * self.camera_distance
        # self.camera.setPos(new_pos)
        if self.is_recording:
            frame = self.capture_frame()
            if frame is not None:
                self.recorded_frames.append(frame)

    def start_recording(self):
        """开始录制"""
        self.is_recording = True
        # 启动帧捕获任务
        self.frame_capture_task = self.taskMgr.add(self.capture_frame_task, "frame_capture")
        print("开始录制视频...")

    def capture_frame_task(self, task):
        """持续捕获帧的任务"""
        if self.is_recording:
            # 确保渲染完成后再捕获
            self.graphicsEngine.renderFrame()
            frame = self.capture_frame()
            if frame is not None:
                self.recorded_frames.append(frame)
                # 推送
                # 推送帧到 Kafka
                if self.kafka_producer:
                    self.broadcast_frame_kfk()

        return task.cont  # 继续执行任务

    def capture_frame(self):
        """捕获当前帧"""
        if self.win:
            # 获取屏幕截图

            screenshot = self.win.getScreenshot()
            if screenshot:
                # 转换为numpy数组
                img_data = screenshot.getRamImage().getData()
                img = np.frombuffer(img_data, dtype=np.uint8)
                img = img.reshape((screenshot.getYSize(), screenshot.getXSize(), 4))
                # 转换RGBA到BGR（OpenCV格式）
                # img = cv2.cvtColor(img, cv2.COLOR_RGBA2BGR)

                # 正确的颜色转换（从 RGBA 到 BGRA 再到 BGR）
                # 分离 alpha 通道
                b, g, r, a = cv2.split(img)
                # 重新组合为 BGR
                img = cv2.merge([b, g, r])


                # 垂直翻转图像
                img = cv2.flip(img, 0)
                return img
        return None
    def stop_recording(self):
        """停止录制并保存视频"""
        self.is_recording = False
        # 停止帧捕获任务
        if self.frame_capture_task:
            self.taskMgr.remove(self.frame_capture_task)
            self.frame_capture_task = None
        self.save_video()
        print("视频录制完成")

    def save_video(self, filename="recording.mp4"):
        """保存录制的视频"""
        if not self.recorded_frames:
            print("没有录制的帧")
            return

        # 获取第一帧的尺寸
        height, width = self.recorded_frames[0].shape[:2]

        # 创建视频写入器
        # fourcc = cv2.VideoWriter_fourcc(*'mp4v')
        fourcc = cv2.VideoWriter_fourcc(*'XVID')
        self.video_writer = cv2.VideoWriter(
            filename,
            fourcc, 60.0, (width, height),
            True
        )

        # 写入所有帧
        for frame in self.recorded_frames:
            self.video_writer.write(frame)

        # 释放资源
        self.video_writer.release()
        self.recorded_frames = []
        print(f"视频已保存为 {filename}")



    def setup_kafka(self):
        """初始化 Kafka 生产者"""
        try:
            self.kafka_producer = KafkaProducer(
                bootstrap_servers=['127.0.0.1:9092'],
                value_serializer=lambda x: json.dumps(x).encode('utf-8')
            )
            print("Kafka 生产者初始化成功")
        except Exception as e:
            print(f"Kafka 初始化失败: {e}")

    def broadcast_frame_kfk(self):
        """通过 Kafka 广播当前帧"""
        if not self.kafka_producer:
            return

        try:
            # 捕获当前帧
            frame = self.capture_frame()

            # 编码为 JPEG
            _, buffer = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, 80])

            # 转换为 base64
            jpg_as_text = base64.b64encode(buffer).decode('utf-8')

            # 创建消息
            message = {
                "type": "frame",
                "data": jpg_as_text,
                "timestamp": time.time()
            }

            # 发送到 Kafka
            self.kafka_producer.send(self.kafka_topic, value=message)
            self.kafka_producer.flush()

        except Exception as e:
            print(f"Kafka 推送错误: {e}")

if __name__ == "__main__":

    renderer = PandaRenderer()

    renderer.run()

开始用python写websocket一直有问题，所有用java读取kafka 推给前端
前端代码

<!DOCTYPE html>
<html>
<head>
    <title>Kafka WebSocket 视频流客户端</title>
    <style>
        body {
            margin: 0;
            padding: 20px;
            font-family: Arial, sans-serif;
            background-color: #f0f0f0;
        }
        .container {
            max-width: 1280px;
            margin: 0 auto;
            text-align: center;
        }
        #videoCanvas {
            border: 2px solid #333;
            background-color: #000;
            box-shadow: 0 4px 8px rgba(0,0,0,0.1);
        }
        .controls {
            margin: 20px 0;
        }
        button {
            padding: 10px 20px;
            font-size: 16px;
            background-color: #4CAF50;
            color: white;
            border: none;
            border-radius: 4px;
            cursor: pointer;
            margin: 0 10px;
        }
        button:hover {
            background-color: #45a049;
        }
        button:disabled {
            background-color: #cccccc;
            cursor: not-allowed;
        }
        .status {
            margin: 10px 0;
            padding: 10px;
            border-radius: 4px;
            font-weight: bold;
        }
        .connected {
            background-color: #d4edda;
            color: #155724;
            border: 1px solid #c3e6cb;
        }
        .disconnected {
            background-color: #f8d7da;
            color: #721c24;
            border: 1px solid #f5c6cb;
        }
    </style>
</head>
<body>
    <div class="container">
        <h1>Kafka WebSocket 实时视频流</h1>
        
        <div class="controls">
            <button id="connectBtn">连接视频流</button>
            <button id="disconnectBtn" disabled>断开连接</button>
        </div>
        
        <div id="status" class="status disconnected">
            未连接 - 点击"连接视频流"开始接收视频
        </div>
        
        <div>
            <canvas id="videoCanvas" width="800" height="630"></canvas>
        </div>
        
        <div style="margin-top: 20px; color: #666;">
            <p>当前状态: <span id="connectionStatus">未连接</span></p>
            <p>接收帧数: <span id="frameCount">0</span></p>
        </div>
    </div>

    <script>
        class VideoStreamClient {
            constructor() {
                this.ws = null;
                this.canvas = document.getElementById('videoCanvas');
                this.ctx = this.canvas.getContext('2d');
                this.frameCount = 0;
                
                this.connectBtn = document.getElementById('connectBtn');
                this.disconnectBtn = document.getElementById('disconnectBtn');
                this.statusDiv = document.getElementById('status');
                this.connectionStatus = document.getElementById('connectionStatus');
                this.frameCountSpan = document.getElementById('frameCount');
                
                this.setupEventListeners();
            }
            
            setupEventListeners() {
                this.connectBtn.addEventListener('click', () => this.connect());
                this.disconnectBtn.addEventListener('click', () => this.disconnect());
            }
            
            connect() {
                try {
                 
                    // 连接到 WebSocket 服务
                    this.ws = new WebSocket('ws://192.168.31.190:8080/ws');
                // 在前端代码中添加更多调试信息
                    this.ws.onopen = () => {
                        console.log('WebSocket 连接已建立');
                        this.updateStatus('已连接', true);
                        this.connectBtn.disabled = true;
                        this.disconnectBtn.disabled = false;
                        this.connectionStatus.textContent = '已连接';
                        
                        // 发送测试消息
                        this.ws.send(JSON.stringify({
                            type: "ping",
                            message: "hello server"
                        }));
                    };

                    this.ws.onmessage = (event) => {
                        console.log('收到服务器消息:', event.data);
                        const data = JSON.parse(event.data);
                        if (data.type === 'frame') {
                            this.displayFrame(data.data);
                            this.frameCount++;
                            this.frameCountSpan.textContent = this.frameCount;
                        } else if (data.type === 'status') {
                            console.log('服务器状态:', data.message);
                        }
                    };
                    
                    this.ws.onerror = (error) => {
                        console.error('WebSocket 错误:', error);
                        this.updateStatus('连接错误', false);
                    };
                    
                } catch (error) {
                    console.error('连接失败:', error);
                    this.updateStatus('连接失败: ' + error.message, false);
                }
            }
            
            disconnect() {
                if (this.ws) {
                    this.ws.close();
                    this.ws = null;
                }
            }
            
            displayFrame(base64Data) {
                const img = new Image();
                img.onload = () => {
                    // 在 canvas 上绘制图像
                    this.ctx.drawImage(img, 0, 0, this.canvas.width, this.canvas.height);
                };
                img.src = 'data:image/jpeg;base64,' + base64Data;
            }
            
            updateStatus(message, isConnected) {
                this.statusDiv.textContent = message;
                this.statusDiv.className = 'status ' + (isConnected ? 'connected' : 'disconnected');
            }
        }
        
        // 页面加载完成后初始化客户端
        document.addEventListener('DOMContentLoaded', () => {
            const client = new VideoStreamClient();
            window.videoClient = client; // 便于调试
        });
    </script>
</body>
</html>