This repository has been archived by the owner on Jun 10, 2022. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 451
在swoole中,作为常驻进程,如何使用该扩展呢? #260
Comments
我也遇到这个问题,刚开始很好,但是空闲一段时间后,下一次访问就会报错,并且导致Worker退出,我猜想可能是Client丢失了连接
|
我试过php-rdkafka,像我上面那样写,多个请求共享一个连接,Kafka收不到数据,你是怎么解决共享连接的问题? |
下面这个是我现在使用的:
}` 然后再swoole中使用的方式: 每个task进程都有一个kafka连接,在task进程里面是可以一直使用这个连接的。 |
@laiason 感谢,我的方法是把连接放在server上,同步阻塞的方式投递消息,我觉得使用同步方式投递消息是可以接受的,kafka本来就是一个队列系统,投递消息的代价很小,没有必要把投递消息的过程放到另一个进程里
|
<?php
$http = new Swoole\Http\Server('127.0.0.1', 5200);
$http->on('workerStart', function (Swoole\Http\Server $server, $workerId) {
global $topic;
$brokerList = '192.168.15.83:9092';
$topic = 'bi_test';
$conf = new RdKafka\Conf();
$conf->set('queue.buffering.max.messages', 1000000);
$rk = new RdKafka\Producer($conf);
$rk->setLogLevel(LOG_WARNING);
$rk->addBrokers($brokerList);
$topic = $rk->newTopic($topic);
});
$http->on('request', function (Swoole\Http\Request $request, Swoole\Http\Response $response) {
global $topic;
$payload = 'Message payload: ' . date('Y-m-d H:i:s ') . microtime(true);
$response->end($payload);
$topic->produce(RD_KAFKA_PARTITION_UA, 0, $payload);
});
$http->start(); |
rdkafka的包太大了,打包docker的话要多出几百M |
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
最近使用php swoole做项目,启动一个常驻进程,需要把接口日志发送到kafka,问题如下:
1、在github上的示例,异步和同步的生成者方式每次都是new一个生成者。那么在常驻进程中,怎么实现一个生产者连接池,不需要每次都是new一个生成者。
2、使用异步生成者方式进行测试,发现每次会发送多次数据到kafka,这个是怎么回事呢?
3、使用同步方式,把kafka生产者赋给一个类属性,这样不需要每次都去new一个生产者,但是发现会报如下的错误:
Fatal error: Uncaught Kafka\Exception\Socket: After 4 attempts could not write 444 bytes to stream, completed writing only 0 bytes in kafka/nmred/kafka-php/src/Kafka/SocketSync.php:380
The text was updated successfully, but these errors were encountered: