/**

     * 生产单个消息

     * @param string $topic

     * @param null $post

     */

    function kafka_produce($key=null,$post=null)

    {

        $rk = new \RdKafka\producer();

        $rk->setLogLevel(LOG_DEBUG);

        $rk->addBrokers("localhost:9092,localhost:9093,localhost:9094,localhost:9095");

        $topics = $rk->newTopic('engine.com');

        $topics->produce(1, 0,$post,$key);

        echo 'kafka_produce success!!!';

    }

/**

     * 高级消费模式

     * @param $topic

     * @return int

     * @throws Exception

     */

    function kafka_high_consume($topic='engine.com'){

        $conf = new \RdKafka\Conf();

        $conf->setRebalanceCb(function (RdKafka\KafkaConsumer $kafka, $err, array $partitions = null) {

            switch ($err) {

                case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:

                    echo "Assign: ";

                    var_dump($partitions);

                    $kafka->assign($partitions);

                    break;

                case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:

                    echo "Revoke: ";

                    var_dump($partitions);

                    $kafka->assign(NULL);

                    break;

                default:

                    throw new \Exception($err);

            }

        });

        $conf->set('group.id', '0');

        $conf->set('metadata.broker.list', 'localhost:9092,localhost:9093,localhost:9094,localhost:9095');

//        针对低延迟进行了优化的配置。这允许PHP进程/请求尽快发送消息并快速终止

        $conf->set('socket.timeout.ms', 50);

        if (function_exists('pcntl_sigprocmask')) {

            pcntl_sigprocmask(SIG_BLOCK, array(SIGIO));

            $conf->set('internal.termination.signal', SIGIO);

        } else {

            $conf->set('queue.buffering.max.ms', 1);

        }

        $topicConf = new \RdKafka\TopicConf();

        $topicConf->set('auto.commit.interval.ms', 100);

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

        $topicConf->set('offset.store.path', 'kafka_offset.log');

        $conf->setDefaultTopicConf($topicConf);

        $consumer = new \RdKafka\KafkaConsumer($conf);

//        $topics->consumeStart(1, RD_KAFKA_OFFSET_STORED);

//        更新订阅集（自动分配partitions ）

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

//        指定topic分配partitions使用那个分区

//        $consumer->assign([

//            new \RdKafka\TopicPartition("zzy8", 0),

//            new \RdKafka\TopicPartition("zzy8", 1),

//            ]);

        while (true) {

//            设置120s为超时

            $message = $consumer->consume(120 * 1000);

            if (!empty($message)) {

                switch ($message->err) {

                    case RD_KAFKA_RESP_ERR_NO_ERROR:

                        info('New message received ：', $message);

//                        拆解对象为数组

                        $payload = json_decode($message->payload,true);

                        $Orders = new OrdersController();

                        $key = $message->key;

//                        根据kafka中不同key，调用对应方法传递处理数据

                        ...
　　　　　　　　　　　　　　 ...
　　　　　　　　　　　　　　　　...
break;

                    case RD_KAFKA_RESP_ERR__PARTITION_EOF:

                        echo "No more messages; will wait for more\n";

                        break;

                    case RD_KAFKA_RESP_ERR__TIMED_OUT:

                        echo "Timed out\n";

                        var_dump("##################");

                        break;

                    default:

                        var_dump("nothing");

                        throw new \Exception($message->errstr(), $message->err);

                        break;

                }

            } else {

                var_dump('this is empty obj!!!');

            }

        }

        return 0;

    }