php kafka使用

<?php
/**
 * Kafka.php.
 * User: lvfk
 * Date: 2018/2/7 0007
 * Time: 11:04
 * Desc: Kafka服务
 */

namespace appmodels;


use yiiaseInvalidConfigException;

class Kafka
{
    public $broker_list = '10.168.1.99:9092';//配置kafka，可以用逗号隔开多个kafka
    public $topic = 'topic';
    public $partition = 0;

    protected $producer = null;
    protected $consumer = null;

    public function __construct()
    {
        if (empty($this->broker_list)) {
            throw new InvalidConfigException("broker not config");
        }
        $rk = new RdKafkaProducer();
        if (empty($rk)) {
            throw new InvalidConfigException("producer error");
        }
        $rk->setLogLevel(LOG_DEBUG);
        if (!$rk->addBrokers($this->broker_list)) {
            throw new InvalidConfigException("producer error");
        }
        $this->producer = $rk;
    }

    /**
     * 生产者
     * @param array $messages
     * @return mixed
     */
    public function send($messages = [])
    {
        $topic = $this->producer->newTopic($this->topic);
        return $topic->produce(RD_KAFKA_PARTITION_UA, $this->partition, json_encode($messages));
    }

    /**
     * 消费者
     */
    public function consumer($object, $callback){
        $conf = new RdKafkaConf();
        $conf->set('group.id', 0);
        $conf->set('metadata.broker.list', $this->broker_list);

        $topicConf = new RdKafkaTopicConf();
        $topicConf->set('auto.offset.reset', 'smallest');

        $conf->setDefaultTopicConf($topicConf);

        $consumer = new RdKafkaKafkaConsumer($conf);

        $consumer->subscribe([$this->topic]);

        echo "waiting for messages.....
";
        while(true) {
            $message = $consumer->consume(120*1000);
            switch ($message->err) {
                case RD_KAFKA_RESP_ERR_NO_ERROR:
                    echo "message payload....";
                    $object->$callback($message->payload);
                    break;
            }
            sleep(1);
        }
    }
}

//配置kafka生产者
'asyncLog' => [
    'class' => '\app\models\Kafka',
    'broker_list' => '10.168.1.99:9092',
    'topic' => 'asynclog'
]

<?php
/**
 * @link http://www.yiiframework.com/
 * @copyright Copyright (c) 2008 Yii Software LLC
 * @license http://www.yiiframework.com/license/
 */

namespace appcommands;

use yiiconsoleController;

/**
 * This command echoes the first argument that you have entered.
 *
 * This command is provided as an example for you to learn how to create console commands.
 *
 * @author Qiang Xue <qiang.xue@gmail.com>
 * @since 2.0
 */
class KafkaController extends Controller
{
    /**
     * This command echoes what you have entered as the message.
     * @param string $message the message to be echoed.
     */
    public function actionConsume()
    {
        Yii::$app->asyncLog->consumer($this, 'callback');

    }

    public function callback($message)
    {
        Yii::info($message, 'testkafka');
        Yii::$app->log->setflushInterval(1);
    }

}