Java-API+Kafka实现自定义分区
Posted 追码人
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Java-API+Kafka实现自定义分区相关的知识，希望对你有一定的参考价值。
目录章节：
1.pom.xml导入kafka依赖包；
2.kafka普通生产者实现方式；
3.kafka带回调函数的生产者；
4.生产者自定义分区；
4.1使用自定义分区
1.pom.xml导入kafka依赖包：
<!--kafka依赖-->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.11.0.0</version>
</dependency>
PS:kafkaProducer发送数据流程及ACK、重复消费与数据丢失问题：
1.Kafka 的 Producer 发送消息采用的是 异步发送的方式。在消息发送的过程中，涉及到了两个线程 ——main 线程和Sender线程，以及 一个线程共享变量 ——RecordAccumulator。main 线程将消息发送给 RecordAccumulator，Sender 线程不断从 RecordAccumulator 中拉取 消息发送到 Kafka broker。 2.异步和ack并不冲突，生产者一直发送数据，不等应答，如果某条数据迟迟没有应答，生产者会再发一次;
3.acks： -1 代表所有处于isr列表中的follower partition都会同步写入消息成功 0 代表消息只要发送出去就行，其他不管 1 代表发送消息到leader partition写入成功就可以;
4.重复消费与数据丢失:
说明： 已经消费的数据对于kafka来说，会将消费组里面的offset值进行修改，那什么时候进行修改了？是在数据消费 完成之后，比如在控制台打印完后自动提交；
提交过程：是通过kafka将offset进行移动到下个message所处的offset的位置。拿到数据后，存储到hbase中或者mysql中，如果hbase或者mysql在这个时候连接不上，就会抛出异常，如果在处理数据的时候已经进行了提交，
那么kafka上的offset值已经进行了修改了，但是hbase或者mysql中没有数据，这个时候就会出现数据丢失。什么时候提交offset值？在Consumer将数据处理完成之后，再来进行offset的修改提交。默认情况下offset是 自动提交，
需要修改为手动提交offset值。如果在处理代码中正常处理了，但是在提交offset请求的时候，没有连接到kafka或者出现了故障，那么该次修 改offset的请求是失败的，那么下次在进行读取同一个分区中的数据时，会从已经处理掉的offset值再进行处理一 次，
那么在hbase中或者mysql中就会产生两条一样的数据，也就是数据重复。
PS:数据来源:
/**
* 获取数据库数据
* @param
* @return
* @throws SQLException
*/
public static List<KafKaMyImage> getKafKaMyImages() throws SQLException
List<KafKaMyImage> kafKaMyImages=new ArrayList<>();
KafKaMyImage kafKaMyImage=null;
String sql="select id,loginip,updatetime,username,loginaddr from adminlogin";
Connection conection = SingleJavaJDBC.getConection();
PreparedStatement preparedStatement = conection.prepareStatement(sql);
ResultSet resultSet = preparedStatement.executeQuery();
while (resultSet.next())
kafKaMyImage=new KafKaMyImage(Integer.parseInt(resultSet.getString("id")),
resultSet.getString("loginip"),
resultSet.getString("updatetime"),
resultSet.getString("username"),
resultSet.getString("loginaddr"));
kafKaMyImages.add(kafKaMyImage);
// SingleJavaJDBC.close(resultSet,preparedStatement,conection);
return kafKaMyImages;
2.kafka普通生产者实现方式：
public void producerOne()
2 Properties props = new Properties();
3 // Kafka服务端的主机名和端口号
4 props.put("bootstrap.servers", "hadoop01:9092");
5 // 所有副本都必须应答后再发送
6 props.put("acks", "all");
7 // 发送失败后,再重复发送的次数
8 props.put("retries", 0);
9 // 一批消息处理大小
10 props.put("batch.size", 16384);
11 // 请求时间间隔
12 props.put("linger.ms", 1);
13 // 发送缓存区内存大小
14 props.put("buffer.memory", 33554432);
15 // key序列化
16 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
17 // value序列化
18 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
19 //2.定义kafka生产者
20 Producer<String, String> producer = new KafkaProducer<>(props);
21 //3.发送消息
22 for (int i = 0; i < 5; i++)
23 //top,指定分区，数据
24 //("second",0，key,"");指定分区
25 //("second",key,"");指定key，根据key分区
26 //("second","");不指定，随机分区，轮询
27 producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), Integer.toString(i)));
28
29 producer.close();
30
3.kafka带回调函数的生产者：
/**
* 创建生产者带回调函数02
* @throws SQLException
*/
public static void producerThree() throws SQLException
//step1 配置参数，这些跟优化kafka性能有关系
Properties props=new Properties();
// props.put("partitioner.class","com.comment.kafka.demo.producer.MyPartitioner");
//1 连接broker
props.put("bootstrap.servers","hadoop01:9092,hadoop02:9092,hadoop03:9092");
//2 key和value序列化
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//3 acks
// -1 代表所有处于isr列表中的follower partition都会同步写入消息成功
// 0 代表消息只要发送出去就行，其他不管
// 1 代表发送消息到leader partition写入成功就可以
props.put("acks","-1");
//4 重试次数
props.put("retries",3);//大部分问题，设置这个就可以解决，生产环境可以设置多些 5-10次
// 5 隔多久重试一次
props.put("retry.backoff.ms",2000);
//6 如果要提升kafka的吞吐量，可以指定压缩类型，如lz4
props.put("compression.type","none");
//7 缓冲区大小，默认是32M
props.put("buffer.size",33554432);
//8 一个批次batch的大小，默认是16k，需要根据一条消息的大小去调整
props.put("batch.size",323840);//设置为32k
//9 如果一个batch没满，达到如下的时间也会发送出去
props.put("linger.ms",200);
//10 一条消息最大的大小,默认是1M，生产环境中一般会修改变大，否则会报错
props.put("max.request.size",1048576);
//11 一条消息发送出去后，多久还没收到响应，就认为是超时
props.put("request.timeout.ms",5000);
//step2 创建生产者对象
KafkaProducer<String,String> producer=new KafkaProducer<String, String>(props);
//step3 使用消息的封装形式，注意value一般是json格式
List<KafKaMyImage> kafKaMyImages = getKafKaMyImages();
for (int i = 0; i < kafKaMyImages.size(); i++)
//step4 调用生产者对象的send方法发送消息，有异步和同步两种选择
//1 异步发送，一般使用异步，发送后会执行一个回调函数
//top,指定分区，数据
KafKaMyImage kafKaMyImage = kafKaMyImages.get(i);
JSONObject jsonObject = JSONObject.fromObject(kafKaMyImage);
producer.send(new ProducerRecord<String, String>("topicC","0",jsonObject.toString()), new Callback()
@Override
public void onCompletion(RecordMetadata metadata, Exception exception)
//判断是否有异常
if(exception==null)
System.out.println("消息发送到分区"+metadata.partition()+"成功");
else
System.out.println("消息发送失败");
// TODO 可以写入到redis，或mysql
);
try
Thread.sleep(10*1000);
catch (InterruptedException e)
e.printStackTrace();
//2 同步发送，需要等待一条消息发送完成，才能发送下一条消息
//RecordMetadata recordMetadata = producer.send(record).get();
//System.out.println("发送到的分区是："+recordMetadata.partition());
//step5 关闭连接
producer.close();
4.生产者自定义分区：
Kafka自定义分区需要实现Partitioner类，这里实现的是根据某个字段的值把数据写入相应分区
package com.comment.kafka.demo.producer;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import java.util.List;
import java.util.Map;
/**
* @className: MyPartitioner
* @description: TODO 类描述
* @author: 东林
* @date: 2022/2/26
**/
public class MyPartitioner implements Partitioner
/**
* 主要重写这个方法，假设有topic country三个分区，producer将key为china、usa和korea的消息分开存储到不同的分区，否则都放到0号分区
* @param topic 要使用自定义分区的topic
* @param key 消息key
* @param keyBytes 消息key序列化字节数组
* @param value 消息value
* @param valueBytes 消息value序列化字节数组
* @param cluster 集群元信息
* @return
*/
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster)
int partitions=0;
String keyStr=(String) key;
//获取分区信息
List<PartitionInfo> partitionInfoList=cluster.availablePartitionsForTopic(topic);
//获取当前topic的分区数
int partitionInfoListSize=partitionInfoList.size();
//判断是否有三个分区
if(partitionInfoListSize==3)
switch (Integer.parseInt(keyStr))
case 1:
partitions=0;
break;
case 0:
partitions=1;
break;
default:
partitions=2;
break;
//返回分区序号
return partitions;
@Override
public void close()
/**
* 文件加载时
* @param map
*/
@Override
public void configure(Map<String, ?> map)
4.1使用自定义分区
public static void producerPartition() throws SQLException
//step1 配置参数，这些跟优化kafka性能有关系
Properties props=new Properties();
//1 连接broker
props.put("bootstrap.servers","hadoop01:9092,hadoop02:9092,hadoop03:9092");
//2 key和value序列化
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//3 acks
// -1 代表所有处于isr列表中的follower partition都会同步写入消息成功
// 0 代表消息只要发送出去就行，其他不管
// 1 代表发送消息到leader partition写入成功就可以
props.put("acks","-1");
//4 重试次数
props.put("retries",3);//大部分问题，设置这个就可以解决，生产环境可以设置多些 5-10次
// 5 隔多久重试一次
props.put("retry.backoff.ms",2000);
//6 如果要提升kafka的吞吐量，可以指定压缩类型，如lz4
props.put("compression.type","none");
//7 缓冲区大小，默认是32M
props.put("buffer.size",33554432);
//8 一个批次batch的大小，默认是16k，需要根据一条消息的大小去调整
props.put("batch.size",323840);//设置为32k
//9 如果一个batch没满，达到如下的时间也会发送出去
props.put("linger.ms",200);
//10 一条消息最大的大小,默认是1M，生产环境中一般会修改变大，否则会报错
props.put("max.request.size",1048576);
//11 一条消息发送出去后，多久还没收到响应，就认为是超时
props.put("request.timeout.ms",5000);
//12 使用自定义分区器
props.put("partitioner.class","com.comment.kafka.demo.producer.MyPartitioner");
//step2 创建生产者对象
KafkaProducer<String,String> producer=new KafkaProducer<String, String>(props);
//step3 使用消息的封装形式，注意value一般是json格式
List<KafKaMyImage> kafKaMyImages = getKafKaMyImages();
for (int i = 0; i < kafKaMyImages.size(); i++)
//step4 调用生产者对象的send方法发送消息，有异步和同步两种选择
//1 异步发送，一般使用异步，发送后会执行一个回调函数
//top,指定分区，数据
KafKaMyImage kafKaMyImage = kafKaMyImages.get(i);
JSONObject jsonObject = JSONObject.fromObject(kafKaMyImage);
producer.send(new ProducerRecord<String, String>("topicD",kafKaMyImages.get(i).getIsdel(),jsonObject.toString()), new Callback()
@Override
public void onCompletion(RecordMetadata metadata, Exception exception)
//判断是否有异常
if(exception==null)
System.out.println("消息发送到分区"+metadata.partition()+"成功");
else
System.out.println("消息发送失败");
// TODO 可以写入到redis，或mysql
);
try
Thread.sleep(10*1000);
catch (InterruptedException e)
e.printStackTrace();
//2 同步发送，需要等待一条消息发送完成，才能发送下一条消息
//RecordMetadata recordMetadata = producer.send(record).get();
//System.out.println("发送到的分区是："+recordMetadata.partition());
producer.flush();
//step5 关闭连接
producer.close();
本文来自博客园，作者：zhuzhu&you，转载请注明原文链接：https://www.cnblogs.com/zhuzhu-you/p/15948155.html
以上是关于Java-API+Kafka实现自定义分区的主要内容，如果未能解决你的问题，请参考以下文章