java kafka 自定义分区

- 日理万妓 2022-05-08 10:26 168阅读 0赞

由于某些特殊情况，可能要采取自定义分区，依赖上一篇有

1.  新建分区类

import org.apache.kafka.clients.producer.Partitioner;
    import org.apache.kafka.common.Cluster;
    
    import java.util.Map;
    
    /** 1. @author lf 2. @描述 自定义分区 3. @date 2018-09-03 */
    public class MyPartition implements Partitioner { 
       /** * * @param topic 主题 * @param key * @param keyBytes key的值的字节 * @param value * @param valueBytes * @param cluster 集成的元数据 * @return */
       public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { 
           // 总共几个分区
           Integer count = cluster.partitionCountForTopic(topic);
           System.out.println("count:"+count);
           String keyStr = key.toString();
           if(count ==3 && keyStr != null){ 
               if("key".equals(keyStr)){ 
                   return 0;
               } else if("key1".equals(keyStr)){ 
                   return 1;
               }else if("key2".equals(keyStr)){ 
                   return 2;
               }
           }
           return 0;
       }
    
       public void close() { 
    
       }
    
       public void configure(Map<String, ?> configs) { 
    
       }
    }

注意，key,key1,key2是你发送消息时，写的key的值  
这里只是试一下效果，具体怎么判断，自行书写

**加入配置**  
找到生产者的类，加入

properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.lf.kafka.MyPartition");

上述都只是一个简单的实现

发表评论取消回复

表情：

评论列表（有 0 条评论，168人围观）

还没有评论，来说两句吧...

相关阅读

相关 Kafka：自定义分区器

实现 Partitioner 接口，重写 partition()方法需求：发送过来的数据中如果包含 hello，就发往 0 号分区，不包含 hello，就发往 1

绝地灬酷狼/ 2024年03月25日 21:00/ 0 赞/ 58 阅读

相关 MapReduce自定义分区Partition

Partition分区默认Partitioner 分区 public class HashPartitioner<K, V> extends Partitio

傷城~/ 2023年10月04日 16:34/ 0 赞/ 47 阅读

相关 Flink 输出到kafka 自定义分区遇到问题，请教大家

简单说一下，scala代码添加了自定义分区器，一直报错 ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM

浅浅的花香味﹌/ 2022年12月17日 01:23/ 0 赞/ 190 阅读

相关 Kafka自定义offset存储

在Kafka中，offset默认存储在broker的Topic，我们也可以自定义存储位置，为了保证消费和提交偏移量同时成功或失败，我们可以利用数据库事务来实现，下面是把offs

野性酷女/ 2022年12月03日 15:59/ 0 赞/ 185 阅读

相关 Kafka分区策略及自定义

默认分区策略默认分区策略是：取正（bytearray生成32位hash值）%numpartitions 这个公式的结果是得到0-(numpartitions-1)间

柔情只为你懂/ 2022年06月14日 07:29/ 0 赞/ 185 阅读

相关 java kafka 自定义分区

由于某些特殊情况，可能要采取自定义分区，依赖上一篇有 1. 新建分区类 import org.apache.kafka.clients.producer.Part

- 日理万妓/ 2022年05月08日 10:26/ 0 赞/ 169 阅读

相关 Kafka学习（9）- kafka使用Java开发-5-自定义分区实例

目录： 1. 操作流程： 1. 创建一个maven项目，项目结构目录 2. 生产者代码（往主题test-1，key为139开头的号码（此号码属

港控/mmm°/ 2022年03月27日 05:59/ 0 赞/ 177 阅读

相关 Kafka学习（8）- kafka使用Java开发-4-自定义分区

目录： 1. 如何在某个主题下的某个分区拿数据（假设my-topic有3个分区） 2. 为什么要自定义分区 3. 自定义分区操作流程 1. 创建自定

墨蓝/ 2022年03月27日 04:17/ 0 赞/ 220 阅读

相关 Hadoop mapreduce自定义分区HashPartitioner

本文发表于本人[博客][Link 1]。在上一篇[文章][Link 2]我写了个简单的WordCount程序，也大致了解了下关于mapreduce运行原来，其中说到

- 日理万妓/ 2021年11月23日 11:56/ 0 赞/ 336 阅读

相关 Hadoop入门——自定义分区Partitioner

一.背景为了使得MapReduce计算后的结果显示更加人性化，Hadoop提供了分区的功能，可以使得MapReduce计算结果输出到不同的分区中，方便查看。Hadoop提供

「爱情、让人受尽委屈。」/ 2021年09月27日 07:26/ 0 赞/ 472 阅读