producer发送消息的分区策略
1.默认策略 DefaultPartitioner
发送消息时指定分区，优先级最高
发送消息时指定key，会根据key值hash取模
发送消息时没有指定分区和key。会随机发送到任意分区
2.自定义分区策略，需实现Partitioner接口的partition方法
producer参数配置
bootstrap-servers：broker地址，ip:port # 消息重发的次数 retries: 1 # 提交后等待返回的超时时间上线 和retries作用一致，控制生产者等待响应时间，两个参数可以只配置一个 默认2分钟 可不配置 delivery.timeout.ms: 30000 #一个批次可以使用的内存大小 默认16kb，无需配置 表示生产者生产消息先放缓存，达到内存或者时间限制后，批量发送到broker，与linger.ms搭配使用 batch-size: 16384 #如果想减少请求的数量，可以设置 linger.ms 大于0，即消息在缓冲区保留的时间，超过设置的值就会被提交到服务端，默认值是0 linger.ms：5 # 设置生产者内存缓冲区的大小。Kafka Producer能够使用的内存缓冲的大小的，默认值32MB，无需修改 buffer-memory: 33554432 # 键的序列化方式 key-serializer: org.apache.kafka.common.serialization.StringSerializer # 值的序列化方式 value-serializer: org.apache.kafka.common.serialization.StringSerializer #0 1 all；0表示生产者不等待任何确认，直接响应。1表示leader将数据落到磁盘，不等follower做出确认，直接响应。all表示等所有副本落到磁盘，确认才会响应。kafka的isr机制保证不会由于副本同步耗时过长导致生产者一直等待 acks: all
peoducer生产消息代码开发
public class KafkaProducerTest {
public static final String TOPIC_NAME = "default_topic";
public static Properties getProperties(){
Properties props = new Properties();
//填上自己的ip和port
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "ip:port");
// 当producer向leader发送数据时，可以通过request.required.acks参数来设置数据可靠性的级别,分别是0, 1，all。
props.put(ProducerConfig.ACKS_CONFIG, "all");
// 请求失败，生产者会自动重试，指定是0次，如果启用重试，则会有重复消息的可能性
props.put(ProducerConfig.RETRIES_CONFIG, 0);
// 生产者缓存每个分区未发送的消息,缓存的大小是通过 batch.size 配置指定的，默认值是16KB
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
/**
* 默认值就是0，消息是立刻发送的，即便batch.size缓冲空间还没有满
* 如果想减少请求的数量，可以设置 linger.ms 大于0，即消息在缓冲区保留的时间，超过设置的值就会被提交到服务端
* 通俗解释是，本该早就发出去的消息被迫至少等待了linger.ms时间，相对于这时间内积累了更多消息，批量发送减少请求
* 如果batch被填满或者linger.ms达到上限，满足其中一个就会被发送
*/
props.put("linger.ms", 5);
/**
* buffer.memory的用来约束Kafka Producer能够使用的内存缓冲的大小的，默认值32MB。
* 如果buffer.memory设置的太小，可能导致消息快速的写入内存缓冲里，但Sender线程来不及把消息发送到Kafka服务器
* 会造成内存缓冲很快就被写满，而一旦被写满，就会阻塞用户线程，不让继续往Kafka写消息了
* buffer.memory要大于batch.size，否则会报申请内存不#足的错误，不要超过物理内存，根据实际情况调整
* 需要结合实际业务情况压测进行配置
*/
props.put("buffer.memory", 33554432);
/**
* key的序列化器，将用户提供的 key和value对象ProducerRecord 进行序列化处理，key.serializer必须被设置，
* 即使消息中没有指定key，序列化器必须是一个实
org.apache.kafka.common.serialization.Serializer接口的类，
* 将key序列化成字节数组。
*/
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
return props;
}
/**
* send()方法是异步的，添加消息到缓冲区等待发送，并立即返回
* 生产者将单个的消息批量在一起发送来提高效率,即 batch.size和linger.ms结合
*
* 实现同步发送：一条消息发送之后，会阻塞当前线程，直至返回 ack
* 发送消息后返回的一个 Future 对象，调用get即可
*
* 消息发送主要是两个线程：一个是Main用户主线程，一个是Sender线程
* 1)main线程发送消息到RecordAccumulator即返回
* 2)sender线程从RecordAccumulator拉取信息发送到broker
* 3) batch.size和linger.ms两个参数可以影响 sender 线程发送次数
*/
@Test
public void testSend(){
Properties properties = getProperties();
Producer<String,String> producer = new KafkaProducer<>(properties);
for(int i=0;i<23 ;i++){
Future<RecordMetadata> future = producer.send(new ProducerRecord<>(TOPIC_NAME,"test-value"));
try {
//不关心结果则不用写这些内容
RecordMetadata recordMetadata = future.get();
// topic - 分区编号@offset
System.out.println("发送结果："+recordMetadata.toString());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
producer.close();
}
/**
* 发送消息携带回调函数
*/
@Test
public void testSendWithCallback(){
Properties properties = getProperties();
Producer<String,String> producer = new KafkaProducer<>(properties);
for(int i=0;i<3 ;i++) {
producer.send(new ProducerRecord<>(TOPIC_NAME, "test-key" + i, "test-value" + i), new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if(exception == null){
System.err.println("发送结果:"+metadata.toString());
} else {
//发送失败
exception.printStackTrace();
}
}
});
}
producer.close();
}
/**
* 发送消息携带回调函数，指定某个分区
* 实现顺序消息
*/
@Test
public void testSendWithCallbackAndPartition(){
Properties properties = getProperties();
Producer<String,String> producer = new KafkaProducer<>(properties);
for(int i=0;i<10 ;i++) {
//将消息指定发送到下标为4的partition
producer.send(new ProducerRecord<>(TOPIC_NAME, 4,"test-key" + i, "test-value" + i), new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if(exception == null){
//执行业务逻辑
System.err.println("发送状态:"+metadata.toString());
} else {
exception.printStackTrace();
}
}
});
}
producer.close();
}
/**
*
* 自定义分区策略
*/
@Test
public void testSendWithPartitionStrategy(){
Properties properties = getProperties();
properties.put("partitioner.class", "net.xdclass.xdclasskafka.config.CustomPartitioner");
Producer<String,String> producer = new KafkaProducer<>(properties);
for(int i=0;i<10 ;i++) {
producer.send(new ProducerRecord<>(TOPIC_NAME, "custom-test-key", "test-value" + i), new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if(exception == null){
System.err.println("发送状态:"+metadata.toString());
} else {
exception.printStackTrace();
}
}
});
}
producer.close();
}
}自定义分区策略：如果key为空，则报错，如果key为指定值，消息发送到指定partition,否则按key值hash取模
public class CustomPartitioner implements Partitioner {
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
if (keyBytes == null) {
throw new IllegalArgumentException("key 参数不能为空");
}
if("custom-test-key".equals(key)){
//指定下标为0的partition
return 0;
}
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
int numPartitions = partitions.size();
// hash the keyBytes to choose a partition
return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}
@Override
public void close() {
}
@Override
public void configure(Map<String, ?> configs) {
}
}consumer参数配置
bootstrap-servers: broker地址，ip:port #组id group.id: test-1 # offset的值， earliest从最早位置开始消费 latest从最近保存的位置开始消费 auto.offset.reset: latest # 心跳时间，超过这个时间将消费者踢出 默认10秒 session.timeout.ms: 10000 # 是否定时向ZooKeeper发送已经获取到的消息的offset 默认为true 改为false手动提交 enable.auto.commit: false # Acknowledgment.acknowledge()方法被监听器调用时立即提交偏移量 手动提交时设置为此值 ack.mode: MANUAL_IMMEDIATE # 消费线程数量 如果设置为1的情况下, 这一个实例消费Topic的所有分区;如果设置多个,那么会平均 分配所有分区; 如果是分布式，则实例数=机器数*concurrency concurrency: 1 # 批量消费 batch.listener: true # 批量消费最大条数 max.poll.records: 10
comsumer消费消息 代码开发
public class KafkaConsumerTest {
public static Properties getProperties() {
Properties props = new Properties();
//broker地址 填上自己的ip和port
props.put("bootstrap.servers", "ip:port");
//消费者分组ID，分组内的消费者只能消费该消息一次，不同分组内的消费者可以重复消费该消息
props.put("group.id", "test-g1");
//默认是latest，如果需要从头消费partition消息，需要改为 earliest 且消费者组名变更，才生效
props.put("auto.offset.reset","latest");
//开启自动提交offset
props.put("enable.auto.commit", "true");
//序列化
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
return props;
}
@Test
public void simpleConsumerTest(){
Properties properties = getProperties();
KafkaConsumer<String,String> kafkaConsumer = new KafkaConsumer<>(properties);
//订阅主题
kafkaConsumer.subscribe(Arrays.asList(KafkaProducerTest.TOPIC_NAME));
while (true){
//领取时间，阻塞超时时间
ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
for(ConsumerRecord record : records){
System.err.printf("topic=%s, offset=%d,key=%s,value=%s %n",record.topic(),record.offset(),record.key(),record.value());
}
//同步阻塞提交offset
//kafkaConsumer.commitSync();
if(!records.isEmpty()){
//异步提交offset
kafkaConsumer.commitAsync(new OffsetCommitCallback() {
@Override
public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
if(exception == null){
System.err.println("手工提交offset成功:"+offsets.toString());
}else {
System.err.println("手工提交offset失败:"+offsets.toString());
}
}
});
}
}
}
}
© 版权声明
文章版权归作者所有，未经允许请勿转载。
THE END
喜欢就支持一下吧
相关推荐
![设计模式第四弹 – 结构型模式[二]-IT面试星球](https://mianshi.site/wp-content/uploads/2025/04/93fde35337d313ed288a7112489779b5-470x800.png)
暂无评论内容