Java Kafka分区发送及消费实战

目录
  • 前言
  • 业务场景
  • 业务实现
    • 不指定分区
  • 指定分区
    • topic分区初始化及配置
    • 生产者分区发送方案
    • 消费者

前言

Kafka是现在非常热门的分布式消息队列,常用于微服务间异步通信,业务解耦等场景。kafka的性能非常强大,但是单个微服务吞吐性能是有上限的,我们就会用到分布式微服务,多消费者多生产者进行数据处理,保证性能同时也能根据业务量进行横向拓展,对于同一个微服务的多个实例,输入输出的topic是同一个,这时候我们就可以利用Kafka分区消费来解决这个问题。

业务场景

我们开发的是一个物联网系统,大量设备接入到平台实时发送数据,有秒级数据和分钟级别数据等等,处理流程包含接入、处理、存储,这三个模块间就是使用kafka进行数据流转,数据处理模块中包含多个微服务,单条数据会经历多次处理,部分业务耗时较长,导致在高频率接收到数据时候单体服务无法达到吞吐平衡,于是对于这些服务进行了分布式部署,多个实例进行消费处理。

业务实现

不指定分区

我们在给kafka发送消息时候,如果不指定分区,是不需要手动创建topic的,发送时没有topic,kafka会自动创建一个分区为1的topic,如下:

@Service
public class ProductService {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void send(String msg, String topic) {
        kafkaTemplate.send(topic, msg);
    }
}

指定分区

topic分区初始化及配置

指定分区发送时候,如果未配置topic分区数,指定>0的分区,会提示分区不存在,这时候我们就需要提前创建好topic及分区

手动创建,服务启动前,使用kafka tool手动创建topic 不推荐 x

自动创建,服务启动时,使用KafkaClient创建 推荐 √

/**
 * 初始化多分区的topic 基于springboot2
 */
@Component
public void TopicInitRunner implements ApplicationRunner {

    @Autowired
    private AdminClient adminClient;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        // 通过配置文件读取自定义配置的topic名及分区数 省略...
        // Key topic V 分区数
        Map<String, Integer> topicPartitionMap = new HashMap<>();
        for (Map.Entry<String, Integer> e : topicPartitionMap.entrySet()) {
            createTopic(e.getKey(), e.getValue());
        }

    }

    public void createTopic(String topic, int partition) {
        NewTopic newTopic = new NewTopic(topic, partition);
        adminClient.createTopics(Lists.newArrayList(newTopic));
    }
}

/**
 * 配置类参考 基于springboot2
 * 如果只进行普通的单消息发送 无需添加此配置到项目中
 */
@Configuration
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String servers;

    @Bean
    public AdminClient adminClient() {
        return AdminClient.create(kafkaAdmin().getConfig());
    }

    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> props = Maps.newHashMap();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        return new KafkaAdmin(props);
    }
}

生产者分区发送方案

上面讲到如何初始化分区topic,这时候我们的kafka环境已经准备好了,我们先使用TopicInitRunner为我们创建一个名称为 partition-topic 分区数为三,现在讲一讲如何均匀的讲消息发送的每个分区上,如何保证多消费者实例是负载均衡的,具体方案如下:

  • 1.因为每条消息都是设备上传的,都会有设备id,先给每个设备生成一个自增号,这样1000个设备,每个设备就会有0到999的自增号,放到缓存中,每次根据消息中的设备id获取到该设备的自增号
  • 2.使用自增号对分区数进行取模操作,代码实现如下:
public class ProductService {
    /**
     * data为需要发送的数据
     */
    public void partitionSend(String topic, int partition, JSONObject data) {
         // 获取设备id
        String deviceId = data.getString("deviceId");
        // 获取自增数 如果是新设备会创建一个并放入缓存中
        int inc = getDeviceInc(deviceId);
        // 如果分区数为3 设备自增id为1 取模结果为1 就是发送到1分区 这样1000个设备就可以保证每个分区发送数据量是1000 / 3
        int targetPartition = Math.floorMod(inc, partition);
        // 分区发送时候 需要指定一个唯一k 可以使用uuid或者百度提供的雪花算法获取id 字符串即可
        kafkaTemplate.send(topic, partition, getUuid(), data.toJSONString());
    }
}

消费者

我们讲到消费者使用分布式部署,一个微服务有多个实例,我们只需要按照服务监听的topic分区数创建对应数目的服务实例即可,这样kafka就会自动分配对应分区的数据到每个实例。

我们采取批量消费,进一步提高服务吞吐性能,消费及配置代码如下,配置文件参考springbootkafka配置即可,主要设计kafka服务配置,消费及生产配置,比较核心的是

@Component
public class DataListener {

    @Autowired
    private MongoTemplate mongoTemplate;

    /**
     * 站点报文监听消费
     *
     * @param records
     */
    @KafkaListener(topics = "partition-topic", containerFactory = "batchConsumerFactory")
    public void iotSiteHistoryMessageConsumer(List<ConsumerRecord<String, String>> records) {
    }

    /**
     * 消费者配置
     */
    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = Maps.newHashMap();
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 120000);
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 180000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }

    /**
     * 批量消费配置
     */
    @Bean
    public KafkaListenerContainerFactory batchConsumerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
        factory.setBatchListener(true);
        return factory;
    }
}

到此这篇关于Java Kafka分区发送及消费实战的文章就介绍到这了,更多相关Kafka分区发送及消费内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

(0)

相关推荐

  • 使用kafka如何选择分区数及kafka性能测试

    kafka选择分区数及kafka性能测试 1.简言 ​ 如何选择合适的分区,这是我们经常面临的问题,不过针对这个问题,在网上并没有搜到固定的答案.因此,今天在这里主要通过性能测试的工具来告诉如何选择相对应的kafka分区. 2.性能测试工具 ​ kafka本身提供了比较的性能测试工具,我们可以使用它来测试适用于我们机器的kafka分区. ① 生产者性能测试 分别创建三个topic,副本数设置为1. bin/kafka-topics.sh --zookeeper zk --create --rep

  • 解决kafka消息堆积及分区不均匀的问题

    目录 kafka消息堆积及分区不均匀的解决 1.先在kafka消息中创建 2.添加配置文件application.properties 3.创建kafka工厂 4.展示kafka消费者 kafka出现若干分区不消费的现象 定位过程 验证 解决方法 kafka消息堆积及分区不均匀的解决 我在环境中发现代码里面的kafka有所延迟,查看kafka消息发现堆积严重,经过检查发现是kafka消息分区不均匀造成的,消费速度过慢.这里由自己在虚拟机上演示相关问题,给大家提供相应问题的参考思路. 这篇文章有点

  • 详解kafka中的消息分区分配算法

    目录 背景 RangeAssignor 定义 源码分析 场景 RoundRobinAssignor 定义 源码分析 场景 StickyAssignor 定义 场景 背景 kafka有分区机制,一个主题topic在创建的时候,会设置分区.如果只有一个分区,那所有的消费者都订阅的是这一个分区消息:如果有多个分区的话,那消费者之间又是如何分配的呢? 分配算法 RangeAssignor 定义 Kafka默认采⽤RangeAssignor的分配算法. RangeAssignor策略的原理是按照消费者总数

  • kafka手动调整分区副本数的操作步骤

    目录 前言 前置准备 操作步骤 增加副本 前言 在生产环境中,akfka集群下的每台服务器的配置和性能可能不一样,但Kafka集群只会根据配置规则创建对应的分区副本,这样一来可能就会导致个别服务器存储压力较大. 在这种情况下,就需要手动调整分区副本的存储.我们不妨看看下面的这张图就明白了 上图的传达的意思是: broker0和broker1所在的服务器配置较高,存储容量较大,但是broker2和broker3所在的机器配置稍差存储容量较小,一开始创建出主题之后,集群只会按照默认的配置规则,将le

  • Apache Kafka 分区重分配的实现原理解析

    目录 一.前言 二.工具的使用 三.元数据管理及协调器 3.1 ZooKeeper 3.2 Kafka Controller 四.分区重分配流程分析 4.1 kafka-reassign-partitions 客户端 4.2 controller 维护分区的元数据信息 4.3 broker 端数据跨路径迁移 五.总结 本文作者为中国移动云能力中心大数据团队软件开发工程师孙大鹏,本文结合 2.0.0 版本的 Kafka 源码,详细介绍了 Kafka 分区副本重分配的流程和逻辑,供大家参考. 一.前

  • Java kafka如何实现自定义分区类和拦截器

    生产者发送到对应的分区有以下几种方式: (1)指定了patition,则直接使用:(可以查阅对应的java api, 有多种参数) (2)未指定patition但指定key,通过对key的value进行hash出一个patition: (3)patition和key都未指定,使用轮询选出一个patition. 但是kafka提供了,自定义分区算法的功能,由业务手动实现分布: 1.实现一个自定义分区类,CustomPartitioner实现Partitioner import org.apache

  • Java Kafka分区发送及消费实战

    目录 前言 业务场景 业务实现 不指定分区 指定分区 topic分区初始化及配置 生产者分区发送方案 消费者 前言 Kafka是现在非常热门的分布式消息队列,常用于微服务间异步通信,业务解耦等场景.kafka的性能非常强大,但是单个微服务吞吐性能是有上限的,我们就会用到分布式微服务,多消费者多生产者进行数据处理,保证性能同时也能根据业务量进行横向拓展,对于同一个微服务的多个实例,输入输出的topic是同一个,这时候我们就可以利用Kafka分区消费来解决这个问题. 业务场景 我们开发的是一个物联网

  • Java Kafka 消费积压监控的示例代码

    后端代码: Monitor.java代码: package com.suncreate.kafkaConsumerMonitor.service; import com.suncreate.kafkaConsumerMonitor.model.ConsumerInfo; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.KafkaConsumer; i

  • Java Kafka实现延迟队列的示例代码

    目录 基于kafka如何实现延迟队列 完善细节 Java代码实现 还需要做什么 kafka作为一个使用广泛的消息队列,很多人都不会陌生,但当你在网上搜索“kafka 延迟队列”,出现的都是一些讲解时间轮或者只是提供了一些思路,并没有一份真实可用的代码实现,今天我们就来打破这个现象,提供一份可运行的代码,抛砖引玉,吸引更多的大神来分享. 基于kafka如何实现延迟队列 想要解决一个问题,我们需要先分解问题.kafka作为一个高性能的消息队列,只要消费能力足够,发出的消息都是会立刻收到的,因此我们需

  • java中javamail发送带附件的邮件实现方法

    本文实例讲述了java中javamail发送带附件的邮件实现方法.分享给大家供大家参考.具体分析如下: JavaMail,顾名思义,提供给开发者处理电子邮件相关的编程接口.它是Sun发布的用来处理email的API.它可以方便地执行一些常用的邮件传输,JavaMail是可选包,因此如果需要使用的话你需要首先从java官网上下载.目前最新版本是JavaMail1.5.0,下面我们来看看javamail发送带附件的邮件实例 mail.java 代码: 复制代码 代码如下: package mail;

  • Java HtmlEmail 邮件发送的简单实现代码

    Java 项目中常常回遇到发送邮件 Java 发送邮件有几种,今天先给大家介绍用 HtmlEmail 来发送邮件,我这里是用 Maven 来搭建的 HtmlEmail 可以抄带HTML 首先 需要导入jar 包 <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-email</artifactId> <version>1.4</versio

  • Java 注册时发送激活邮件和激活的实现示例

    Java 注册时发送激活邮件和激活的实现示例 最近从项目分离出来的注册邮箱激活功能,整理一下,方便下次使用 1.RegisterController.java package com.app.web.controller; import java.text.ParseException; import javax.annotation.Resource; import javax.servlet.http.HttpServletRequest; import javax.servlet.http

  • java 使用HttpURLConnection发送数据简单实例

    java 使用HttpURLConnection发送数据简单实例 每个 HttpURLConnection 实例都可用于生成单个请求,但是其他实例可以透明地共享连接到 HTTP 服务器的基础网络.请求后在 HttpURLConnection 的 InputStream 或 OutputStream 上调用 close() 方法可以释放与此实例关联的网络资源,但对共享的持久连接没有任何影响.如果在调用 disconnect() 时持久连接空闲,则可能关闭基础套接字.JAVA使用HttpURLCon

  • Java Web stmp发送带附件邮件(附SSL版)

    本文实例为大家分享了Java Web stmp发送带附件邮件的具体代码,供大家参考,具体内容如下 public class MailFileSendUtils { private Properties props; //系统属性 private Session session; //邮件会话对象 private MimeMessage mimeMsg; //MIME邮件对象 private Multipart mp; //Multipart对象,邮件内容,标题,附件等内容均添加到其中后再生成Mi

  • Java实现后台发送及接收json数据的方法示例

    本文实例讲述了Java实现后台发送及接收json数据的方法.分享给大家供大家参考,具体如下: 本篇博客试用于编写java后台接口以及两个项目之间的接口对接功能: 具体的内容如下: 1.java后台给指定接口发送json数据 package com.utils; import java.io.BufferedReader; import java.io.InputStreamReader; import java.io.OutputStreamWriter; import java.net.Htt

随机推荐