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 消息队列的集成。你可以根据需要扩展这些服务，以处理更复杂的业务逻辑。