文章目录
1、创建maven工程并添加jar包
创建maven工程并添加以下依赖jar包的坐标到pom.xml
<dependencies>
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
<version>2.0.0</version>
</dependency>
</dependencies>
<build>
<plugins>
<!-- java编译插件 -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.2</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
<encoding>UTF-8</encoding>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<id>maven-ali</id>
<url>http://maven.aliyun.com/nexus/content/groups/public//</url>
<releases>
<enabled>true</enabled>
</releases>
<snapshots>
<enabled>true</enabled>
<updatePolicy>always</updatePolicy>
<checksumPolicy>fail</checksumPolicy>
</snapshots>
</repository>
</repositories>
2、生产者代码
1、使用生产者，生产数据
/*
用于生产数据到kafka集群
*/
public class Producer1 {
/*
程序的入口
*/
public static void main(String[] args){
//编写生产数据的程序
//1、配置kafka集群环境（设置）
Properties props = new Properties();
//kafka服务器地址
props.put("bootstrap.servers", "node01:9092,node02:9092,node03:9092");
//消息确认机制
props.put("acks", "all");
//重试机制
props.put("retries", 0);
//批量发送的大小
props.put("batch.size", 16384);
//消息延迟
props.put("linger.ms", 1);
//批量的缓冲区大小
props.put("buffer.memory", 33554432);
// kafka key 和value的序列化
props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
//2、实例一个生产者对象
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(props);
for (int i = 0; i < 9; i++) {
//3、发送数据 ,需要一个producerRecord对象,最少参数 String topic, V value
//ProducerRecord<K, V> record
ProducerRecord producerRecord = new ProducerRecord<>("18BD12","bbbb___"+i);
//4、通过生产者对象将数据发送到kafka集群
kafkaProducer.send(producerRecord);
# //3、发送数据
# kafkaProducer.send(new ProducerRecord<String, String>("testpart","0","value"+i));
}
//4、关闭成产者
kafkaProducer.close();
}
}
2、kafka当中的数据分区
kafka生产者发送的消息，都是保存在broker当中，
我们可以自定义分区规则，决定消息发送到哪个partition里面去进行保存
查看
ProducerRecord
这个类的源码，就可以看到kafka的各种不同分区策略
kafka当中支持以下四种数据的分区方式：
//第一种分区策略，如果既没有指定分区号，也没有指定数据key，那么就会使用轮询的方式将数据均匀的发送到不同的分区里面去
//ProducerRecord<String, String> producerRecord1 = new ProducerRecord<>("mypartition", "mymessage" + i);
//kafkaProducer.send(producerRecord1);
//第二种分区策略 如果没有指定分区号，指定了数据key，通过key.hashCode % numPartitions来计算数据究竟会保存在哪一个分区里面
//注意：如果数据key，没有变化 key.hashCode % numPartitions = 固定值 所有的数据都会写入到某一个分区里面去
//ProducerRecord<String, String> producerRecord2 = new ProducerRecord<>("mypartition", "mykey", "mymessage" + i);
//kafkaProducer.send(producerRecord2);
//第三种分区策略：如果指定了分区号，那么就会将数据直接写入到对应的分区里面去
// ProducerRecord<String, String> producerRecord3 = new ProducerRecord<>("mypartition", 0, "mykey", "mymessage" + i);
// kafkaProducer.send(producerRecord3);
//第四种分区策略：自定义分区策略。如果不自定义分区规则，那么会将数据使用轮询的方式均匀的发送到各个分区里面去
kafkaProducer.send(new ProducerRecord<String, String>("mypartition","mymessage"+i));
//1、没有指定分区编号，没有指定key,时采用轮询方式存户数据
ProducerRecord producerRecord = new ProducerRecord<>("18BD12","bbbb___"+i);
//2、没有指定分区编号，指定key时，数据分发策略为对key求取hash值，这个值与分区数量取余，于数就是分区编号。
//ProducerRecord producerRecord = new ProducerRecord("18BD12","test","aaaa___"+i);
//3、指定分区编号，所有数据输入到指定的分区内
//ProducerRecord producerRecord = new ProducerRecord("18BD12",1,"test","aaaa___"+i);
//4、自定义分区策略。如果不自定义分区规则，那么会将数据使用轮询的方式均匀的发送到各个分区里面去
//ProducerRecord producerRecord = new ProducerRecord("18BD12","test","aaaa___"+i);
其中，自定义分区策略需要我们单独创建一个类，并在类中定义我们所想要的分区规则。
public class KafkaCustomPartitioner implements Partitioner {
@Override
public void configure(Map<String, ?> configs) {
}
@Override
public int partition(String topic, Object arg1, byte[] keyBytes, Object arg3, byte[] arg4, Cluster cluster) {
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
int partitionNum = partitions.size();
Random random = new Random();
int partition = random.nextInt(partitionNum);
return partition;
}
@Override
public void close() {
}
}
并在主代码中添加配置,其中partitioner.class的值对应的就是我们单独写的一个实现Partitioner 的类在项目中具体带包名的路径
props.put("partitioner.class", "com.czxy.demo_test.Demo05.KafkaCustomPartitioner ");
我们也可以通过IDEA中的快捷键来实现快速获取
3、消费者代码
消费者要从kafka Cluster进行消费数据，必要条件有以下四个
#1、地址
bootstrap.servers=node01:9092
#2、序列化
key.serializer=org.apache.kafka.common.serialization.StringSerializer value.serializer=org.apache.kafka.common.serialization.StringSerializer
#3、主题（topic） 需要制定具体的某个topic（order）即可。
#4、消费者组 group.id=test
1、自动提交offset
消费完成之后，自动提交offset
public class Consumer01 {
public static void main(String[] args) {
//1、添加配置文件
Properties props = new Properties();
//指定kafka服务器
props.put("bootstrap.servers", "node01:9092,node02:9092,node03:9092");
//消费组
props.put("group.id", "test");
//以下两行代码 ---消费者自动提交offset值
props.put("enable.auto.commit", "true");
//自动提交的周期
props.put("auto.commit.interval.ms", "1000");
//kafka key 和value的反序列化
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//2、实例消费者对象
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props);
//3、设置读取的topic
kafkaConsumer.subscribe(Arrays.asList("student"));
//循环遍历
while (true){
//4、拉取数据，并输出
//获取到所有的数据
ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(1000);
//遍历所有数据，获得到一条
for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
//一条数据
System.out.println("当前数据："+consumerRecord.value()+", 偏移量:offset:"+consumerRecord.offset());
}
}
}
}
2、手动提交offset
如果Consumer在获取数据后，需要加入处理，数据完毕后才确认offset，需要程序来控制offset的确认？
答：关闭自动提交确认选项
props.put("enable.auto.commit", "false");
然后在循环遍历消费的过程中，消费完毕就手动提交。
while (true){
//4、拉取数据，并输出
ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(1000);
//遍历所有数据获取一条
for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
System.out.println(consumerRecord.value() +" "+consumerRecord.offset());
}
//手动提交offset
kafkaConsumer.commitSync();
}
# 或者也可以将手动提交offset的语句放置到循环体中，每消费一条数据，就手动提交一次offset也是可以的。
3、消费完每个分区之后手动提交offset
上面的示例使用commitSync将所有已接收的记录标记为已提交。 在某些情况下，您可能希望通过明确指定偏移量 来更好地控制已提交的记录。 在下面的示例中，我们在完成处理每个分区中的记录后提交偏移量。
try {
while(running) {
ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
for (TopicPartition partition : records.partitions()) {
List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
for (ConsumerRecord<String, String> record : partitionRecords) { System.out.println(record.offset() + ": " + record.value());
}
long lastOffset = partitionRecords.get(partitionRecords.size() -1).offset();
consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
}
}
} finally {
consumer.close();
}
-
注意事项：
提交的偏移量应始终是应用程序将读取的下一条消息的偏移量。 因此，在调用
commitSync
（偏移量）时，应该 在最后处理的消息的偏移量中添加一个
4、指定分区数据进行消费
1、如果进程正在维护与该分区关联的某种本地状态（如本地磁盘上的键值存储），那么它应该只获取它在磁盘上 维护的分区的记录。
2、如果进程本身具有高可用性，并且如果失败则将重新启动（可能使用YARN，Mesos或AWS工具等集群管理框 架，或作为流处理框架的一部分）。 在这种情况下，Kafka不需要检测故障并重新分配分区，因为消耗过程将在另 一台机器上重新启动。
// 第一个参数为消费的Topic,第二个参数为消费的Partition
TopicPartition topicPartition0 = new TopicPartition("18BD12",0);
TopicPartition topicPartition2 = new TopicPartition("18BD12",1);
kafkaConsumer.assign(Arrays.asList(topicPartition0,topicPartition2));
注意事项：
- 1、要使用此模式，您只需使用要使用的分区的完整列表调用assign（Collection），而不是使用subscribe订阅 主题。
-
2、
主题与分区订阅只能二选一
5、重复消费与数据丢失
说明：
- 1、已经消费的数据对于kafka来说，会将消费组里面的offset值进行修改，那什么时候进行修改了？是在数据消费 完成之后，比如在控制台打印完后自动提交；
- 2、提交过程：是通过kafka将offset进行移动到下个message所处的offset的位置。
- 3、拿到数据后，存储到hbase中或者mysql中，如果hbase或者mysql在这个时候连接不上，就会抛出异常，如果在处理数据的时候已经进行了提交，那么kafka伤的offset值已经进行了修改了，但是hbase或者mysql中没有数据，这个时候就会出现数据丢失。
-
4、什么时候提交offset值？
在Consumer将数据处理完成之后，再来进行offset的修改提交
。默认情况下offset是 自动提交，需要修改为手动提交offset值。 -
5、如果在处理代码中正常处理了，但是在提交offset请求的时候，没有连接到kafka或者出现了故障，那么该次修 改offset的请求是失败的，那么下次在进行读取同一个分区中的数据时，会从已经处理掉的offset值再进行处理一 次，那么在hbase中或者mysql中就会产生两条一样的数据，也就是
数据重复
6、consumer消费者消费数据流程
流程描述
-
Consumer连接指定的Topic partition所在leader broker，采用pull方式从kafkalogs中获取消息。对于不同的消费模式，会将offset保存在不同的地方
官网关于high level API 以及low level API的简介
http://kafka.apache.org/0100/documentation.html#impl_consumer
高阶API（High Level API）
- kafka消费者高阶API简单；隐藏Consumer与Broker细节；相关信息保存在zookeeper中。
/* create a connection to the cluster */
ConsumerConnector connector = Consumer.create(consumerConfig);
interface ConsumerConnector {
/**
This method is used to get a list of KafkaStreams, which are iterators over
MessageAndMetadata objects from which you can obtain messages and their
associated metadata (currently only topic).
Input: a map of <topic, #streams>
Output: a map of <topic, list of message streams>
*/
public Map<String,List<KafkaStream>> createMessageStreams(Map<String,Int> topicCountMap);
/**
You can also obtain a list of KafkaStreams, that iterate over messages
from topics that match a TopicFilter. (A TopicFilter encapsulates a
whitelist or a blacklist which is a standard Java regex.)
*/
public List<KafkaStream> createMessageStreamsByFilter( TopicFilter topicFilter, int numStreams);
/* Commit the offsets of all messages consumed so far. */ public commitOffsets()
/* Shut down the connector */ public shutdown()
}
#说明：大部分的操作都已经封装好了，比如：当前消费到哪个位置下了，但是不够灵活（工作过程推荐使用）
低级API(Low Level API)
- kafka消费者低级API非常灵活；需要自己负责维护连接Controller Broker。保存offset，Consumer Partition对应 关系。
class SimpleConsumer {
/* Send fetch request to a broker and get back a set of messages. */
public ByteBufferMessageSet fetch(FetchRequest request);
/* Send a list of fetch requests to a broker and get back a response set. */
public MultiFetchResponse multifetch(List<FetchRequest> fetches);
/**
Get a list of valid offsets (up to maxSize) before the given time.
The result is a list of offsets, in descending order.
@param time: time in millisecs,
if set to OffsetRequest$.MODULE$.LATEST_TIME(), get from the latest
offset available. if set to OffsetRequest$.MODULE$.EARLIEST_TIME(), get from the earliest
available. public long[] getOffsetsBefore(String topic, int partition, long time, int maxNumOffsets);
* offset
*/
#说明：没有进行包装，所有的操作有用户决定，如自己的保存某一个分区下的记录，你当前消费到哪个位置。
4、kafka Streams API开发
需求：使用StreamAPI获取test这个topic当中的数据，然后将数据全部转为大写，写入到test2这个topic当中去
第一步：创建一个topic
node01服务器使用以下命令来常见一个topic 名称为test2
cd /export/servers/kafka_2.11-1.0.0/
bin/kafka-topics.sh --create --partitions 3 --replication-factor 2 --topic test2 --zookeeper node01:2181,node02:2181,node03:2181
第二步：开发StreamAPI
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import java.util.Properties;
public class Stream {
public static void main(String[] args) {
Properties props = new Properties();
//设置程序的唯一标识
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application");
//设置kafka集群
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "node01:9092");
//设置序列化与反序列化
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
//实例一个计算逻辑
StreamsBuilder streamsBuilder = new StreamsBuilder();
//设置计算逻辑 stream 在哪里读取数据 -> to 将数据写入哪里
streamsBuilder.stream("18BD12").mapValues(line->line.toString().toUpperCase()).to("18BD12-1");
//构建Topology对象（拓扑，流程）
final Topology topology = streamsBuilder.build();
//实例 kafka流
KafkaStreams streams = new KafkaStreams(topology, props);
//启动流计算
streams.start();
}
}
第三步：生产数据
node01执行以下命令，向test这个topic当中生产数据
cd /export/servers/kafka_2.11-1.0.0
bin/kafka-console-producer.sh --broker-list node01:9092,node02:9092,node03:9092 --topic test
第四步：消费数据
node02执行一下命令消费test2这个topic当中的数据
cd /export/servers/kafka_2.11-1.0.0
bin/kafka-console-consumer.sh --from-beginning --topic test2 --zookeeper node01:2181,node02:2181,node03:2181
版权声明：本文为qq_44509920原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。