laravel框架引用kafka

在 Laravel 中操作 Kafka，可以使用 php-rdkafka 扩展或 confluent-kafka-php 扩展。

以下展示如何使用 confluent-kafka-php 扩展来在 Laravel 中使用 Kafka。

操作步骤说明：

1、安装 confluent-kafka-php 扩展。您可以使用 Composer 进行安装：

composer require edenhill/php-rdkafka

2、需要在 Laravel 配置文件中配置 Kafka 连接信息。打开 config/database.php 文件，在 connections 数组中添加以下配置：

'kafka' => ['driver' => 'kafka','brokers' => env('KAFKA_BROKERS', 'localhost:9092'), // Kafka broker(s) information'group_id' => env('KAFKA_GROUP_ID', 'my-group'), // Consumer group ID],

3、.env 文件中设置 Kafka 的连接信息：

KAFKA_BROKERS=localhost:9092
KAFKA_GROUP_ID=my-group

4、创建一个 Kafka 服务提供者，以便将 Kafka 服务添加到 Laravel 的容器中。运行以下命令来生成服务提供者：

php artisan make:provider KafkaServiceProvider

5、在 KafkaServiceProvider.php 文件中注册 Kafka 服务：

use Illuminate\Support\ServiceProvider;
use Confluent\Kafka\Producer;
use Confluent\Kafka\Consumer;
use Confluent\Kafka\ConsumerTopic;class KafkaServiceProvider extends ServiceProvider
{public function register(){$this->app->singleton('kafka-producer', function ($app) {$config = new \RdKafka\Producer\Conf();$config->set('metadata.broker.list', config('database.connections.kafka.brokers'));return new Producer($config);});$this->app->singleton('kafka-consumer', function ($app) {$config = new \RdKafka\Conf();$config->set('group.id', config('database.connections.kafka.group_id'));$consumer = new Consumer($config);$consumer->subscribe([config('database.connections.kafka.topic')]);return $consumer;});}
}

6、使用 Laravel 的依赖注入来访问 Kafka 生产者和消费者。例如，在控制器中：

use Illuminate\Http\Request;
use Confluent\Kafka\Producer;class KafkaController extends Controller
{protected $producer;public function __construct(Producer $producer){$this->producer = $producer;}public function produceMessage(Request $request){// 生产消息到 Kafka$message = $request->input('message');$this->producer->produce('kafka-topic', 0, $message);return response()->json(['message' => 'Message sent to Kafka']);}
}

7、可以创建一个消费者定时任务服务来处理 Kafka 消息。创建一个消费者命令：

php artisan make:command KafkaConsumer

在 KafkaConsumer.php 文件中编写消费者逻辑：

use Illuminate\Console\Command;
use Confluent\Kafka\Consumer;class KafkaConsumer extends Command
{protected $signature = 'kafka:consume';protected $description = 'Consume messages from Kafka topic';public function handle(){$consumer = app('kafka-consumer');while (true) {$message = $consumer->consume(120 * 1000); // 2 minutes timeoutif ($message->err) {$this->error('Error consuming message: ' . $message->errstr());} else {$this->info('Received message: ' . $message->payload);// 处理消息的逻辑}}}
}

在 kernel.php 文件中添加计划任务以运行 Kafka 消费者：

protected $commands = [// ...\App\Console\Commands\KafkaConsumer::class,
];protected function schedule(Schedule $schedule)
{$schedule->command('kafka:consume')->everyMinute(); // 每分钟运行一次
}

最后，使用以下命令运行 Kafka 消费者：