Symfony與Kafka消息隊列集成

將 Symfony 與 Kafka 消息隊列集成可以幫助你在 Symfony 應用中處理異步任務和事件驅動的架構。以下是一個基本的步驟指南，幫助你完成這個集成。

1. 安裝依賴

首先，你需要在你的 Symfony 項目中安裝 Kafka 客戶端庫。你可以使用 Composer 來安裝 php-kafka 庫。

composer require nmred/kafka-php

2. 配置 Kafka

在你的 Symfony 項目中創建一個新的配置文件，例如 config/packages/kafka.yaml，并添加 Kafka 的配置信息。

kafka:
    bootstrap_servers: 'localhost:9092'
    default_topic: 'symfony_events'

3. 創建 Kafka 生產者和消費者

生產者

創建一個新的服務來處理 Kafka 消息的生產。

php bin/console make:service KafkaProducer

在 src/Service/KafkaProducer.php 中添加以下代碼：

<?php

namespace App\Service;

use PhpKafka\ProducerConfig;
use PhpKafka\Producer;

class KafkaProducer
{
    private $producer;

    public function __construct(array $config)
    {
        $config = ProducerConfig::getInstance();
        $config->setMetadataRefreshIntervalMs(10000);
        $config->setIsAsyn(false);
        $config->setProduceInterval(500);
        $config->setIsAsync(false);
        $config->setBrokerVersion('1.0.0');
        $config->setRequiredAck(1);
        $config->setIsIdempotent(false);

        $this->producer = new Producer($config);
    }

    public function sendMessage($topic, $message)
    {
        $this->producer->send([
            [
                'topic' => $topic,
                'value' => $message,
                'key' => '',
            ],
        ]);
    }
}

消費者

創建一個新的服務來處理 Kafka 消息的消費。

php bin/console make:service KafkaConsumer

在 src/Service/KafkaConsumer.php 中添加以下代碼：

<?php

namespace App\Service;

use PhpKafka\ConsumerConfig;
use PhpKafka\KafkaConsumer;

class KafkaConsumer
{
    private $consumer;

    public function __construct(array $config)
    {
        $config = ConsumerConfig::getInstance();
        $config->setMetadataRefreshIntervalMs(10000);
        $config->setBrokerVersion('1.0.0');
        $config->setGroupId('symfony_group');
        $config->setBrokerVersion('1.0.0');
        $config->setTopics(['symfony_events']);

        $this->consumer = new KafkaConsumer($config);
    }

    public function startConsuming()
    {
        $this->consumer->start(function ($topic, $part, $message) {
            // 處理消息
            echo "Received message: " . $message['message']['value'] . "\n";
        });
    }
}

4. 注冊服務

在你的 services.yaml 文件中注冊這兩個服務。

services:
    App\Service\KafkaProducer:
        arguments:
            $config: '%kafka.bootstrap_servers%'

    App\Service\KafkaConsumer:
        arguments:
            $config: '%kafka.bootstrap_servers%'

5. 使用 Kafka 生產者和消費者

在你的 Symfony 應用中使用這些服務來發送和接收消息。

發送消息

在你的控制器或服務中，使用 Kafka 生產者發送消息。

<?php

namespace App\Controller;

use App\Service\KafkaProducer;
use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
use Symfony\Component\HttpFoundation\Response;

class KafkaController extends AbstractController
{
    private $kafkaProducer;

    public function __construct(KafkaProducer $kafkaProducer)
    {
        $this->kafkaProducer = $kafkaProducer;
    }

    public function sendMessageAction()
    {
        $message = 'Hello, Kafka!';
        $this->kafkaProducer->sendMessage('symfony_events', $message);

        return new Response('Message sent');
    }
}

接收消息

在你的消費者服務中啟動消費循環。

<?php

namespace App\Service;

use PhpKafka\ConsumerConfig;
use PhpKafka\KafkaConsumer;

class KafkaConsumer
{
    private $consumer;

    public function __construct(array $config)
    {
        $config = ConsumerConfig::getInstance();
        $config->setMetadataRefreshIntervalMs(10000);
        $config->setBrokerVersion('1.0.0');
        $config->setGroupId('symfony_group');
        $config->setBrokerVersion('1.0.0');
        $config->setTopics(['symfony_events']);

        $this->consumer = new KafkaConsumer($config);
    }

    public function startConsuming()
    {
        $this->consumer->start(function ($topic, $part, $message) {
            // 處理消息
            echo "Received message: " . $message['message']['value'] . "\n";
        });
    }
}

6. 運行消費者

在你的命令行中運行消費者服務。

php bin/console kafka:consume

這樣，你就完成了 Symfony 與 Kafka 消息隊列的集成。你可以根據需要擴展這些服務，以處理更復雜的業務邏輯。