php kafka 问题,php的kafka踩坑（二）_王高乒的博客-CSDN博客
php kafka 问题,php的kafka踩坑（二）
王高乒
于 2021-03-29 04:11:32 发布
546
收藏
文章标签：
php kafka 问题
接上一篇文章，上次没有解决的一个问题就是在做一个队列的时候，存在多消费者消费到同一个消息的情况，今天终于解决了这个问题，问题的本质是因为运维给我创建的topic是有问题的，他创建的分区数量是0，我今天上容器看了一下，终于发现了，然后删了自己重新建了一个，具体容器操作kafka的topic教程可以看我另一个文档基于kafka容器操作topic。
在这里，我们从头开始介绍一下topic(主题)，partition(分区)，group(分组)，consumer(消费者)，producer(生产者)的关系
producer，生产者，生产数据
consumer，消费者，消费数据
topic，简单点说就是一个队列，生产者生产数据和消费者消费数据都必须指定一个Topic，就是生产的数据要放到哪个队列去给消费者消费
partition和group，一个topic可以配置多个partition，consumer消费数据时是按照group来消费的，kafka确保每个partition只能由同一个group中的同一个consumer消费，如果想要重复消费，那么需要其他的组来消费，所以同一个group的消费者数量应当小于等于partition数量。Zookeerper中保存这每个topic下的每个partition在每个group中消费的offset。(此段介绍引用这篇文章)
consumer读取时，会指定读取的group，同一个消息在同一个group下只会读取到一次，如果要重复消费数据，需要新建group
如果group只有一个，并且有多个partition，一个consumer时，所有partition里的消息都会发往该consumer，如果consumer不止一个时，可能会存在有的consumer里面数据消费的多，有的消费的少，做多消费者的队列就是用这个特性，当partition数量=consuerm时，消息可以达到负载均衡。
下面上一下下代码，修改之后topic的partition是3个，依旧是基于 enqueue/rdkafka 这个包
生产者，不指定partition，kafka会自动分配
$connFactory = new RdKafkaConnectionFactory([
'global' => [
'metadata.broker.list' => '127.0.0.1:9092',
'socket.timeout.ms' => '50'
]);
$context = $connFactory->createContext();
$topic = $context->createQueue('app');
for ($i = 0; $i <= 5; $i++) {
$message = $context->createMessage('hello world!' . $i);
$context->createProducer()->send($topic, $message);
复制代码消费者
$config = [
'global' => [
'group.id' => date('Ymd'), // 指定一个分区，分区名自定义，做队列分区名必须一样
'metadata.broker.list' => '127.0.0.1:9092',
'enable.auto.commit' => 'false',
],
'topic' => [
'auto.offset.reset' => 'latest',
],
];
$connFactory = new RdKafkaConnectionFactory($config);
$context = $connFactory->createContext();
$queue = $context->createQueue('app');
$consumer = $context->createConsumer($queue);
while (true) {
$message = $consumer->receive(30 * 1000);
if (!is_object($message)) {
continue;
var_dump($message->getBody());
$consumer->acknowledge($message);
$consumer->reject($message);
复制代码
启动三个消费者，运行一次生产者，可以得到以下结果
消费者1
string(13) "hello world!1"
string(13) "hello world!3"
string(13) "hello world!4"
string(13) "hello world!5"
复制代码
消费者2
string(13) "hello world!2"
复制代码
消费者3
string(13) "hello world!0"
复制代码
到此就实现了kafka做队列的需求了，本文内容就到这里，相关kafka知识可以看这篇文章。
王高乒
关注
关注
点赞
收藏
评论
php kafka 问题,php的kafka踩坑（二）
接上一篇文章，上次没有解决的一个问题就是在做一个队列的时候，存在多消费者消费到同一个消息的情况，今天终于解决了这个问题，问题的本质是因为运维给我创建的topic是有问题的，他创建的分区数量是0，我今天上容器看了一下，终于发现了，然后删了自己重新建了一个，具体容器操作kafka的topic教程可以看我另一个文档基于kafka容器操作topic。在这里，我们从头开始介绍一下topic(主题)，part...
复制链接
扫一扫
【kafka】kafka 错误代码解释
九师兄
04-22
1228
1.概述
Kafka 原生API交互的Code代表，使用client 与 kafka 交互时，会返回错误代码，官方给出了，代码表示的具体含义
官网地址 https://kafka.apache.org/protocol
下表是Error code 以及相对应的关系
Error
Code
Retriable
Description
UNKNOWN_SERVER_ERROR-1FalseThe server experienced an unexpected error when processing t.
Kafkaphp使用纯粹的PHP编写的kafka客户端
08-07
Kafka-php 使用纯粹的 PHP 编写的 kafka 客户端
参与评论
您还未登录，请先
登录
后发表或查看评论
Kafka: 异常知其所以然
最新发布
dinl_vin的博客
10-09
497
kafka踩坑记录
php 生产kafka 不生效问题
钚该钚想
04-21
733
记录一下php 调用 kafka 生产者代码但是消费者端口收不到数据的情况
PHP代码如下：
<?php
$objRdKafka = new RdKafka\Producer();
$objRdKafka->addBrokers("localhost:9092");
$oObjTopic = $objRdKafka->newTopic("demo");
$oObjTopic-&g...
php rdkafka实例,php使用rdkafka进行消费
weixin_42400669的博客
03-09
1058
如仅作为消费者或生产者，直接使用下面消费者或生产者的代码，并安装扩展即可。PHP要安装rdkafka扩展，而rdkafka又依赖librdkafka，因此你需要安装rdkafka和librdkafka，之后就可以与kafka服务器交互了。消费者/*** 代码中的输出注释都可以打开供调试使用* 对 中台生产的 用户信息 进行消费* Date: 2019/7/31*/// 设置将要消费消息的主题$t...
php多进程消费kafka消息业务逻辑处理demo
lc的大脑备份
03-06
1513
这里提供的demo只是简单的打印了msg信息，大家可以按需调整。
class EchoKafkaConsumerLow extends KafkaConsumerLowService
function handleLogic($msg)
print_r($msg);
class EchoKafkaConsumerHigh extends...
php kafka storm,php的kafka踩坑（二）
weixin_33748493的博客
03-17
160
接上一篇文章，上次没有解决的一个问题就是在做一个队列的时候，存在多消费者消费到同一个消息的情况，今天终于解决了这个问题，问题的本质是因为运维给我创建的topic是有问题的，他创建的分区数量是0，我今天上容器看了一下，终于发现了，然后删了自己重新建了一个，具体容器操作kafka的topic教程可以看我另一个文档基于kafka容器操作topic。在这里，我们从头开始介绍一下topic(主题)，part...
kafka php 教程,php的kafka踩坑（一）
weixin_39599372的博客
03-10
446
最近项目上有一个需要用到消息队列的功能，从网上找了一些php相关的kafka使用的教程和博客，大抵都是安装php的拓展librdkafka(这里就不讲这个拓展的安装方法了，搜一下还是有很多教程的)，然后直接用这个拓展进行开发，但是我直接用这个拓展开发的时候，不知道为啥运行不起来，一直报错(应该是我太菜了，哈哈哈哈哈哈)......我从github上找了一些相关的包想直接用一下，但是发现很多包都是几...
kafka消费数据
laihechaba的博客
04-08
3348
kafka消费数据最简单的实现方法
from kafka import KafkaConsumer
def consumer_kafka():
kafkaServers = []
topic = ""
consumer = KafkaConsumer(topic, bootstrap_servers=kafkaServers,
auto_offset_reset='latest', #从最后开始消费（倒序）
在 php 中简单调用 kafka
潘广宇的博客
03-12
3362
一、确保 kafka 已被安装在 Linux 服务器中
若未安装，查看此博客快速安装：
https://blog.csdn.net/panguangyuu/article/details/88408320
二、启动 kakfa 服务
# 启动zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties &...
php的kafka踩坑（一）
m0_37654408的博客
03-29
1470
最近项目上有一个需要用到消息队列的功能，从网上找了一些php相关的kafka使用的教程和博客，大抵都是安装php的拓展librdkafka（这里就不讲这个拓展的安装方法了，搜一下还是有很多教程的），然后直接用这个拓展进行开发，但是我直接用这个拓展开发的时候，不知道为啥运行不起来，一直报错（应该是我太菜了，哈哈哈哈哈哈）…我从github上找了一些相关的包想直接用一下，但是发现很多包都是几年前的了，...
php kafka 消费代码
fyj的博客
12-14
1593
<?php
date_default_timezone_set('PRC');
//kfk 代理列表
$brokerList = 'xxx:9092,xxx:9092,xxx:9092';
$baiduAppIdLen = 2;
$conf = new \RdKafka\Conf;
#定义消费组
$conf->set('group.id', 'test-consumer-gr...
php多进程消费kafka消息低阶API封装
lc的大脑备份
03-06
736
php的kafka低阶API是指需要自行管理topic的partition信息，如果partition变动（增加或者减少），client侧需要自行处理负载均衡等，不然可能出现部分消息不处理的情形。
abstract class KafkaConsumerLowService extends KafkaConsumerBaseService
function consumer($pa...
kafka消息消费转发代理模式在php中的使用实践
MICweaver的技术blog
12-18
1280
目录整体架构消息发送到kafka集群消费者代理转发运维操作
整体架构
消息发送到kafka集群
消费者代理转发
运维操作
php调用kafka,php中使用kafka
weixin_39647734的博客
03-10
127
早想用消息队列了，但是一直没有动手今天搞起来安装kafka启动服务bin/zookeeper-server-start.sh -daemon config/zookeeper.propertiesbin/kafka-server-start.sh config/server.properties创建topicbin/kafka-topics.sh --create --zookeeper loca...
php kafka
u013713010的专栏
04-01
2913
Kafka 是一种高吞吐的分布式消息系统，能够替代传统的消息队列用于解耦合数据处理，缓存未处理消息等，同时具有更高的吞吐率，支持分区、多副本、冗余，因此被广泛用于大规模消息数据处理应用。
Kafka的特点：
以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间复杂度的访问性能。高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条以上消息
PHP下kafka的实践
搁浅在PHP沙滩上的博客
01-22
481
一、Kafka
1、简介
Kafka 是一种高吞吐量的分布式发布订阅消息系统
2、kafka角色必知
producer：生产者。
consumer：消费者。
topic: 消息以topic为类别记录,Kafka将消息种子(Feed)分类, 每一类的消息称之为一个主题(Topic)。
broker：以集群的方式运行,可以由一个或多个服务组成，每个服务叫做一个broker;消费者可以订阅一个或...
30个Kafka常见错误小集合
东城庞太师
08-06
5854
本文是一个Kafka使用过程中的常见错误的总结。希望对你有帮助。
1、UnknownTopicOrPartitionException
org.apache.kafka.common.errors.UnknownTopicOrPartitionException:
This server does not host this topic-partition
报错内容：分区数据不在
原因分析：producer向不存在的topic发送消息，用户可以检查topic是否存在 或者设置auto.creat
RabbitMQ+PHP 消息队列环境配置
weixin_33717298的博客
03-17
103
参考文档：http://www.cnblogs.com/phpinfo/p/4104551...http://blog.csdn.net/historyasamirror/ar...
依赖包安装
yum install ncurses-devel unixODBC unixODBC-devel
erlang环境
wget http:/...
php处理kafka消息
梁吉林的博客
06-04
3247
如果安装kafka，看上一篇文章CentOS7安装kafka
php如果要使用kafka的话，需要安装一下kafka-php：
# composer require nmred/kafka-php
kafka-php的GitHub地址
先写一个kafka_producer.php，用来做为生产者：
<?php
require '/var/www/extend/vendor/au...
“相关推荐”对你有帮助么？
非常没帮助
没帮助
一般
有帮助
非常有帮助
提交
©️2022 CSDN
皮肤主题：数字20
设计师：CSDN官方博客
返回首页
王高乒
CSDN认证博客专家
CSDN认证企业博客
码龄6年
暂无认证
78
原创
周排名
166万+
总排名
11万+
访问
等级
115
积分
粉丝
获赞
评论
68
收藏
私信
关注
热门文章
80psi等于多少kpa_kpa与psi的换算(kpa与psi对照表)
12935
idea创建java项目_使用IDEA创建java项目（hello word）
9620
linux删除目录非空,rmdir命令 – 删除空目录
7353
如何区分网线是几类的_网线怎么区分是几类的-怎么看网线是几类线
4653
985计算机专业考研难度排名,全国100所大学考研难易排名，985反而更好考，有意思...
4455
您愿意向朋友推荐“博客详情页”吗？
强烈不推荐
不推荐
一般般
推荐
强烈推荐
提交
最新文章
装文件服务器要装AD吗,Server 2008 R2之从介质安装AD DS
网吧主服务器虚拟内存,教你如何设置虚拟内存？
亚马逊服务器芯片查询,Amazon Lambda
2021年150篇
2020年14篇
目录
目录
最新文章
装文件服务器要装AD吗,Server 2008 R2之从介质安装AD DS
网吧主服务器虚拟内存,教你如何设置虚拟内存？
亚马逊服务器芯片查询,Amazon Lambda
2021年150篇
2020年14篇
目录
评论
被折叠的 条评论
为什么被折叠?
到【灌水乐园】发言
查看更多评论
实付元
使用余额支付
点击重新获取
扫码支付
钱包余额
抵扣说明：
1.余额是钱包充值的虚拟货币，按照1:1的比例进行支付金额的抵扣。 2.余额无法直接购买下载，可以购买VIP、C币套餐、付费专栏及课程。
余额充值