简易版 JAVA 开发

import org.apache.kafka.clients.producer.KafkaProducer;

import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**

* Created by 30869 on 2017/6/1.

*/

public class MyProducer {

//0.9版本过后，使用新的API KAFKA PRODUCER构建

private static KafkaProducer producer;

public static void main(String[] args) {

Properties properties=new Properties();

// properties.setProperty("metadata.broker.list","192.168.133.134:19092,192.168.133.133:19092," +

// "192.168.133.130:19092");

//必要参数 3个

//设置消息发送到broker 集群

properties.setProperty("bootstrap.servers","192.168.133.134:19092,192.168.133.133:19092," +

"192.168.133.130:19092");

//指定我们的消息序列化类

properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

//构建producer的时候需要指定producer发送的参数

producer=new KafkaProducer(properties);

//如果发送的主题不存在 ，会自动进行创建

producer.send(new ProducerRecord("changjinlu","Hello changjinglu"));

producer.close();

}

}

KAFKA 最终优化

1， 写负载均衡

2， 批量提交，使用producer.send(list)

3， 大批量提交过后 关闭producer，然后new Producer（）

4， 使用异步发送

KAFKA 并发读：

涉及到的概念，分区与CONSUMER的关系

理论上 ，一个分区使用一个CONSUMER线程进行处理

前提：

一个PRODUCER ， 消息并发写，然后使用的是负载均衡

例如： 分区数量为 3 ， producer 同时写三个分区

最终我们要实现： consumer并发的读取 三个分区的数据，并且 一个线程只能取一个分区的数据。

最终展现如下：

Message1 from partion 0 , thread 1

Message2 from partion 1 , thread 2

Message3 from partion 2 , thread 3

新版本特性:

KafkaProducer 的发送消息都是异步方式发送

发送端可配置参数及其含义

Batch.size 针对partition级别的缓存，当发送到partition上的消息到达一定量过后，才进行发送。

linger.ms 针对PRODUCER级别，如果没有到达batch.size ，那么就会以linger.ms为准。

例如： batch.size 设置为 16384

Linger.ms 设置为 3000

消息发送过后（1S的延时）：没有到达batch.size， 首先触发 linger.ms

buffer.memory 针对PRODUCER级别的缓存，当所有partition的缓存加起来超过producer缓存时，触发 消息的批量发送。

Acks 配置选项 ：0 1 all

分别代表 0 没有ack确认消息

1 发送一条确认，LEAD写入成功即成功，其他FOLLOWER不管

All 发送所有确认，所有集群机器确认

Retries 消息发送重试 ，如果发送失败会重新发送，会出现 消息重复发送

enable.auto.commit 可选项： ture false

当为true的时候 kafka 自动提交我们的消费纪录

为false的时候，关闭自动提交功能

如果关闭会出现： 消息虽然读取了，但是 再开consumer又会重新读取一次，因为当为false的时候，消费的offset没有提交到 kafka。

解决上面的情况：

通过 consumer.commitSync() 进行手动提交当前消费的offset

底层存储多了一个 timeindex

主要保存两个内容，一个 producer 发送这条消息的时候，一个是 broker写入消息成功的时间

主要用于消息的合并，还有kStream的实时消费。

SEEK 跳转到固定的OFFSET进行消费？

要实现 from beginning

1， 去掉 consumer.subscribe() ，添加consumer.assign()

两者 二选其一

2， 因为consumer.seek（）需要传入 topicPartition

所以呢需要自己定义 topicPartition(topic,parition)

3，设置consumer.seekToBeginning(parts);

例如代码：

TopicPartition topicPartition=new TopicPartition(topic,0);

// TopicPartition topicPartition1=new TopicPartition(topic,1);

// TopicPartition topicPartition2=new TopicPartition(topic,2);

List<TopicPartition> parts=new ArrayList<TopicPartition>();

parts.add(topicPartition);

// parts.add(topicPartition1);

// parts.add(topicPartition2);

consumer.assign(parts);

// consumer.subscribe(topis);

consumer.seekToBeginning(parts);

需要定位到 指定的 offset 偏移量

使用函数： consumer.seek(topPartition,offset)

针对单分区的偏移量记录

KAFKA SHELL 高阶

[hadoop@master bin]$ ./kafka-consumer-offset-checker.sh --zookeeper master:12181 --topic pengyong --group guoxu

#PS ： 以上命令，通过 kafka工具类定位 guoxu 这一个group组消费到了哪一个offset

OFFSET 是针对PARTITION级别的。

[2017-06-01 06:01:43,203] WARN WARNING: ConsumerOffsetChecker is deprecated and will be dropped in releases following 0.9.0. Use ConsumerGroupCommand instead. (kafka.tools.ConsumerOffsetChecker$)

Group Topic Pid Offset logSize Lag

guoxu pengyong 0 502595 502595 0

guoxu pengyong 1 567011 567011 0

guoxu pengyong 2 540550 540550 0

group 代表 consumer group的名称

topic 主题ID

PID 代表 topic的分区

Offset 当前分区，当前group组，消费topic到那一条消息

LAG 还剩下多少没读

LOGSIZE 这一个PARTITION总共有多少条消息

最终确定 ：

KAFKA纪录消费者的 四种属性 为

Group Topic Pid Offset

@来自科多大数据