JavaKafka分区发送及消费实战

JavaKafka分区发送及消费实战

目录

前言

业务场景

业务实现

不指定分区

指定分区

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分区发送及消费内容请搜索易知道(ezd.cc)以前的文章或继续浏览下面的相关文章希望大家以后多多支持易知道(ezd.cc)!

推荐阅读

    金蝶担保机构业务管理系统

    金蝶担保机构业务管理系统,,1.金蝶财务软件怎么使用?软件使用方法 使用需要先登陆到数据库,可只导出V3数据或只导出最新年度数据。功能介绍

    提高3A四核羿龙II游戏配置的性能

    提高3A四核羿龙II游戏配置的性能,,以节能环保为主题的IT产业,目前3A低端平台处理器、主板芯片组、独立开发卡性能突出,特别是在与AMD的处理

    优化PostgreSQL中的批量更新性能

    优化PostgreSQL中的批量更新性能,数据,表格,在Ubuntu 12.04上使用PG 9.1. 我们目前需要24小时才能运行大量UPDATE数据库上的语句,其形式

    诺基亚威图性能好到哪里

    诺基亚威图性能好到哪里,诺基亚,手机,诺基亚威图性能好到哪里这是一部以前列出的手机。即使当时配置不高,该品牌的手机也不依赖于该功能吸

    魅蓝note6性能参数有哪些

    魅蓝note6性能参数有哪些,摄像头,蓝牙,魅蓝note6性能参数有哪些魅力蓝色Note6最好拍照。电池寿命更长。蓝色Note6使用高通 snapdragon 625

    国产电脑cpu测试|国产CPU性能

    国产电脑cpu测试|国产CPU性能,,国产CPU性能天玑9000答: 天玑9000更厉害。因为天玑9000是 最新发布的cpu,也是现在的天花板。而麒麟9000是 2

    主流电脑cpu性能分析|cpu性能对比表

    主流电脑cpu性能分析|cpu性能对比表,,1. cpu性能对比表一、参数对比1、r7 5800H:制程工艺为7nm,主频3.2GHz,睿频4.4GHz,线程数是8核16线程,45W

    新老酷睿i3性能对比试验i34130和i33220

    新老酷睿i3性能对比试验i34130和i33220,,新的英特尔酷睿i3-4130 Haswell处理器架构已经推出了很长一段时间,虽然市场的时间还不长,已经成为