java分布式流式处理组件Producer分区的作用是什么 |
您所在的位置：网站首页 › java打开是什么样子 › java分布式流式处理组件Producer分区的作用是什么 |
java分布式流式处理组件Producer分区的作用是什么
|
java分布式流式处理组件Producer分区的作用是什么
发布时间：2023-03-07 11:31:17
来源：亿速云
阅读：82
作者：iii
栏目：开发技术
这篇文章主要讲解了“java分布式流式处理组件Producer分区的作用是什么”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“java分布式流式处理组件Producer分区的作用是什么”吧！ 为什么需要分区分区的作用合理的使用存储资源：把海量的数据按照分区切割成一小块的数据存储在多台Broker上。此时能够保证每台服务器存储资源能够被充分利用到。而且小块数据在寻址时间上更有优势~ 如果将全部的数据存储在一台机器上，那么要对当前数据做副本的时候，由于服务器资源配置不同，就有可能会出现副本数据存放失败，从而增加数据丢失的可能性。 同时，如果单个文件过大，副本放置时间、内容检索时间都会极大的延长，从而导致Kafka性能降低。 负载均衡： 数据生产或消费期间，生产者已分区的单位发送数据，消费者分区的单位进行消费。 期间，各分区生产和消费数据互不影响，这样能够达到合理控制分区任务的程度，提高任务的并行度。从而达到负载均衡的效果。 刚才我们提到：生产者已分区为单位向Broker发送数据。那么问题来了： 生产者是怎么知道该向哪个分区发送数据呢？ 这就是我们接下来要研究的分区策略。 分区策略private int partition(ProducerRecord record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) { // 如果在消息中指定了分区 if (record.partition() != null) return record.partition(); if (partitioner != null) { // 分区器通过计算得到分区 int customPartition = partitioner.partition( record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster); if (customPartition new AtomicInteger(0)); return counter.getAndIncrement(); } public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { List partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); int nextValue = nextValue(topic); List availablePartitions = cluster.availablePartitionsForTopic(topic); if (!availablePartitions.isEmpty()) { int part = Utils.toPositive(nextValue) % availablePartitions.size(); return availablePartitions.get(part).partition(); } else { // no partitions are available, give a non-available partition return Utils.toPositive(nextValue) % numPartitions; } }这个类的解释，嗯。。你们看那个合适吧～
其实这个逻辑非常简单： 通过AtomicInteger.getAndIncrement()的方式将每次写入平均分配到不同的分区中 不同与其他分区策略类，它不关心Key是否为NULL 我们先来做个小实验吧： 将分区策略类修改为RoundRobinPartitioner，也方便后续自定义分区器的配置操作 config.setProperty( ProducerConfig.PARTITIONER_CLASS_CONFIG, "org.apache.kafka.clients.producer.RoundRobinPartitioner" );就这样就能实现，看结果验证～
中间穿插了一点小知识，那么接下来就会进入到我们最后一个环节：尝试自定义分区器 自定义分区器前面我们也提到过，相信大家没有忘记partitioner.class这个配置 那么接下来就进入到重头戏：自定义分区器实战编码环节。 public class CustomPartitioner implements Partitioner { @Override public void configure(Map configs) { // nothing } @Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { // 如果keyBytes == null // 直接去0号位置 if (null == keyBytes) { return 0; } // 已默认分区策略实现 int numPartitions = cluster.partitionsForTopic(topic).size(); return BuiltInPartitioner.partitionForKey(keyBytes, numPartitions); } @Override public void close() { // nothing } }我们就先做的简单一点，主要是想让大家明白自定义分区器的实现： 如果没有给定指定key，那么就默认全部去0号分区 否则就通过key做取模计算 当自定义分区器实现完成之后，接下来我们就需要通过发送者进行验证。当然了，主要还是通过partitioner.class进行修改 // 给出关键代码，其他的都是一样的。就不赘述了~~~ config.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG, "top.zopx.kafka.partitioner.CustomPartitioner");通过执行之后，我们来看看它的运行效果是否满足我们的预期
另一种运行结果与默认分区器有Key的情况类似，这里就不再重复贴图 感谢各位的阅读，以上就是“java分布式流式处理组件Producer分区的作用是什么”的内容了，经过本文的学习后，相信大家对java分布式流式处理组件Producer分区的作用是什么这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是亿速云，小编将为大家推送更多相关知识点的文章，欢迎关注！ 推荐阅读： Ubuntu16 安装Oracle JDK 8 Oracle Java SE 任意代码执行漏洞加固免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：[email protected]进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。 java producer 上一篇新闻：go语言int64整型转字符串如何实现 下一篇新闻：windows中program files可不可以移到D盘 猜你喜欢 vue中mixins属性怎么用 怎么使用vue-router实现组件切换功能 vue下如何设置二级路由 怎么用Vue代码实现一个搜索匹配功能 如何使用vue-prop实现父组件向子组件进行传值 如何使用vue实现一个评星组件 vue下如何用axios实现表单提交上传图片 vue实现子组件向父组件传值的代码怎么写 Vue下怎么用v-for指令实现数据分组 如何使用vue实现导入导出功能 |
【本文地址】
今日新闻 |
推荐新闻 |