new_sdk_wu/www/new_sdk/application/command/KafkaConsumer.php

<?php
namespace app\command;

use think\console\Command;
use think\console\Input;
use think\console\Output;
use app\common\library\Kafka;
use think\Log;

class KafkaConsumer extends Command
{
    protected function configure()
    {
        $this->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>Error: ' . $e->getMessage() . '</error>');
            Log::error('Kafka consumer error: ' . $e->getMessage());
        }
    }
}