开发环境使用的是laradock ,安装kafka及拓展可以看历史文章
配置好后,laravel使用kafka代码
在 Laravel 项目中的 .env 文件中添加以下行来配置 Kafka:
KAFKA_BROKERS=kafka:9092
KAFKA_TOPIC=my-topic
生产者
use RdKafka\Producer;
use RdKafka\ProducerTopic;
use RdKafka\Conf;
$config = new Conf();
$config->set('metadata.broker.list', env('KAFKA_BROKERS'));
$producer = new Producer($config);
$topic = $producer->newTopic(env('KAFKA_TOPIC'));
$message = "Hello, Kafka!";
$topic->produce(RD_KAFKA_PARTITION_UA, 0, $message);
$producer->poll(0);
在这个例子中,我们使用了 Laravel 的 .env
文件中配置的 Kafka 代理和主题。由于我们设置了RD_KAFKA_PARTITION_UA,消息将被广播到所有 Kafka 分区。
消费者
use RdKafka\Conf;
use RdKafka\KafkaConsumer;
use RdKafka\TopicPartition;
$conf = new Conf();
$conf->set('group.id', 'my-group');
$conf->set('metadata.broker.list', env('KAFKA_BROKERS'));
$consumer = new KafkaConsumer($conf);
$consumer->subscribe([env('KAFKA_TOPIC')]);
while (true) {
$message = $consumer->consume(120 * 1000);
if ($message) {
switch ($message->err) {
case RD_KAFKA_RESP_ERR_NO_ERROR:
// 处理消息
echo sprintf(
"Message payload: %s\n",
$message->payload
);
break;
case RD_KAFKA_RESP_ERR__TIMED_OUT:
break;
default:
throw new \Exception($message->errstr(), $message->err);
}
}
}
在这个例子中,我们首先创建了一个 Kafka 消费者并订阅了我们要消费的主题。然后,我们进入一个无限循环,在循环中使用 consume
方法从 Kafka 中读取消息。如果没有消息,我们等待超时(我们设置的是 120 秒)。如果有新消息到达,我们首先检查其错误码以确定它是否有效,然后在必要时处理消息。
核心就是 配置项 kafka:9092
使用web接口fpm消费时 会提示 code -185, local time out
网络问题,使用cli模式消费解决问题