Laravel使用Kafka实践教程（从安装到使用0-1）

本文介绍: 在 Laravel 中使用 Kafka 消息队列，根据你的操作系统和 PHP 版本，安装 rdkafka 扩展。当消息被推送到 Kafka 队列后，Superviso进程管理消息队列监听作业将自动从队列中获取并执行

首先Linux服务器需要安装好Kafka,这里不再具体细说，可以看我前面的文章：Linux搭建Kafka详细一步一步指南(linux启动kafka脚本)-CSDN博客

composer require enqueue/rdkafka:0.10.18

注意：版本号一定要选对！！！至于版本号该怎么选择，这里也不再细说，可以看我前面的文章：Laravel 使用rdkafka_laravel详细教程（实操避坑）_laravel kafka-CSDN博客

根据你的操作系统和 PHP 版本，安装 rdkafka 扩展。你可以参考 rdkafka 扩展的官方文档来完成安装

可参考文章里面【windows】PHP扩展安装教程，Linux也差不多，下载对应linux的就可以：Laravel 使用rdkafka_laravel详细教程（实操避坑）_laravel kafka-CSDN博客

打开 config/queue.php 文件，找到 connections 数组，并添加以下配置：

'connections' => [
     //kafka消息队列
     'rdkafka' => [
            'driver' => 'rdkafka',
            'brokers' => env('KAFKA_BROKERS', 'localhost'),
            'topic' => env('KAFKA_TOPIC', 'default_topic'),
            'group_id' => env('KAFKA_GROUP_ID', 'default_group'),
            'security_protocol' => 'plaintext', // 可选配置，根据实际情况设置，安全协议，用于与 Kafka 代理进行通信的协议，plaintext，表示明文通信，可以根据实际情况设置其他支持的安全协议，例如 ssl 或 sasl_ssl
            'sasl_mechanisms' => null, // 可选配置，根据实际情况设置
            'sasl_username' => null, // 可选配置，根据实际情况设置
            'sasl_password' => null, // 可选配置，根据实际情况设置
            'commit_async' => true, // 可选配置，根据实际情况设置
        ],
],

<?php

namespace AppJobs;

use EnqueueRdKafkaRdKafkaContext;
use IlluminateContractsQueueShouldQueue;
use IlluminateFoundationBusDispatchable;
use IlluminateQueueInteractsWithQueue;
use IlluminateQueueSerializesModels;
use IlluminateSupportFacadesLog;

class KafkaMessage implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, SerializesModels;

    protected $message;

    /**
     * Create a new job instance.
     *
     * @param  string  $message
     * @return void
     */
    public function __construct(string $message)
    {
        $this->message = $message;
    }

    /**
     * Execute the job.
     *
     * @return void
     */
    public function handle()
    {
        Log::info('Received Kafka message: ' . $this->message);
    }
}

以下是一个示例，演示如何将消息推送到 Kafka 队列：

<?php

namespace AppHttpControllers;

use AppJobsProcessKafkaMessage;
use IlluminateHttpRequest;

class KafkaController extends Controller
{
    public function sendMessage(Request $request)
    {
        $message = $request->input('message');

        KafkaMessage::dispatch($message)
            ->onQueue('kafka')
            ->delay(now()->addSeconds(5));

        return response()->json(['message' => 'Message sent to Kafka']);
    }
}

在这个示例中，将接收到的消息作为参数传递给 KafkaMessage 作业，并将作业推送到名为 “kafka” 的队列上，并设置了延迟时间为 5 秒。

在 Laravel 中，你可以使用 queue:work Artisan 命令来监听 Kafka 消息队列并处理消息。以下是监听 Kafka 消息队列的步骤：

php artisan queue:work kafka --queue=kafka

显示所有内容

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。

一、Linux安装Kafka

二、Laravel安装Kafka

三、安装 rdkafka 扩展

四、配置 Laravel Enqueue

五、创建Job

六、将消息推送到 Kafka 队列

七、监听队列消费消息

发表回复取消回复