以下代碼實現了，按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>