我们上一章介绍了中间件：Zookeeper，本章将介绍另外一个中间件：Kafka。目前这2个中间件都是基于JAVA语言的。
本篇文章大概是我3年前写的，当时写这篇文章的背景是一个容器化部署的Kafka集群，其中一个节点宕机以后无法恢复，所以想了解下这个Kafak集群崩溃以后的恢复过程（下面的内容参考原文做了部分修改和脱敏）。
分别在hadoop103和hadoop104主机上修改配置文件server.properties中broker.id=103、broker.id=104
https://www.isharkfly.com/t/java-kafak/14999
作者介绍：简历上没有一个精通的运维工程师，下面的思维导图也是预计更新的内容和当前进度(不定时更新)。
最近有学习些Kafak的源码，想给大家分享下Kafak中改进的二分查找算法。二分查找，是每个程序员都应掌握的基础算法，而Kafka是如何改进二分查找来应用于自己的场景中，这很值得我们了解学习。...由于Kafak把二分查找应用于索引查找的场景中，所以本文会先对Kafka的日志结构和索引进行简单的介绍。...在Kafak中，消息以日志的形式保存，每个日志其实就是一个文件夹，且存有多个日志段，一个日志段指的是文件名（起始偏移）相同的消息日志文件和4个索引文件，如下图所示。 ?...之前在只有12号页的时候，Kafak读取索引时会频繁访问6、9、11、12号页，而由于Kafka使用了mmap来提高速度，即读写操作都将通过操作系统的page cache，所以6、9、11、12号页会被缓存到
概念 Producer 消息的生产者 Consumer 消息的消费者 ConsumerGroup 消费者组，实现单播和广播的手段 Broker kafak服务集群节点，Kafka集群中的一台或多台服务器统称...Producers 消息和数据生成者，向Kafka的一个topic发布消息的 过程叫做producers Consumers 消息和数据的消费者，订阅topic并处理其发布的消费过程叫做consumers kafak...routing Priority Queue A standard protocol message queue 生产者： 生产者生产消息不仅必须指定Topic，还可按照需求指定发往特定的分区 消费者： Kafak...Kafak争抢模式实现 多个消费者，同一个Topic同一个Group Kafak广播模式实现 多个消费者，同一个topic,不同Group C# 生产者 public static async Task...SessionTimeoutMs = 1000 * 60 * 30, //30min //MaxPollIntervalMs = 1000 * 60 * 30, //30min,30分钟不轮询kafak
node02) ❞ 架构图： 参数parseAsFlumeAgent设置为false 上游：数据通过node01的Kafka Channel存储到node02的Kafka主题（只有body数据）中，再从Kafak...主题中读取数据 下游：拦截器处理，利用Kafka Channel将数据从Kafak主题中读取出来， 此方案数仓采集过程一共3个链路（数据传输环节） 如下图： ❝与方案二相比，该方案节省一个Sink，节省一个数据传输环节
streaming是基于kafka-0.8.2编译的，虽然官网建议kafka-0.8及其以上，但kafka-0.9在更新zk的offset的api，完全不兼容kafka-0.8的api,所以说用高版本的kafak
"SECONDS" val KAFKA_TOPIC = "topic" val KAFKA_DEFAULT_TOPIC = "kafka-sink-topic" val KAFAK_BROKERS...= "kafka-brokers" val KAFAK_DEFAULT_BROKERS = "XXX:9092" val TOPIC = Option(property.getProperty...(KAFKA_TOPIC)).getOrElse(KAFKA_DEFAULT_TOPIC) val BROKERS = Option(property.getProperty(KAFAK_BROKERS
日志存储结构 首先我们来看一张 kafak 的存储结构图。 ?...zookeeper 127.0.0.1:2181 --topic test --partitions 12 复制代码 分区内每条消息都会被分配一个唯一的消息 id,也就是我们通常所说的 offset, 因此 kafak...kafak 中的 Log 对应了一个命名为- 的文件夹。...日志清理机制 由于 kafak 是把消息存储 在磁盘上，为了控制消息的不断增加我们就必须对消息做一定的清理和压缩。kakfa 中的每一个分区副本都对应的一个 log 日志文件。
日志采集使用KF，即kfka+flume，有程序直接输出到kafka，flume拉取存储到文件或 elasticsearch中，但是发现第一次启动程序时，能够正确输入到kafak，但是第二次启动时却直接打印到控制台
1、下载kafak包并解压 cd /usr/local wget https://dlcdn.apache.org/kafka/3.9.0/kafka_2.12-3.9.0.tgz tar -xzf kafka...dataDir修改为以下 dataDir=/usr/local/kafka/zookeeperDataDir 添加以下： tickTime=2000 initLimit=10 syncLimit=5 3、修改kafak
我们会以 Flink 最新版本1.10为基础，会讲到如下内容： 实战篇： Flink 消费 Kafak的基本方式，比如如何从指定 offset 处消费记录 Flink 消费 Kafka
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning 配置flume 创建配置文件kafak.conf
下载kafka https://mirror-hk.koddos.net/apache/kafka/2.8.0/kafka_2.12-2.8.0.tgz 启动kafak Linux .
kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"可知, Kafka启动时的入口类为kafka.Kafka, 我们直接来看这个类; ---- Kafka启动入口类:kafk.Kafak
--引入kafak和spring整合的jar--> org.springframework.kafka</groupId
格式的日志, 和 path 对齐 json.keys_under_root: true json.overwrite_keys: true ###输出到logstash,可以直接输出到es、kafak
云服务器
ICP备案
即时通信 IM
云直播
对象存储
