网友回复
设置('enable.auto.commit', 'false');
消费完数据后进行 $consumer->commit($message);
示例代码如下:
<?php
$conf = new RdKafka\Conf();
// Set a rebalance callback to log partition assignments (optional)
// 当有新的消费进程加入或者退出消费组时,kafka 会自动重新分配分区给消费者进程,这里注册了一个回调函数,当分区被重新分配时触发
$conf->setRebalanceCb(function (RdKafka\KafkaConsumer $kafka, $err, array $partitions = null) {
switch ($err) {
case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
echo "Assign: ";
var_dump($partitions);
$kafka->assign($partitions);
break;
case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
echo "Revoke: ";
var_dump($partitions);
$kafka->assign(NULL);
break;
default:
...点击查看剩余70%
有没有兼容Puppeteer和 Playwright使用的docker独立chrome浏览器?
geo与seo区别?
chrome插件能否实现网页远程控制鼠标选择网页文字滚动网页?
nativescript开发的安卓与ios app应用是原生的吗?
go如何写一个类似redis的nosql数据库让python客户端调用?
php7中为啥无法使用$_SERVER['HTTP_RAW_POST_DATA'] ?
chrome插件能否实现2台电脑的远程协助桌面控制?
python如何通过webrtc网页远程协助控制python运行的电脑?
systemd-nspawn与unshare区别?
Firejail、Bubblewrap沙箱与docker、podman容器区别?


