首先启动Zookeeper，再启动Kafka服务，通过jps检查进程，并使用kafka-topics.sh创建主题，通过生产者和消费者命令行工具测试消息收发，确认服务正常；随后安装librdkafka库和rdkafka PHP扩展，配置php.ini启用扩展，最后通过PHP代码使用RdKafkaProducer和RdKafkaKafkaConsumer类实现消息的发送与接收，关键参数包括bootstrap.servers、group.id、acks、enable.auto.commit等，确保通信正常与消息可靠性。
在CentOS上配置Kafka并搭建PHP客户端，核心步骤包括安装Java环境、部署Zookeeper和Kafka服务器本身，随后在PHP环境中安装并配置
librdkafka
库以及
rdkafka
PHP扩展。这是一个相对直接的过程，但细节处理不当很容易遇到各种坑。
解决方案
搭建Kafka和PHP客户端，我们通常会按部就班地来。我个人觉得，先搞定服务端，再考虑客户端，这样思路会比较清晰。
1. Java环境准备Kafka是基于JVM运行的，所以Java是必不可少的。我一般会直接用OpenJDK。
sudo yum install java-1.8.0-openjdk java-1.8.0-openjdk-devel -yjava -version # 检查版本，确保是Java 8或更高
确保
JAVA_HOME
环境变量也设置好了，虽然很多时候系统会自动处理，但明确一下总是好的。
2. Zookeeper安装与配置Kafka依赖Zookeeper来管理集群元数据。
# 下载Zookeeper，我通常会去Apache官网找最新的稳定版wget https://downloads.apache.org/zookeeper/zookeeper-3.8.3/apache-zookeeper-3.8.3-bin.tar.gztar -zxvf apache-zookeeper-3.8.3-bin.tar.gzsudo mv apache-zookeeper-3.8.3-bin /opt/zookeepercd /opt/zookeeper/confcp zoo_sample.cfg zoo.cfg
编辑
zoo.cfg
，主要关注
dataDir
，我习惯放在
/var/lib/zookeeper
。
立即学习“PHP免费学习笔记（深入）”；
# zoo.cfgdataDir=/var/lib/zookeeperclientPort=2181
创建数据目录：
sudo mkdir -p /var/lib/zookeeper
启动Zookeeper：
/opt/zookeeper/bin/zkServer.sh start
你可以通过
jps
命令查看是否有
QuorumPeerMain
进程，或者用
zkCli.sh -server 127.0.0.1:2181
连接测试。
3. Kafka安装与配置现在轮到Kafka本身了。
# 下载Kafka，同样去官网找最新稳定版wget https://downloads.apache.org/kafka/3.6.1/kafka_2.13-3.6.1.tgztar -zxvf kafka_2.13-3.6.1.tgzsudo mv kafka_2.13-3.6.1 /opt/kafka
编辑Kafka的配置文件
server.properties
：
cd /opt/kafka/configvim server.properties
几个关键配置：
broker.id
: 每个Kafka实例的唯一ID，集群中不能重复。
listeners
: 监听地址，我通常设成
PLAINTEXT://:9092
或
PLAINTEXT://你的IP地址:9092
。如果只是本机测试，
0.0.0.0:9092
也行。
log.dirs
: Kafka存储消息日志的目录，建议独立挂载磁盘。
zookeeper.connect
: Zookeeper地址，比如
localhost:2181
。
启动Kafka：
/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties
同样可以用
jps
查看是否有
Kafka
进程。
4. PHP客户端安装与配置PHP与Kafka交互主要通过
librdkafka
C库和
rdkafka
PHP扩展。
安装
librdkafka
：这个库是核心，PHP扩展只是它的一个包装。
# 安装依赖sudo yum install gcc-c++ make zlib-devel openssl-devel -y# 下载librdkafka，通常用最新稳定版git clone https://github.com/confluentinc/librdkafka.gitcd librdkafka./configuremakesudo make install
如果遇到权限问题，
sudo make install
可能会报错，确保
/usr/local/lib
等路径是可写的，或者调整安装路径。安装完成后，更新共享库缓存：
sudo ldconfig
安装
rdkafka
PHP扩展：
# 确保你安装了php-devel，这是编译PHP扩展的必要组件sudo yum install php-devel -y# 通过PECL安装rdkafkasudo pecl install rdkafka
安装过程中，它可能会问你
librdkafka
的安装路径，通常直接回车默认即可，因为它会尝试在标准路径找到。
安装成功后，需要在
php.ini
中启用这个扩展。
echo "extension=rdkafka.so" | sudo tee -a /etc/php.ini # 或者你的php.ini路径
重启PHP-FPM或Apache/Nginx服务，让配置生效。
sudo systemctl restart php-fpm # 或 apache/nginx
检查扩展是否加载成功：
php -m | grep rdkafka
如果能看到
rdkafka
，那就说明客户端环境也搞定了。
如何在CentOS系统上启动并测试Kafka服务？
启动Kafka服务，说起来不复杂，但需要注意顺序。首先，Zookeeper必须先跑起来，因为Kafka启动时会去连接它。我通常的习惯是先确认Zookeeper服务是健康的。
你可以用
zkServer.sh status
命令来检查Zookeeper的状态，确保它处于
Mode: standalone
或
Mode: follower/leader
。如果Zookeeper没问题，接着就可以启动Kafka了。
我们之前已经通过
kafka-server-start.sh -daemon /opt/kafka/config/server.properties
让Kafka在后台运行了。要确认它是否真的跑起来，最直接的方法是看进程：
琅琅配音
全能AI配音神器
208 查看详情
jps
如果看到
QuorumPeerMain
（Zookeeper）和
Kafka
这两个进程，那基本上就没问题了。
命令行测试Kafka：Kafka自带了一些命令行工具，非常适合做初步的功能测试。
创建Topic：
/opt/kafka/bin/kafka-topics.sh --create --topic my_test_topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
这里
--bootstrap-server
就是你Kafka监听的地址。
查看Topic列表：
/opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092
你应该能看到
my_test_topic
。
生产者测试：
/opt/kafka/bin/kafka-console-producer.sh --topic my_test_topic --bootstrap-server localhost:9092
输入一些消息，比如
Hello Kafka from Producer!
，然后回车。
消费者测试：打开另一个终端窗口，运行消费者：
/opt/kafka/bin/kafka-console-consumer.sh --topic my_test_topic --bootstrap-server localhost:9092 --from-beginning
你将看到生产者发送的消息。如果这些都正常工作，那么恭喜你，Kafka服务在CentOS上已经成功启动并可以进行基本的数据交互了。这套测试流程，我每次部署都会走一遍，确保基础功能没问题。
PHP应用如何与Kafka进行生产者和消费者交互？
PHP应用与Kafka的交互，核心就是使用
rdkafka
扩展提供的API。它的设计理念其实很直接，无非就是初始化一个生产者或消费者实例，然后进行消息的发送或接收。
PHP生产者示例：
set('bootstrap.servers', 'localhost:9092'); // Kafka broker地址$conf->set('client.id', 'php-producer-app'); // 客户端ID，方便调试// 设置消息发送失败时的回调，这对于生产环境很重要$conf->setDrMsgCb(function ($kafka, $message) { if ($message->err) { // 这里可以记录日志，或者进行重试逻辑 error_log(sprintf("Message delivery failed: %s (%s)", $message->errstr(), $message->err)); } else { // error_log(sprintf("Message delivered to topic %s [%d] at offset %d", $message->topic_name, $message->partition, $message->offset)); }});// 创建生产者实例$producer = new RdKafkaProducer($conf);// 获取一个Topic的生产器$topic = $producer->newTopic("my_test_topic"); // 替换成你的Topic名称$message = "Hello from PHP Producer! " . date('Y-m-d H:i:s');$key = "my_key"; // 消息的key，用于分区// 发送消息// RD_KAFKA_PARTITION_UA 表示由librdkafka自动选择分区$topic->produce(RD_KAFKA_PARTITION_UA, 0, $message, $key);echo "Sent message: " . $message . "n";// 循环调用poll，等待消息发送结果回调// 这是一个非阻塞的等待，实际应用中可能需要更复杂的循环或异步处理for ($i = 0; $i poll(100); // Poll for 100ms if ($producer->getOutQLen() === 0) { // 如果发送队列为空，说明消息已发出 break; }}if ($producer->getOutQLen() > 0) { echo "Still " . $producer->getOutQLen() . " messages in queue, waiting for delivery.n"; // 强制等待所有消息发送完成 $producer->flush(10000); // 最多等待10秒}if ($producer->getOutQLen() === 0) { echo "All messages delivered.n";} else { echo "Some messages failed to deliver.n";}?>
这里
$producer->poll()
和
$producer->flush()
是关键，它们负责处理内部事件和消息的回调。我第一次用的时候，就因为没调用
poll
，导致消息一直发不出去，排查了好久。
PHP消费者示例：
set('bootstrap.servers', 'localhost:9092');$conf->set('group.id', 'php-consumer-group'); // 消费者组ID// 设置自动提交offset，也可以手动控制$conf->set('enable.auto.commit', 'true');$conf->set('auto.offset.reset', 'earliest'); // 如果没有offset，从最早的消息开始消费// 设置错误回调$conf->setErrorCb(function ($kafka, $err, $reason) { error_log(sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason));});// 创建消费者实例$consumer = new RdKafkaKafkaConsumer($conf);// 订阅Topic$consumer->subscribe(['my_test_topic']); // 可以订阅多个Topicecho "Waiting for messages... (Press Ctrl+C to stop)n";while (true) { $message = $consumer->consume(120*1000); // 最多等待120秒 switch ($message->err) { case RD_KAFKA_RESP_ERR_NO_ERROR: // 收到消息 echo sprintf("Message received: Topic %s, Partition %d, Offset %d, Key %s, Payload: %sn", $message->topic_name, $message->partition, $message->offset, $message->key, $message->payload ); // 如果是手动提交offset，这里需要调用 $consumer->commit($message); break; case RD_KAFKA_RESP_ERR__PARTITION_EOF: // 到达分区末尾，但没有新消息 echo "No more messages, waiting for new ones...n"; break; case RD_KAFKA_RESP_ERR__TIMED_OUT: // 等待超时，没有收到消息 echo "Waiting for messages timed out...n"; break; default: // 其他错误 error_log(sprintf("Consumer error: %s (%s)", $message->errstr(), $message->err)); break; }}?>
消费者这边，
$consumer->consume()
是阻塞的，它会一直等待直到有消息或者超时。
group.id
非常重要，它决定了消费者在哪个组里，以及消息的分配策略。我建议在生产环境中使用手动提交offset，这样可以更好地控制消息的处理逻辑，避免重复消费或消息丢失。
配置Kafka PHP客户端时，有哪些关键参数需要注意？
配置
rdkafka
PHP客户端时，有些参数确实是“重中之重”，它们直接影响到客户端的行为、性能以及消息的可靠性。我个人在实践中，最常关注和调整的就是下面这些：
bootstrap.servers
: 这是最基础也最重要的参数，指定Kafka broker的地址列表。格式通常是
host1:port1,host2:port2
。如果只配置一个，当这个broker挂掉时，客户端就无法连接了。所以，我总是建议配置至少两个或更多的broker地址，即使是单机测试，也尽量模拟集群环境，这样客户端就能自动发现并连接到可用的broker。
client.id
: 客户端ID，一个任意字符串，用于在Kafka日志和监控中识别你的客户端。虽然不是强制的，但强烈建议设置一个有意义的ID，尤其是在复杂的系统里，这对于排查问题简直是救命稻草。比如
php-web-order-producer
或者
php-analytics-consumer
。
生产者相关参数：
acks
: 这个参数控制生产者发送消息的可靠性。
0
: 生产者发送后不等待任何确认。性能最好，但消息丢失风险最高。
1
: Leader副本收到消息后即确认。平衡了性能和可靠性。
all
(或
-1
): Leader副本收到消息，并等待所有ISR（In-Sync Replicas）副本都同步完成后才确认。可靠性最高，但性能最低。我通常会根据业务对消息丢失的容忍度来选择，对于核心业务，
all
是首选。
message.timeout.ms
: 消息发送的超时时间，超过这个时间还没收到确认，就会被认为是失败。
retries
/
message.send.max.retries
: 消息发送失败后的重试次数。配合
retry.backoff.ms
（重试间隔）一起使用，可以提高消息的发送成功率。但要注意，重试可能导致消息重复，下游消费者需要具备幂等性处理能力。
compression.codec
: 消息压缩算法，如
gzip
、
snappy
、
lz4
、
zstd
。选择合适的压缩算法可以在网络传输和存储上节省资源，但会增加CPU开销。
消费者相关参数：
group.id
: 消费者组ID。这是Kafka实现消息广播和负载均衡的关键。同一
group.id
下的消费者会共同消费一个Topic的不同分区，实现负载均衡；不同
group.id
下的消费者则会收到所有消息，实现消息广播。理解并正确使用
group.id
至关重要。
enable.auto.commit
: 是否自动提交offset。设置为
true
时，
rdkafka
会定期自动提交消费者已处理的offset。这很方便，但如果程序在提交前崩溃，可能导致消息重复消费。我个人更倾向于设置为
false
，然后手动控制offset的提交，这样可以更精确地控制消息处理的事务性。
auto.offset.reset
: 当消费者组第一次启动，或者之前提交的offset无效（比如数据已过期）时，如何处理。
earliest
: 从Topic的最早可用offset开始消费。
latest
: 从Topic的最新offset开始消费。
none
: 如果没有有效offset，直接抛出错误。这个参数的选择取决于你的业务需求，是希望重新处理所有历史消息，还是只关心新消息。
max.poll.interval.ms
: 消费者两次
consume()
调用之间的最大允许间隔。如果超过这个时间没有调用
consume()
，Kafka会认为这个消费者已经挂了，并触发Rebalance，将它的分区分配给组内其他消费者。对于长时间处理消息的场景，需要注意调整这个值。
这些参数的调整，往往需要结合实际的业务场景和对Kafka的理解。没有一劳永逸的配置，只有最适合你当前系统的配置。我建议在开发和测试阶段就多尝试不同的参数组合，观察其对系统行为的影响。
以上就是CentOS怎么配置Kafka PHP_CentOS搭建Kafka并配置PHP客户端教程的详细内容，更多请关注php中文网其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/585404.html
微信扫一扫 
支付宝扫一扫 
