org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

序 本文主要展示一下如何使用reactor-kafka maven {代码...} 准备 启动zookeeper {代码...} 启动kafka {代码...} 创建topic {代码...} 实例 producer {代码...} consumer {代码...} 小结 reactor-kafka对kafka的api进行封装，改造为reactive streams模式，这样用起来更为顺手，熟悉reactor的开发人员可以轻车熟路。 doc ...

前面一篇文章讨论了ConsumerFetcherManager的MaxLag与ConsumerOffsetChecker的lag值的区别。但是关于MaxLag的值还没有讲的太透彻，这里再深入一下，如何让ConsumerFetcherManager的MaxLag有值。

使用jmx查询出来的MaxLag跟使用ConsumerOffsetChecker查出来的总是不一样，几乎是jmx查出来的是0，但是实际是存在lag的。这里探究一下这个MaxLag的计算。

序 本文主要讨论一下kafka consumer offset lag的监控 方案 利用官方的类库 ConsumerOffsetCheckerConsumerGroupCommand 利用官方的JMX ConsumerOffsetChecker 在0.8.2.2版本如下kafka_2.10-0.8.2.2-sources.jar!/kafka/tools/ConsumerOffsetChecker.scala {代码...} 缺点就是该类是给命令行调用的，每调用一次，就new一...

序 本文主要解析一下遇到的一个kafka consumer offset lag不断增大的异常。 查看consumer消费情况 {代码...} 发现消费者的offset与logSize差距太大，lag值都过10w了。 正常的情况 {代码...} 像这种lag差距比较少的，是正常的。 查看topic的partition {代码...} topic是4个分区，因此4个consumer来消费是正常的。问题可能...

本文主要研究一下kafka的partition分配，主要是key到parition的映射，partition对consumer的分配，以及partition的replica对broker/machine的分配。

A join operation merges two streams based on the keys of their data records, and yields a new stream. A join over record streams usually needs to be performed on a windowing basis because otherwise the number of records that must be maintained for performing the join may grow indefinitely.

kafka-streams-0.10.2.1-sources.jar!/org/apache/kafka/streams/kstream/KStreamBuilder.java

log4j-core-2.7-sources.jar!/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppender.java

kafka呢其实正道不是消息队列，本质是日志存储系统，而stream processing是其最近大力推广的特性，本文简单介绍下word count的实例。

本文主要讲述一下spring for kafka的consumer在spring.kafka.consumer.enable-auto-commit是false情况下，AckMode的选项

spring-kafka-1.2.3.RELEASE-sources.jar!/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

spring-kafka-1.2.3.RELEASE-sources.jar!/org/springframework/kafka/listener/adapter/AbstractRetryingMessageListenerAdapter.java主要有两个实现类RetryingAcknowledgingMessageListenerAdapter以及RetryingMessageListenerAdapter

kafka-clients-0.10.2.1-sources.jar!/org/apache/kafka/clients/consumer/KafkaConsumer.java

spring-kafka-1.2.3.RELEASE-sources.jar!/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java

spring-kafka-1.2.3.RELEASE-sources.jar!/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

KafkaAutoConfigurationspring-boot-autoconfigure-1.5.7.RELEASE-sources.jar!/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java

序 本文主要简单梳理梳理java应用中生产/消费kafka消息的一些使用选择。 可用类库 kafka client spring for apache kafka spring integration kafka spring cloud stream binder kafka 除了官方的java api类库外，spring生态中又额外包装了很多，这里一一简单介绍下。 spring for apache kafka 基于java版的kafka client...

可以看到跟0.8版本的不一样，不需要topicCountMap了This client transparently handles the failure of Kafka brokers, and transparently adapts as topic partitions it fetches migrate within the cluster.The consumer is not thread-safe.

序 本文简单介绍下docker运行kafka以及kafka manager 镜像 {代码...} 启动 {代码...} 也可以compose启动 {代码...} 访问 192.178.99.100:9000,然后添加cluster即可，其中zk地址填写zookeeper:2181 bug，明明docker是kafka0.9的，这个manager界面确只有0.8版本的可供选择 doc DockerKafka/kafka-manager-docker

本文主要来讲一个kafka的group coordinator。在kafka0.9.0版本的时候，开始启用了新的consumer config，这个新的consumer config采用bootstrap.servers替代之前版本的zookeeper.connect，主要是要渐渐弱化zk的依赖，把zk依赖隐藏到broker背后。

默认是true，也就是当memory buffer耗尽的时候阻塞等待；如果为false则抛出BufferExhaustedException异常

序 本文简单解析一下kafka0.8.2.2版本中的java producer的异常处理。 概况 kafka的java producer的发送是异步，主要的分几步： append到RecordAccumulator sender从RecordAccumulator取出RecordBatch，交给client去发送 NetworkClient跟broker打交道，把RecordBatch发送出去 这里就涉及到了几个步骤的异常，append的时候...

序 本文简单介绍下kafka0.8的client的producer的实例。 maven {代码...} 初始化 {代码...} 准备消息 {代码...} 发送 同步发送 {代码...} 异步发送(默认) {代码...} 使用callback {代码...} doc kafka08-documentation Kafka Clients (At-Most-Once, At-Least-Once, Exactly-Once, and Avro Client)

序 这里简单展示一下如何使用kafka0.8的client去消费一个topic。 maven {代码...} 初始化客户端 {代码...} 并发消费 {代码...} 注意事项 消费者实例数*每个实例的消费线程数 <= topic的partition数量，否则多余的就浪费了。

物理上把topic分成一个或多个partition，每个partition在物理上对应一个文件夹，该文件夹下存储 这个partition的所有消息和索引文件。

spring-cloud-stream-1.0.3.RELEASE-sources.jar!/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java

maven {代码...} 生产者配置 {代码...} java代码 {代码...} 消费者 {代码...} java代码 {代码...} 运行 先运行生产者，再运行消费者 {代码...} doc Apache Kafka Binder mac本地搭建kafka SpringCloudStream 构建消息驱动的微服务框架

简介 kafka0.10，消费topic的话，可以不用知道zk的地址，只要broker的地址即可 maven {代码...} 消费 {代码...} github kafka-consumer-demo