setName('kafka:consume') ->setDescription('Kafka consumer command'); } protected function execute(Input $input, Output $output) { $output->writeln('Starting Kafka consumer...'); try { $kafka = Kafka::getInstance(); $topics = ['new_sdk_topic']; $kafka->consume($topics, function($message) use ($output) { $data = json_decode($message->payload, true); // 记录消息到日志 Log::info('Received Kafka message: ' . $message->payload); // 输出消息到控制台 $output->writeln(sprintf( "Received message:\nTopic: %s\nPartition: %s\nOffset: %s\nKey: %s\nPayload: %s\n", $message->topic_name, $message->partition, $message->offset, $message->key, $message->payload )); // 这里可以添加具体的业务处理逻辑 switch ($data['event']) { case 'test_event': $output->writeln('Processing test event...'); break; case 'custom_event': $output->writeln('Processing custom event...'); break; default: $output->writeln('Processing unknown event...'); } }); } catch (\Exception $e) { $output->writeln('Error: ' . $e->getMessage() . ''); Log::error('Kafka consumer error: ' . $e->getMessage()); } } }