README

安装

php composer require glsv/yii2kafka:dev-master

支持的 php-clients

• longyan/phpkafka
• weiboad/kafka-php

步骤顺序

1. 安装 glsv/yii2kafka
2. 选择并安装一些 php-client (longyan/phpkafka 或 weiboad/kafka-php)
3. 配置扩展和客户端

配置 glsv/yii2kafka

components = [
    'class' => \yii2Kafka\YiiKafkaComponent::class,
    'brokers' => ['kafka:9092'],
    'defaultLogPath' => '@runtime/logs/kafka.log',
    'debugMode' => true,
    'adapter => [
       // configure adapter for php-client... 
    ]
]

集成 php-client longyan/phpkafka

php composer require longlang/phpkafka

https://github.com/longyan/phpkafka

配置 yii2

components = [
    'class' => \yii2Kafka\YiiKafkaComponent::class,
    'brokers' => ['kafka:9092']
     /**
     * https://github.com/longyan/phpkafka
     */
    'adapter' => [
        'class' => \yii2Kafka\adapters\longlang\LonglangAdapter::class,
        'params' => [
            ////////
            // Use the params for the client at the links:
            // producer: https://github.com/longyan/phpkafka/blob/master/doc/producer.en.md
            // consumer: https://github.com/longyan/phpkafka/blob/master/doc/consumer.en.md
            ////////
            'producer' => [
                'autoCreateTopic' => false,
                'acks' => 1,
            ],
            'consumer' => [
                'topic' => ['test-topic'],
                'groupId' => 'consumer-group',
                'clientId' => 'client-consumer',
            ]
        ],
    ],
]

集成 php-client nmred/kafka-php

php composer require nmred/kafka-php

https://github.com/weiboad/kafka-php

使用 nmred/kafka-php 配置 yii2

components = [
    'class' => \yii2Kafka\YiiKafkaComponent::class,
    'brokers' => ['kafka:9092']
    /**
     * https://github.com/weiboad/kafka-php
     */
    'adapter' => [
        'class' => \yii2Kafka\adapters\nmred\NmredAdapter::class,
        'params' => [
            ////////
            // Use the params for the client at the link:
            // https://github.com/weiboad/kafka-php/blob/master/docs/Configure.md
            ////////       
            'global' => [],
            'producer' => [
                'isAsyn' => false,
                'requiredAck' => 1,
            ],
            'consumer' => [
                'topics' => ['test-topic'],
                'groupId' => 'consumer-group'
            ]
        ]
    ],
]

生产示例

生产单条消息

/**
 * @var YiiKafkaComponent $kafka
 */
$kafka = \Yii::$app->kafka;

$producer = $kafka->getProducer();
$topicName = "test-topic"
$producer->send($topicName, 'Message: ' . date('d.m H:i:s'), '');

批量生产消息

/**
 * @var YiiKafkaComponent $kafka
 */
$kafka = \Yii::$app->kafka;

$producer = $kafka->getProducer();
$topicName = "test-topic"
$messages = [
    new Message($topicName, 'Message 1: ' . date('d.m H:i:s')),
    new Message($topicName, 'Message 2: ' . date('d.m H:i:s')),
];

$producer->sendBatch($messages);

消费示例

从设置中消费特定主题的消息

/**
 * @var YiiKafkaComponent $kafka
 */
$kafka = \Yii::$app->kafka;

$consumer = $kafka->getConsumer();
$consumer->consume(function ($message) {
    var_dump($message);
});

从设置中消费特定主题的消息

/**
 * @var YiiKafkaComponent $kafka
 */
$kafka = \Yii::$app->kafka;
$topicName = 'another-topic';

$config = new ConsumerConfig($topicName);
$config->setClientId('client_2');
$config->setGroupIp('group_2');

$consumer = $kafka->getAdapterClient()->createConsumer($config);
$consumer->consume(function ($message) {
    var_dump($message);
});