「Kafka应用」PHP实现生产者与消费者

前言#

PHP 中使用 Kafka 需要 RdKafka 扩展,而 RdKafka 依赖于 librdkafka,所以这两个我们都需要安装,具体安装方法自行百度,本篇不做说明了。

生产者(测试)#

创建消费者需要步骤:

  • 生产者配置参数
  • 创建生产者实例
  • 创建主题实例 (依赖生产者)
  • 生产主题消息
  • 推送消息

具体代码如下:

        $conf = new \RdKafka\Conf();
        // 绑定服务节点
        $conf->set('metadata.broker.list', '127.0.0.1:32772');

        // 创建生产者
        $kafka = new \RdKafka\Producer($conf);

        // 创建主题实例
        $topic = $kafka->newTopic('p1r1');
        // 生产主题数据,此时消息在缓冲区中,并没有真正被推送
        $topic->produce(RD_KAFKA_PARTITION_UA, 0, 'Message');
        // 阻塞时间(毫秒), 0为非阻塞
        $kafka->poll(0); 

        // 推送消息,如果不调用此函数,消息不会被发送且会丢失
        $result = $kafka->flush(5000);

        if (RD_KAFKA_RESP_ERR_NO_ERROR !== $result) {
            throw new \RuntimeException('Was unable to flush, messages might be lost!');
        }

消费者#

创建一个消费者需要几个步骤:

  • 消费者配置参数
  • 应用配置参数创建消费者实例
  • 订阅对应主题
  • 拉取数据
  • 提交位移

具体代码如下:

        $conf = new \RdKafka\Conf();
        // 绑定消费者组
        $conf->set('group.id', 'ceshi');
        // 绑定服务节点,多个用,分隔
        $conf->set('metadata.broker.list', '127.0.0.1:32787');
        // 设置自动提交为false
        $conf->set('enable.auto.commit', 'false');
        // 设置当前消费者拉取数据时的偏移量, 可选参数:
        // earliest: 如果消费者组是新创建的,从头开始消费,否则从消费者组当前消费位移开始。
        // latest:如果消费者组是新创建的,从最新偏移量开始,否则从消费者组当前消费位移开始。
        $conf->set('auto.offset.reset', 'earliest');

        // 创建消费者实例
        $consumer = new \RdKafka\KafkaConsumer($conf);
        // 消费者订阅主题,数组形式
        $consumer->subscribe(['topic1','topic2']);
        while (true) {
            // 消费数据,阻塞5秒(5秒内有数据就消费,没有数据等待5秒进入下一轮循环)
            $message = $consumer->consume(5000);
            switch ($message->err) {
                case RD_KAFKA_RESP_ERR_NO_ERROR:
                    // 业务逻辑
                    var_dump($message);

                    // 提交位移
                    $consumer->commit($message);
                    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";
                    break;
                default:
                    throw new \Exception($message->errstr(), $message->err);
                    break;
            }
        }
        // 关闭消费者(一般用在脚本中,不需要关闭)
        $conumser->close();

只消费指定分区中的数据:

    // 对消费者指定分区,注意此方式不能与subscribe一同使用
    $consumer->assign([
        new RdKafka\TopicPartition("topic", 0),
        new RdKafka\TopicPartition("topic", 1),
    ]);
本作品采用《CC 协议》,转载必须注明作者和本文链接
本帖由系统于 4年前 自动加精
《L04 微信小程序从零到发布》
从小程序个人账户申请开始,带你一步步进行开发一个微信小程序,直到提交微信控制台上线发布。
《L02 从零构建论坛系统》
以构建论坛项目 LaraBBS 为线索,展开对 Laravel 框架的全面学习。应用程序架构思路贴近 Laravel 框架的设计哲学。
讨论数量: 8

大佬,现在碰到一个问题,在 broker 都 down 掉的时候就会一直阻塞,这个有办法解决么?把所有..ms 相关的配置都设置了,还是会阻塞住。。。。。

4年前 评论
ligkwww (楼主) 4年前

你好,如果生产者 broker list 中存在某个 broker 服务不可用了,如何实现故障转移?

3年前 评论
ligkwww (楼主) 3年前

大佬,有 kafka 的调用疑问,有群后者私聊吗

3年前 评论
ligkwww (楼主) 3年前

offset reset (at offset 4) to END: fetch failed due to requested offset not available on the broker: Broker: Offset out of range 大佬,订阅报这个错误,什么问题

2年前 评论

请问 如果脚本挂了 重启 是会从头消费吗?或者 offset 哪里传递 谢谢

2年前 评论