PHP Kafka client is used in PHP-FPM and Swoole. PHP Kafka client supports 50 APIs, which might be one that supports the most message types ever.

Swoole Project

Last update: Dec 31, 2022

Related tags

Overview

longlang/phpkafka

Introduction

English | 简体中文

PHP Kafka client is used in PHP-FPM and Swoole.

The communication protocol is based on the JSON file in Java. PHP Kafka client supports 50 APIs, which might be one that supports the most message types ever.

The produce and the consume of messages are implemented. The component is in developing and testing.

Features

Environment

PHP >= 7.1
Kafka >= 1.0.0
Swoole >= 4.5 (optional)

Installation

composer require longlang/phpkafka

Documentation and Examples

Refer to examples for code examples.

Comments

The requested offset is not within the range of offsets maintained by the server

php/7.2.33 ckafka/1.1.1 (https://cloud.tencent.com/document/product/597/11173) longyan/phpkafka 1.1.2

使用examples里的代码运行，producer正常，consumer提示The requested offset is not within the range of offsets maintained by the server

是因为groupId的问题吗
bug

opened by Dawnry 9

不能获取到Header

你遇到了什么问题？

不能获取到Header

Kafka 环境是自建还是云服务？

自建

通过工具查询,头部已经设置上了

请执行下面的命令获取环境信息。

php -v & php --ri swoole & composer info | grep longlang/phpkafka

[1] 1067
[2] 1068
PHP 8.0.6 (cli) (built: Jun 21 2021 15:15:34) ( NTS )
Copyright (c) The PHP Group
Zend Engine v4.0.6, Copyright (c) Zend Technologies
    with Yasd v0.3.8, Our Copyright, by codinghuang

swoole

Swoole => enabled
Author => Swoole Team <[email protected]>
Version => 4.7.0
Built => Aug 12 2021 12:58:54
coroutine => enabled with boost asm context
epoll => enabled
eventfd => enabled
signalfd => enabled
cpu_affinity => enabled
spinlock => enabled
rwlock => enabled
sockets => enabled
openssl => OpenSSL 1.1.1f  31 Mar 2020
dtls => enabled
http2 => enabled
json => enabled
curl-native => enabled
pcre => enabled
zlib => 1.2.11
mutex_timedlock => enabled
pthread_barrier => enabled
futex => enabled
mysqlnd => enabled
async_redis => enabled

Directive => Local Value => Master Value
swoole.enable_coroutine => On => On
swoole.enable_library => On => On
swoole.enable_preemptive_scheduler => Off => Off
swoole.display_errors => On => On
swoole.use_shortname => Off => Off
swoole.unixsock_buffer_size => 8388608 => 8388608
longlang/phpkafka                       v1.1.5     A kafka client. Support php-fpm and Swoole.
[1]-  Done                    php -v
[2]+  Done                    php --ri swoole

提供最小可复现代码：

    public function send(Producer $producer)
    {

        $recordHeader = new RecordHeader();
        $recordHeader->setHeaderKey('test-h');
        $recordHeader->setValue('goods-info');
        $headers = [$recordHeader];
        $producer->send('local_hyperf', 'value', 'key',$headers);
        dump('send...');

    }

/**
 * @Consumer(topic="local_hyperf", nums=1, groupId="hyperf", autoCommit=true)
 */
class DemoKafkaConsumer extends AbstractConsumer
{
    public function consume(ConsumeMessage $message): string
    {
        dump($message->getHeaders());
        dump($message->getValue());
        Log::info(milliseconds() . ' ' . $message->getTopic() . ':' . $message->getKey() . ':' . $message->getValue());
        return Result::ACK;
    }
}

打印$message->getHeaders() 未空

bug enhancement

opened by tw2066 8

消费者过段时间就报错

你遇到了什么问题？ vendor/longlang/phpkafka/src/Client/SwooleClient.php(135): longlang\phpkafka\Socket\SwooleSocket->recv() #1 {main} thrown in /opt/code/novel/vendor/longlang/phpkafka/src/Socket/SwooleSocket.php on line 128 PHP Fatal error: Uncaught longlang\phpkafka\Exception\SocketException: Could not recv data from stream,
Kafka 环境是自建还是云服务？

docker自建

请执行下面的命令获取环境信息。

php -v & php --ri swoole & composer info | grep longlang/phpkafka

# 粘贴到这里

 php -v & php --ri swoole & composer info | grep longlang/phpkafka
[1] 8319
[2] 8320
PHP 7.4.24 (cli) (built: Oct  7 2021 05:44:58) ( NTS )
Copyright (c) The PHP Group
Zend Engine v3.4.0, Copyright (c) Zend Technologies

swoole

Swoole => enabled
Author => Swoole Team <[email protected]>
Version => 4.7.1
Built => Oct  7 2021 05:50:26
coroutine => enabled with boost asm context
epoll => enabled
eventfd => enabled
signalfd => enabled
cpu_affinity => enabled
spinlock => enabled
rwlock => enabled
sockets => enabled
openssl => OpenSSL 1.1.1  11 Sep 2018
dtls => enabled
http2 => enabled
json => enabled
curl-native => enabled
pcre => enabled
zlib => 1.2.11
mutex_timedlock => enabled
pthread_barrier => enabled
futex => enabled
mysqlnd => enabled
async_redis => enabled

Directive => Local Value => Master Value
swoole.enable_coroutine => On => On
swoole.enable_library => On => On
swoole.enable_preemptive_scheduler => Off => Off
swoole.display_errors => On => On
swoole.use_shortname => Off => Off
swoole.unixsock_buffer_size => 8388608 => 8388608
longlang/phpkafka                  v1.2.0    A kafka client. Support php-fpm and Swoole.
[1]-  Done                    php -v
[2]+  Done                    php --ri swoole

* 提供最小可复现代码：

```php
// 你的代码

question

opened by CXW0504 7

support sasl authentication
支持了SASL鉴权

目前只实现了PLAIN方式鉴权。可扩展。

SASL目前是一个数组

$sasl = [ "type"=> longlang\\phpkafka\\Sasl\\PlainSasl::class, // 这个就是确定通过哪个方式鉴权 ... 剩下的信息是鉴权信息，不同的鉴权的类不一样 ]

修改了单元测试。由于kafka的Docker镜像在1.0.0和2.x启动脚本不一样。导致无法环境变量无法兼容。现在kafka的启动脚本是从官方最新的master分支拷贝出来的
opened by qiqizjl 7
[16] This is not the correct coordinator

docker运行 PHP 7.2.29 , swoole 4.5.7 kafka 官方镜像: wurstmeister/kafka:2.12-2.5.0

使用代码为官方demo, produce成功，consumer失败。 Fatal error: Uncaught longlang\phpkafka\Exception\KafkaErrorException: [16] This is not the correct coordinator. in /var/www/html/vendor/longlang/phpkafka/src/Protocol/ ErrorCode.php:385 Stack trace: #0 /var/www/html/vendor/longlang/phpkafka/src/Consumer/OffsetManager.php(78): longlang\phpkafka\Protocol\ErrorCode::check(16) #1 /var/www/html/vendor/longlang/phpkafka/src/Consumer/Consumer.php(70): longlang\phpkafka\Consumer\OffsetManager->updateOffsets() #2 /var/www/html/test/kafka/consumer.php(14): longlang\phpkafka\Consumer\Consumer->__construct(Object(longlang\phpkafka\Consumer\ConsumerConfig)) #3 {main} thrown in /var/www/html/vendor/longlang/phpkafka/src/Protocol/ErrorCode.php on line 385

设置group信息后同样报错
bug

opened by a53abc 7

$Fatal error: Uncaught longlang\phpkafka\Exception\KafkaErrorException$

Fatal error: Uncaught longlang\phpkafka\Exception\KafkaErrorException

你遇到了什么问题？

PHP Warning:  unpack(): Type l: not enough input, need 4, have 0 in /www/vendor/longlang/phpkafka/src/Protocol/Type/Int32.php on line 43

Warning: unpack(): Type l: not enough input, need 4, have 0 in /www/vendor/longlang/phpkafka/src/Protocol/Type/Int32.php on line 43
PHP Fatal error:  Uncaught longlang\phpkafka\Exception\KafkaErrorException: [35] The version of API is not supported. in /www/vendor/longlang/phpkafka/src/Protocol/ErrorCode.php:385
Stack trace:
#0 /www/vendor/longlang/phpkafka/src/Client/SyncClient.php(190): longlang\phpkafka\Protocol\ErrorCode::check()
#1 /www/vendor/longlang/phpkafka/src/Client/SyncClient.php(98): longlang\phpkafka\Client\SyncClient->updateApiVersions()
#2 /www/vendor/longlang/phpkafka/src/Client/SwooleClient.php(52): longlang\phpkafka\Client\SyncClient->connect()
#3 /www/vendor/longlang/phpkafka/src/Broker.php(88): longlang\phpkafka\Client\SwooleClient->connect()
#4 /www/vendor/longlang/phpkafka/src/Producer/Producer.php(39): longlang\phpkafka\Broker->updateBrokers()
#5 /www/vendor/hyperf/kafka/src/Producer.php(173): longlang\phpkafka\Producer\Producer->__construct()
#6 /www/vendor/hyperf/kafka/src/Producer.php(137): Hyperf\Kafka\Producer->makeProducer()
#7 {main}
  thrown in /www/vendor/longlang/phpkafka/src/Protocol/ErrorCode.php on line 385

Fatal error: Uncaught longlang\phpkafka\Exception\KafkaErrorException: [35] The version of API is not supported. in /www/vendor/longlang/phpkafka/src/Protocol/ErrorCode.php:385
Stack trace:
#0 /www/vendor/longlang/phpkafka/src/Client/SyncClient.php(190): longlang\phpkafka\Protocol\ErrorCode::check()
#1 /www/vendor/longlang/phpkafka/src/Client/SyncClient.php(98): longlang\phpkafka\Client\SyncClient->updateApiVersions()
#2 /www/vendor/longlang/phpkafka/src/Client/SwooleClient.php(52): longlang\phpkafka\Client\SyncClient->connect()
#3 /www/vendor/longlang/phpkafka/src/Broker.php(88): longlang\phpkafka\Client\SwooleClient->connect()
#4 /www/vendor/longlang/phpkafka/src/Producer/Producer.php(39): longlang\phpkafka\Broker->updateBrokers()
#5 /www/vendor/hyperf/kafka/src/Producer.php(173): longlang\phpkafka\Producer\Producer->__construct()
#6 /www/vendor/hyperf/kafka/src/Producer.php(137): Hyperf\Kafka\Producer->makeProducer()
#7 {main}
  thrown in /www/vendor/longlang/phpkafka/src/Protocol/ErrorCode.php on line 385
[2021-09-02 02:58:02 *753.1]	ERROR	php_swoole_server_rshutdown() (ERRNO 503): Fatal error: Uncaught longlang\phpkafka\Exception\KafkaErrorException: [35] The version of API is not supported. in /www/vendor/longlang/phpkafka/src/Protocol/ErrorCode.php:385
Stack trace:
#0 /www/vendor/longlang/phpkafka/src/Client/SyncClient.php(190): longlang\phpkafka\Protocol\ErrorCode::check()
#1 /www/vendor/longlang/phpkafka/src/Client/SyncClient.php(98): longlang\phpkafka\Client\SyncClient->updateApiVersions()
#2 /www/vendor/longlang/phpkafka/src/Client/SwooleClient.php(52): longlang\phpkafka\Client\SyncClient->connect()
#3 /www/vendor/longlang/phpkafka/src/Broker.php(88): longlang\phpkafka\Client\SwooleClient->connect()
#4 /www/vendor/longlang/phpkafka/src/Producer/Producer.php(39): longlang\phpkafka\Broker->updateBrokers()
#5 /www/vendor/hyperf/kafka/src/Producer.php(173): longlang\phpkafka\Producer\Producer->__construct()
#6 /www/vendor/hyperf/kafka/src/Producer.php(137): Hyperf\Kafka\Producer->makeProducer()
#7 {main}
  thrown in /www/vendor/longlang/phpkafka/src/Protocol/ErrorCode.php on line 385

其中，“The version of API is not supported.”指明phpkafka不支持，不晓得需要什么具体的版本？

Kafka 环境是自建还是云服务？根据Hypef-2.1手册，使用Docker的hyperf/hyperf:7.4-alpine-v3.13-swoole 镜像建立的容器
请执行下面的命令获取环境信息。

php -v & php --ri swoole & composer info | grep longlang/phpkafka

bash-5.1# php -v & php --ri swoole & composer info | grep longlang/phpkafka
[1] 777
[2] 778
PHP 7.4.23 (cli) (built: Aug 26 2021 23:05:01) ( NTS )
Copyright (c) The PHP Group
Zend Engine v3.4.0, Copyright (c) Zend Technologies
    with Zend OPcache v7.4.21, Copyright (c), by Zend Technologies

swoole

Swoole => enabled
Author => Swoole Team <[email protected]>
Version => 4.7.0
Built => Aug 28 2021 09:20:30
coroutine => enabled with boost asm context
epoll => enabled
eventfd => enabled
signalfd => enabled
spinlock => enabled
rwlock => enabled
openssl => OpenSSL 1.1.1l  24 Aug 2021
dtls => enabled
http2 => enabled
json => enabled
curl-native => enabled
pcre => enabled
zlib => 1.2.11
brotli => E16777225/D16777225
mutex_timedlock => enabled
pthread_barrier => enabled
async_redis => enabled

Directive => Local Value => Master Value
swoole.enable_coroutine => On => On
swoole.enable_library => On => On
swoole.enable_preemptive_scheduler => Off => Off
swoole.display_errors => On => On
swoole.use_shortname => Off => Off
swoole.unixsock_buffer_size => 8388608 => 8388608
longlang/phpkafka                    v1.1.5  A kafka client. Support php-fpm and Swoole.
[1]-  Done                    php -v
[2]+  Done                    php --ri swoole

提供最小可复现代码：

kafka配置

<?php

declare(strict_types=1);

use Hyperf\Kafka\Constants\KafkaStrategy;

return [
    'default'     => [
        'connect_timeout'               => -1,
        'send_timeout'                  => -1,
        'recv_timeout'                  => -1,
        'client_id'                     => '',
        'max_write_attempts'            => 3,
        'bootstrap_servers'             => explode(',', env('KAFKA_LOG_BOOTSTRAP_SERVERS', '')),
        'acks'                          => -1,
        'producer_id'                   => -1,
        'producer_epoch'                => -1,
        'partition_leader_epoch'        => -1,
        'interval'                      => 0,
        'session_timeout'               => 60,
        'rebalance_timeout'             => 60,
        'replica_id'                    => -1,
        'rack_id'                       => '',
        'group_retry'                   => 5,
        'group_retry_sleep'             => 1,
        'group_heartbeat'               => 3,
        'offset_retry'                  => 5,
        'auto_create_topic'             => true,
        'partition_assignment_strategy' => KafkaStrategy::RANGE_ASSIGNOR,
        'pool'                          => [
            'min_connections' => 1,
            'max_connections' => 10,
            'connect_timeout' => 10.0,
            'wait_timeout'    => 3.0,
            'heartbeat'       => -1,
            'max_idle_time'   => 60.0,
        ],
    ],
    'log_service' => [
        'app'       => env('KAFKA_LOG_APP', ''),
        'topic'     => env('KAFKA_LOG_TOPIC', '--topic--’),
        'bootstrap' => env('KAFKA_LOG_BOOTSTRAP_SERVERS', '--bootstrap--‘),
    ],
];

<?php
declare(strict_types=1);

namespace App\Service\QueueService;

use App\Service\PageService\BasePS;
use Hyperf\Contract\ConfigInterface;
use Hyperf\HttpServer\Contract\RequestInterface;
use Hyperf\Kafka\Producer;
use Hyperf\Logger\LoggerFactory;
use Hyperf\Contract\ContainerInterface;

class KafkaQS
{

    protected $producer;
    protected $logger;
    protected $config;
    protected $container;
    protected $request;

    public function __construct(ContainerInterface $container, Producer $producer, RequestInterface $request)
    {
        $this->producer  = $producer;
        $this->container = $container;
        $this->config    = $container->get(ConfigInterface::class)->get('kafka.log_service');
    }

    /**
     * elkLog
     *
     * @param array $param
     * @return bool
     */
    public function elkLog(array $param = []): bool
    {
        $topic       = $this->config['topic'];
        $url         = $this->config['bootstrap'];

        $logStr = 'test demo';
        $this->producer->send($topic, $logStr, $url);
        return true;
    }

}

question

opened by dmtnewton 6

只能连本机的9092端口,连接远程会提示Could not connect to tcp://localhost:9092

你遇到了什么问题？

只能连本机的9092端口,连接远程会提示:Could not connect to tcp://localhost:9092 (Connection refused [111]) File:...../vendor/longlang/phpkafka/src/Socket/SwooleSocket.php Line:95 我的ip和端口号都不是localhost和9092,不知道为什么还是会连接到localhost:9092上

Kafka 环境是自建还是云服务？

自建,尝试使用华为云的kafka会报一个内存溢出的错误,所以先放弃了,华为云的提示为allowed memory size of 134... bytes exhausted... File:../longlang/phpkafka/src/Broker.php Line136

请执行下面的命令获取环境信息。

php -v & php --ri swoole & composer info | grep longlang/phpkafka

# 粘贴到这里

[1] 22822
[2] 22823
PHP 7.3.27 (cli) (built: Apr 10 2021 14:10:50) ( NTS )
Copyright (c) 1997-2018 The PHP Group
Zend Engine v3.3.27, Copyright (c) 1998-2018 Zend Technologies
    with Zend OPcache v7.3.27, Copyright (c) 1999-2018, by Zend Technologies

swoole

Swoole => enabled
Author => Swoole Team <[email protected]>
Version => 4.6.5
Built => Apr 20 2021 10:16:05
coroutine => enabled with boost asm context
epoll => enabled
eventfd => enabled
signalfd => enabled
cpu_affinity => enabled
spinlock => enabled
rwlock => enabled
zlib => 1.2.11
mutex_timedlock => enabled
pthread_barrier => enabled
futex => enabled
async_redis => enabled

Directive => Local Value => Master Value
swoole.enable_coroutine => On => On
swoole.enable_library => On => On
swoole.enable_preemptive_scheduler => Off => Off
swoole.display_errors => On => On
swoole.use_shortname => On => On
swoole.unixsock_buffer_size => 8388608 => 8388608
Composer could not find a composer.json file in /usr/src/app/example
To initialize a project, please create a composer.json file as described in the https://getcomposer.org/ "Getting Started" section
[1]-  Done                    php -v
[2]+  Done                    php --ri swoole

* 提供最小可复现代码：

```php
        $config = new ProducerConfig();
        $config->setBootstrapServer(['124.xxx.xxx.xxx,9095']);
        $config->setUpdateBrokers(true);
        $config->setAcks(-1);
        $producer = new Producer($config);
        $topic = 'test333';
        $value = (string) microtime(true);
        $key = uniqid('', true);
        $producer->send($topic, $value, $key);

question

opened by xilin693 4

请问什么原因导致了fetchMessages 中 $currentList 为空

[ERROR] App\Exception\Handler\ErrorHandler: Invalid argument supplied for foreach()(2) in /Users/zhaoxingyan/data/hives-hyperf/vendor/longlang/phpkafka/src/Consumer/Consumer.php:293 Stack trace: #0 /Users/zhaoxingyan/data/hives-hyperf/vendor/longlang/phpkafka/src/Consumer/Consumer.php(293): ErrorHandler(2, 'Invalid argumen...', '/Users/zhaoxing...', 293, Array) #1 /Users/zhaoxingyan/data/hives-hyperf/vendor/longlang/phpkafka/src/Consumer/Consumer.php(225): longlang\phpkafka\Consumer\Consumer->fetchMessages() #2 /Users/zhaoxingyan/data/hives-hyperf/vendor/longlang/phpkafka/src/Consumer/Consumer.php(203): longlang\phpkafka\Consumer\Consumer->consume() #3 /Users/zhaoxingyan/data/hives-hyperf/vendor/hyperf/kafka/src/ConsumerManager.php(150): longlang\phpkafka\Consumer\Consumer->start() #4 /Users/zhaoxingyan/data/hives-hyperf/vendor/hyperf/utils/src/Functions.php(80): class@anonymous->Hyperf\Kafka{closure}() #5 /Users/zhaoxingyan/data/hives-hyperf/vendor/hyperf/kafka/src/ConsumerManager.php(157): retry(0, Object(Closure), 10) #6 /Users/zhaoxingyan/data/hives-hyperf/vendor/hyperf/process/src/AbstractProcess.php(134): class@anonymous->handle()
question

opened by suibianmzl 4
支持设置producer端参数么
你遇到了什么问题？请问支持设置以下kafka的producer端配置么。 1 retries 就是producer重试的次数设置 2 max.in.flight.requests.per.connection producer的IO线程在单个Socket连接上能够发送未应答produce请求的最大数量

在代码里面没看到有地方可以设置
opened by nazrydirunn 0
updateApiVersions 报错 [ERROR] [10753] Unknown[385]
你遇到了什么问题？报错日志如下，多发于hyperf一段时间无访问时再次访问出错。频繁访问没有报错 PHP Warning: Uninitialized string offset 0 in /www/hyperf-skeleton/vendor/longlang/phpkafka/src/Protocol/Type/UVarInt.php on line 41

Warning: Uninitialized string offset 0 in /www/hyperf-skeleton/vendor/longlang/phpkafka/src/Protocol/Type/UVarInt.php on line 41 [ERROR] [10753] Unknown[385] in /www/hyperf-skeleton/vendor/longlang/phpkafka/src/Protocol/ErrorCode.php [ERROR] #0 /www/hyperf-skeleton/vendor/longlang/phpkafka/src/Client/SyncClient.php(196): longlang\phpkafka\Protocol\ErrorCode::check(10753) #1 /www/hyperf-skeleton/vendor/longlang/phpkafka/src/Client/SyncClient.php(103): longlang\phpkafka\Client\SyncClient->updateApiVersions() #2 /www/hyperf-skeleton/vendor/longlang/phpkafka/src/Client/SwooleClient.php(52): longlang\phpkafka\Client\SyncClient->connect() #3 /www/hyperf-skeleton/vendor/longlang/phpkafka/src/Broker.php(174): longlang\phpkafka\Client\SwooleClient->connect() #4 /www/hyperf-skeleton/vendor/longlang/phpkafka/src/Broker.php(156): longlang\phpkafka\Broker->getClientByBrokerId(1001) #5 /www/hyperf-skeleton/vendor/longlang/phpkafka/src/Producer/Producer.php(160): longlang\phpkafka\Broker->getClient(1001) #6 /www/hyperf-skeleton/vendor/hyperf/kafka/src/Producer.php(94): longlang\phpkafka\Producer\Producer->sendBatch(Array) #7 [internal function]: Hyperf\Kafka\Producer->Hyperf\Kafka{closure}() #8 /www/hyperf-skeleton/vendor/hyperf/kafka/src/Producer.php(148): Closure->call(Object(Hyperf\Kafka\Producer)) #9 {main}

Kafka 环境是自建还是云服务？ docker环境 cat docker-compose.yml

version: '2' services: zookeeper: image: zookeeper:3.5 ports: - "2182:2181" kafka: image: wurstmeister/kafka depends_on: [ zookeeper ] ports: - "9092:9092" environment: KAFKA_ADVERTISED_HOST_NAME: 192.168.0.90 KAFKA_CREATE_TOPICS: "cps_static:1:1" KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 #KAFKA_HOME: "/opt/kafka_2.12-2.4.1" volumes: - /var/run/docker.sock:/var/run/docker.sock - /usr/share/zoneinfo/Asia/Shanghai:/etc/localtime kafka-manager: image: kafkamanager/kafka-manager depends_on: - zookeeper ports: - "8084:9000" environment: ZK_HOSTS: zookeeper:2181

请执行下面的命令获取环境信息。

php -v & php --ri swoole & composer info | grep longlang/phpkafka

[1] 72185 [2] 72186 PHP 8.0.12 (cli) (built: Oct 21 2021 14:38:26) ( NTS ) Copyright (c) The PHP Group Zend Engine v4.0.12, Copyright (c) Zend Technologies with Yasd v0.3.9-alpha, Our Copyright, by codinghuang [1] - 72185 done php -v swoole Swoole => enabled Author => Swoole Team <[email protected]> Version => 5.0.0 Built => Aug 23 2022 18:14:26 coroutine => enabled with boost asm context kqueue => enabled rwlock => enabled openssl => OpenSSL 3.0.0 7 sep 2021 dtls => enabled http2 => enabled json => enabled curl-native => enabled pcre => enabled zlib => 1.2.11 brotli => E16777225/D16777225 mysqlnd => enabled async_redis => enabled Directive => Local Value => Master Value swoole.enable_coroutine => On => On swoole.enable_library => On => On swoole.enable_preemptive_scheduler => Off => Off swoole.display_errors => On => On swoole.use_shortname => Off => Off swoole.unixsock_buffer_size => 262144 => 262144 [2] + 72186 done php --ri swoole longlang/phpkafka v1.2.1 A kafka client. Support php-fpm and Swoole.

提供最小可复现代码：

无法复现
opened by bearzlh 0

Invalid length 352518912 given, it should be lesser than or equals to 5242880

你遇到了什么问题？

prod.ERROR: Invalid length 352518912 given, it should be lesser than or equals to 5242880 {"exception":"[object] (longlang\phpkafka\Exception\SocketException(code: 0): Invalid length 352518912 given, it should be lesser than or equals to 5242880 at /web/vendor/longlang/phpkafka/src/Socket/StreamSocket.php:173) [stacktrace]

Kafka 环境是自建还是云服务？云
请执行下面的命令获取环境信息。

php -v & php --ri swoole & composer info | grep longlang/phpkafka

# 粘贴到这里
[1] 7531
[2] 7532
PHP 7.4.30 (cli) (built: Jul 12 2022 09:31:30) ( NTS )
Copyright (c) The PHP Group
Zend Engine v3.4.0, Copyright (c) Zend Technologies

swoole

Swoole => enabled
Author => Swoole Team <[email protected]>
Version => 4.4.26
Built => Jul 27 2022 03:11:37
coroutine => enabled
epoll => enabled
eventfd => enabled
signalfd => enabled
cpu_affinity => enabled
spinlock => enabled
rwlock => enabled
openssl => OpenSSL 1.1.1n  15 Mar 2022
zlib => 1.2.11
mutex_timedlock => enabled
pthread_barrier => enabled
futex => enabled
async_redis => enabled

Directive => Local Value => Master Value
swoole.enable_coroutine => On => On
swoole.enable_library => On => On
swoole.enable_preemptive_scheduler => Off => Off
swoole.display_errors => On => On
swoole.use_shortname => On => On
swoole.unixsock_buffer_size => 8388608 => 8388608
Do not run Composer as root/super user! See https://getcomposer.org/root for details

提供最小可复现代码：

// 你的代码
 public function recv(int $length, ?float $timeout = null): string
    {
        if ($length > self::READ_MAX_LENGTH) {
            throw new SocketException(sprintf('Invalid length %d given, it should be lesser than or equals to %d', $length, self::READ_MAX_LENGTH));
        }

        if (null === $timeout) {
            $timeout = $this->config->getRecvTimeout();
        }
        $readable = $this->select([$this->socket], $timeout);

        if (false === $readable) {
            $this->close();
            throw new SocketException(sprintf('Could not read %d bytes from stream (not readable)', $length));
        }

        if (0 === $readable) { // select timeout
            $res = $this->getMetaData();
            $this->close();

            if (!empty($res['timed_out'])) {
                throw new SocketException(sprintf('Timed out reading %d bytes from stream', $length));
            }

            throw new SocketException(sprintf('Could not read %d bytes from stream (not readable)', $length));
        }

        $remainingBytes = $length;
        $data = $chunk = '';

        while ($remainingBytes > 0) {
            $chunk = fread($this->socket, $remainingBytes);

            if (false === $chunk || 0 === \strlen($chunk)) {
                // Zero bytes because of EOF?
                if (feof($this->socket)) {
                    $this->close();
                    throw new SocketException(sprintf('Unexpected EOF while reading %d bytes from stream (no data)', $length));
                }
                // Otherwise wait for bytes
                $readable = $this->select([$this->socket], $timeout);
                if (1 !== $readable) {
                    $this->close();
                    throw new SocketException(sprintf('Timed out while reading %d bytes from stream, %d bytes are still needed', $length, $remainingBytes));
                }

                continue; // attempt another read
            }

            $data .= $chunk;
            $remainingBytes -= \strlen($chunk);
        }

        return $data;
    }

opened by WeiDaDe 1

在swoole服务中, kafka生产者每过一段时间就报错

你遇到了什么问题？在swoole服务中, kafka生产者每过一段时间就报错
Kafka 环境是自建还是云服务？自建
请执行下面的命令获取环境信息。

php -v & php --ri swoole & composer info | grep longlang/phpkafka

PHP 7.4.25 (cli) (built: Oct 19 2021 15:18:10) ( NTS )
Copyright (c) The PHP Group
Zend Engine v3.4.0, Copyright (c) Zend Technologies
    with Zend OPcache v7.4.25, Copyright (c), by Zend Technologies

swoole

Swoole => enabled
Author => Swoole Team <[email protected]>
Version => 4.6.1
Built => Jan 11 2021 12:30:02
coroutine => enabled with boost asm context
trace_log => enabled
epoll => enabled
eventfd => enabled
signalfd => enabled
cpu_affinity => enabled
spinlock => enabled
rwlock => enabled
sockets => enabled
openssl => OpenSSL 1.0.2k-fips  26 Jan 2017
http2 => enabled
json => enabled
curl-native => enabled
pcre => enabled
zlib => 1.2.7
mutex_timedlock => enabled
pthread_barrier => enabled
futex => enabled
mysqlnd => enabled
async_redis => enabled

Directive => Local Value => Master Value
swoole.enable_coroutine => On => On
swoole.enable_library => On => On
swoole.enable_preemptive_scheduler => Off => Off
swoole.display_errors => On => On
swoole.use_shortname => On => On
swoole.unixsock_buffer_size => 8388608 => 8388608

提供最小可复现代码：

// 你的代码
declare(strict_types=1);//语法严格模式
require_once dirname(__DIR__).'/vendor/autoload.php';
ini_set('memory_limit', '2048M');
use longlang\phpkafka\Producer\Producer;
use longlang\phpkafka\Producer\ProducerConfig;

$http = new Swoole\Http\Server("0.0.0.0", 9503);
$http->set([
    'tcp_fastopen' => true,
    'max_wait_time' => 1,
    'enable_reuse_port'=>false,//同一个端口复用
    'log_file'=>dirname(__DIR__)."/log/swoole.log"
]);

$config = new ProducerConfig();
$config->setBootstrapServer('172.18.50.74:9092,172.18.50.75:9092');
$config->setUpdateBrokers(true);
$config->setAcks(-1);
$producer = new Producer($config);
$http->on('request', function ($request, $response) use ($producer) {
    try {
        $topic = 'play_history_pre3';
        var_dump($request->server['request_uri']);
        $sendData = json_encode($request->post);
        $producer->send($topic,$sendData);
        unset($sendData,$topic);
        $response->end('success');
    } catch (\Throwable $e) {
        var_dump($e->getMessage());
        $response->end(json_encode(array(
            'code' => '999',
            'message' => '记录失败'
        )));
    }
});

$http->start();

错误信息

PHP Fatal error:  Uncaught longlang\phpkafka\Exception\SocketException: Could not recv data from stream,  [0] in /data/www/kafka/swoole/vendor/longlang/phpkafka/src/Socket/SwooleSocket.php:129
Stack trace:
#0 /data/www/kafka/swoole/vendor/longlang/phpkafka/src/Client/SwooleClient.php(135): longlang\phpkafka\Socket\SwooleSocket->recv(4, -1)
#1 {main}
  thrown in /data/www/kafka/swoole/vendor/longlang/phpkafka/src/Socket/SwooleSocket.php on line 129

opened by insomniazz 5

Releases(v1.2.1)

v1.2.1(Jan 10, 2022)
Update Log:

Fixed

Fix compatibility with PHP 8.1 (#59)

Optimization

Optimize socket close and reconnect (#58)

更新日志：

修复

修复 PHP 8.1 兼容性 (#59)

优化

优化 Socket 的关闭和重连 (#58)

Source code(tar.gz)
Source code(zip)
v1.2.0(Sep 17, 2021)
Update Log:

Enhancement

Support SASL authentication (#47)

Support SSL (#50)

Producer send with header support key-value and RecordHeader (#55)

Fixed

Fix VarIntCompactArray and consumer not found headers

Optimization

更新日志：

增强

支持 SASL (#47)

支持 SSL (#50)

发布者发送头时，同时支持键值数组和 RecordHeader 数组 (#55)

修复

修复 VarIntCompactArray 导致的消费者消费获取不到头的问题

Source code(tar.gz)
Source code(zip)
v1.1.5(Aug 4, 2021)
Update Log:

Fixed

Fix When acks is set to 0, data is inserted repeatedly (#42)

Fix use currentLeaderEpoch when updateListOffsets (#44)

Optimization

Throw NoAliveBrokerException when no brokers are available. (#43)

Optimize (#41)

更新日志：

修复

修复当 acks 设为 0 时，重复插入消息的问题 (#42)

修复 updateListOffsets() 时使用 currentLeaderEpoch (#44)

优化

当没有可用的代理时抛出NoAliveBrokerException (#43)

优化 (#41)

Source code(tar.gz)
Source code(zip)
v1.1.4(May 21, 2021)
Update Log:

Enhancement

Add produceRetry, produceRetrySleep in ProducerConfig

Fixed

Fix typo

Fix array type is null #38

Fix producing messages to an uncreated topic

更新日志：

增强

在 ProducerConfig 中增加了 produceRetry 和 produceRetrySleep

修复

修复拼写问题

修复数组类型为 null 时返回报错问题

修复给未创建的主题生产消息问题

Source code(tar.gz)
Source code(zip)
v1.1.3(Apr 2, 2021)
Update Log:

Enhancement

Add Coding Standards php-cs-fixer

Add phpstan

Add minBytes、maxBytes、maxWait in ConsumerConfig

Add brokers in ConsumerConfig. The broker is alias of brokers now.

Add ISSUE_TEMPLATE

Optimization

Optimize Exception message

Consumer error handling rejoin

Fixed

Fix a possible error

Fix RangeAssignor and RoundRobinAssignor

Fix consumer when after log cleanup

Fix saveOffsets

更新日志：

增强

增加代码标准检测 php-cs-fixer

增加静态分析工具 phpstan

在 ConsumerConfig 中增加 minBytes、maxBytes、maxWait

在 ConsumerConfig 中增加 brokers。broker 现在是 brokers 的别名了

增加 issue 提问模版

优化

优化异常消息文字

部分消费错误时 rejoin()

修复

修复一处可能的错误

修复 RangeAssignor 和 RoundRobinAssignor

修复清理日志后的消费问题

修复保存偏移量

Source code(tar.gz)
Source code(zip)
v1.1.2(Feb 7, 2021)
Update Log:

Enhancement

Add exceptionCallback in CommonConfig

Support producer partition strategy

Optimization

Add English document

Fixed

Fix: dead loop (#22)

Fix ConsumeMessage $value type (#23)

Fix problems with multiple brokers (#24)

更新日志：

增强

在 CommonConfig 中增加了 exceptionCallback

支持分区分配策略

优化

增加了英文文档说明

修复

修复死循环

修复 ConsumeMessage $value 的类型

修复多 broker 情况下的生产和消费问题

Source code(tar.gz)
Source code(zip)
v1.1.1(Jan 18, 2021)
增强

支持消费者分区分配策略之粘性分配-\longlang\phpkafka\Consumer\Assignor\StickyAssignor

消费者现在也会更新 brokers 了（#12）

消费者、生产者配置都增加了 bootstrapServers，原 bootstrapServer 作为别名，依然可以使用

优化

优化了一些代码

修复

修复 ConsumerConfig 中 broker 为数组时的问题（#10）

修复 Broker->getClient()（#11）

修复 PHP <= 7.3 中使用时，会将 phpunit 也引入进来的问题

修复使用对应的 broker 发送分组消息（#12）

Source code(tar.gz)
Source code(zip)
v1.1.0(Dec 24, 2020)
增强

支持 PHP 8.0

支持自动创建主题

支持同时消费多个主题

支持消费者分区分配策略

分区分配策略：范围分配-longlang\phpkafka\Consumer\Assignor\RangeAssignor 轮询分配-\longlang\phpkafka\Consumer\Assignor\RoundRobinAssignor 粘性分配（暂未支持）-\longlang\phpkafka\Consumer\Assignor\StickyAssignor | longlang\phpkafka\Consumer\Assignor\RangeAssignor

优化

优化代码

修复

修复 RecordBatch

修复类型错误 (https://github.com/longyan/phpkafka/pull/8)

不兼容更改

废弃 ConsumerConfig 的 partitions 参数，转而使用分配策略

手动 ack() 写法更改，请参考文档

Source code(tar.gz)
Source code(zip)
v1.0.1(Nov 24, 2020)
增强

支持分组操作重试机制

支持偏移操作重试机制

支持心跳机制

优化

Added code for KafkaErrorException (#4) (@limingxinleo)

优化代码 (#5) (@PandaLIU-1111)

增加测试用例

修复

修复消费者逻辑问题 #2

修复消费者内存泄漏问题 #3

修复消费多个分区时报 crc32 验证错误问题

Source code(tar.gz)
Source code(zip)
v1.0.0(Nov 9, 2020)

Source code(tar.gz)
Source code(zip)