【Kafka】Java实现数据的生产和消费

news2024/11/8 11:06:24

【Kafka】Java实现数据的生产和消费

Kafka介绍

Kafka 是由 Linkedin 公司开发的，它是一个分布式的，支持多分区、多副本，基于 Zookeeper 的分布式消息流平台，它同时也是一款开源的基于发布订阅模式的消息引擎系统。
Kafka 有如下特性：

  • 以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间复杂度的访问性能；
  • 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条以上消息的传输；
  • 支持Kafka Server间的消息分区，及分布式消费，同时保证每个Partition内的消息顺序传输；
  • 同时支持离线数据处理和实时数据处理；
  • Scale out：支持在线水平扩展。

Kafka术语

  • Broker：Kafka集群包含一个或多个服务器，这种服务器被称为broker；
  • Topic：每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）；
  • Partition：Partition是物理上的概念，每个Topic包含一个或多个Partition；
  • Producer：负责发布消息到Kafka broker；
  • Consumer：消息消费者，向Kafka broker读取消息的客户端；
  • Consumer Group：每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）。

Kafka特性

  • 高吞吐、低延迟：kakfa 最大的特点就是收发消息非常快，kafka 每秒可以处理几十万条消息，它的最低延迟只有几毫秒；
  • 高伸缩性： 每个主题(topic) 包含多个分区(partition)，主题中的分区可以分布在不同的主机(broker)中；
  • 持久性、可靠性： Kafka 能够允许数据的持久化存储，消息被持久化到磁盘，并支持数据备份防止数据丢失，Kafka 底层的数据存储是基于 Zookeeper 存储的，Zookeeper 的数据能够持久存储；
  • 容错性： 允许集群中的节点失败，某个节点宕机，Kafka 集群能够正常工作；
  • 高并发： 支持数千个客户端同时读写。

Kafka应用场景

  • 活动跟踪：Kafka 可以用来跟踪用户行为，比如你经常回去App购物，你打开App的那一刻，你的登陆信息，登陆次数都会作为消息传输到 Kafka ，当你浏览购物的时候，你的浏览信息，你的搜索指数，你的购物爱好都会作为一个个消息传递给 Kafka ，这样就可以生成报告，可以做智能推荐，购买喜好等；
  • 传递消息：Kafka 另外一个基本用途是传递消息，应用程序向用户发送通知就是通过传递消息来实现的，这些应用组件可以生成消息，而不需要关心消息的格式，也不需要关心消息是如何发送的；
  • 度量指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告；
  • 日志记录：Kafka 的基本概念来源于提交日志，比如可以把数据库的更新发送到 Kafka 上，用来记录数据库的更新时间，通过Kafka以统一接口服务的方式开放给各种consumer，例如hadoop、Hbase、Solr等；
  • 流式处理：流式处理是有一个能够提供多种应用程序的领域；
  • 限流削峰：Kafka 多用于互联网领域某一时刻请求特别多的情况下，可以把请求写入Kafka 中，避免直接请求后端程序导致服务崩溃。

以上介绍参考Kafka官方文档。

Kafka核心Api

Kafka有4个核心API

  • 应用程序使用producer API发布消息到1个或多个topic中；
  • 应用程序使用consumer API来订阅1个或多个topic，并处理产生的消息；
  • 应用程序使用streams API充当一个流处理器,从1个或多个topic消费输入流，并产生一个输出流到1个或多个topic,有效地将输入流转换到输出流；
  • connector API允许构建或运行可重复使用的生产者或消费者，将topic链接到现有的应用程序或数据系统。

Kafka为何如此之快

Kafka 实现了零拷贝原理来快速移动数据，避免了内核之间的切换。Kafka 可以将数据记录分批发送，从生产者到文件系统（Kafka 主题日志）到消费者，可以端到端的查看这些批次的数据。批处理能够进行更有效的数据压缩并减少 I/O 延迟，Kafka 采取顺序写入磁盘的方式，避免了随机磁盘寻址的浪费。

总结一下其实就是四个要点：

  • 顺序读写
  • 零拷贝
  • 消息压缩
  • 分批发送

案例

项目创建：

dependencies：

构建工具为Maven，Maven的依赖如下：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.12</artifactId>
    <version>1.0.0</version>
    <scope>provided</scope>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>1.0.0</version>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>1.0.0</version>
</dependency>

Kafka Producer

package cn.com.codingce.module;

import java.util.Properties;
import java.util.Random;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

public class Producer {

    // 定义主题
    public static String topic = "codingce_test";

    public static void main(String[] args) throws InterruptedException {
        Properties p = new Properties();
        // bootstrap.servers: kafka的地址, 多个地址用逗号分割
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.31.150:9092");
        // acks:消息的确认机制，默认值是0. acks=0: 如果设置为0，生产者不会等待kafka的响应; acks=1: 这个配置意味着kafka会把这条消息写到本地日志文件中，但是不会等待集群中其他机器的成功响应
        // acks=all: 这个配置意味着leader会等待所有的follower同步完成. 这个确保消息不会丢失, 除非kafka集群中所有机器挂掉. 这是最强的可用性保证. 
        p.put("acks", "all");
        // retries: 配置为大于0的值的话, 客户端会在消息发送失败时重新发送.
        p.put("retries", 0);
        // batch.size: 当多条消息需要发送到同一个分区时，生产者会尝试合并网络请求. 这会提高client和生产者的效率.
        p.put("batch.size", 16384);
        // key.serializer: 键序列化，默认org.apache.kafka.common.serialization.StringDeserializer. 
        p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // value.deserializer:值序列化，默认org.apache.kafka.common.serialization.StringDeserializer. 
        p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(p);
        try {
            do {
                String msg = "后端码匠, " + new Random().nextInt(100);
                ProducerRecord<String, String> record = new ProducerRecord<>(topic, msg);
                kafkaProducer.send(record);
                System.out.println("======消息发送成功: " + msg + " ======");
                Thread.sleep(1000L);
            } while (true);
        } finally {
            kafkaProducer.close();
        }

    }
}

Kafka Consumer

package cn.com.codingce.module;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;

public class Consumer {
    private static final String GROUPID = "codingce_consumer_a";

    public static void main(String[] args) {

        Properties p = new Properties();
        // bootstrap.servers: kafka的地址, 多个地址用逗号分割
        p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.31.150:9092");
        // 消费者所属的分组id, 组名 不同组名可以重复消费.例如你先使用了组名A消费了Kafka的1000条数据, 但是你还想再次进行消费这1000条数据, 
        // 并且不想重新去产生, 那么这里你只需要更改组名就可以重复消费了.
        p.put(ConsumerConfig.GROUP_ID_CONFIG, GROUPID);
        // 是否自动提交, 默认为true.
        p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // 从poll(拉)的回话处理时长
        p.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        // 超时时间
        p.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        // 一次最大拉取的条数
        p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000);
        // 消费规则, 默认earliest
        p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // key.serializer: 键序列化, 默认org.apache.kafka.common.serialization.StringDeserializer.
        p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // value.deserializer:值序列化, 默认org.apache.kafka.common.serialization.StringDeserializer.
        p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(p);
        // 订阅消息
        kafkaConsumer.subscribe(Collections.singletonList(Producer.topic));
        do {
            // 订阅之后, 再从kafka中拉取数据
            ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("-----topic:%s, offset:%d, 消息:%s-----\n", record.topic(), record.offset(), record.value());
            }
        } while (true);
    }
}

本次采用Docker 搭建的单机 Kafka、Zookeeper，Kafka介绍参考官方文档：http://kafka.apache.org/intro

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/142014.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

【LeetCode】1803. 统计异或值在范围内的数对有多少

【LeetCode】1803. 统计异或值在范围内的数对有多少

1803. 统计异或值在范围内的数对有多少 题目描述 给你一个整数数组 nums &#xff08;下标 从 0 开始 计数&#xff09;以及两个整数&#xff1a;low 和 high &#xff0c;请返回 漂亮数对 的数目。 漂亮数对 是一个形如 (i, j) 的数对&#xff0c;其中 0 < i < j <…
阅读更多...
阻塞式队列

阻塞式队列

文章目录一、阻塞队列阻塞队列的概念阻塞队列相关类和方法生产者消费者模型二、自定义实现阻塞队列自定义实现循环队列自定义实现阻塞队列生产者消费者模型体验一、阻塞队列 阻塞队列的概念 队列我们并不默认&#xff0c;一提起队列&#xff0c;我们立马就能想到 "先进先…
阅读更多...
一条 update 语句的执行过程

一条 update 语句的执行过程

1.一条 update 语句的执行流程 一条更新语句&#xff0c;其实是增&#xff0c;删&#xff0c;查的综合体&#xff0c;查询语句需要经过的流程&#xff0c;更新语句全部需要执行一次&#xff0c;因为更新之前必须要先拿到&#xff08;查询&#xff09;需要更新的数据。 Buffer…
阅读更多...
低代码平台组件间通信方案复盘

低代码平台组件间通信方案复盘

背景介绍3年前我开发了一款零代码搭建平台 H5-Dooring, 主要目的是想用更低的成本, 更快的效率, 上线 web 页面(其实是不想写重复的代码了,写麻了). 好在陆陆续续折腾了3年, 目前已经可以满足基本的页面设计和搭建能力, 并能快速上线页面.之前也在社区分享了很多低代码和零代码…
阅读更多...
高、低成本MEMS惯导系统姿态、位置、速度更新算法的对比

高、低成本MEMS惯导系统姿态、位置、速度更新算法的对比

高、低成本MEMS惯导系统姿态、位置、速度更新算法的对比一、高成本MEMS惯导系统姿态、位置、速度更新算法1、速度更新2、位置更新3、姿态更新4、程序仿真及实验结果4.1 主函数4.2 子函数4.3 实验结果一、低成本MEMS惯导系统姿态、位置、速度更新算法1、速度更新2、位置更新3、姿…
阅读更多...
乾元浩在创业板IPO终止：主要生产禽用疫苗产品，中农大是股东

乾元浩在创业板IPO终止：主要生产禽用疫苗产品，中农大是股东

2023年1月4日&#xff0c;深圳证券交易所披露的信息显示&#xff0c;乾元浩生物股份有限公司&#xff08;下称“乾元浩”&#xff09;提交了撤回上市申请文件的申请&#xff0c;保荐人中信证券也撤回对该公司的保荐。因此&#xff0c;深交所终止了乾元浩首次公开发行股票并在创…
阅读更多...
程序员述职报告

程序员述职报告

程序员述职报告笔者能力有限&#xff0c;仅供参考。做研发的小伙伴&#xff0c;不太擅长于写PPT&#xff0c;对于更高一级别的领导&#xff0c;可能只有年终述职的时候才能全面的了解你的工作。所以需要我们在PPT中表达自我价值&#xff0c;突出角色职责。让领导更清楚的认识你…
阅读更多...
访问数据库超时问题排障

访问数据库超时问题排障

1 排障过程 系统从圣诞节那天晚上开始&#xff0c;每天晚上固定十点多到十一点多这个时段&#xff0c;大概瘫痪1h左右&#xff0c;过这时段系统自动恢复。系统瘫痪时的现象就是&#xff0c;网页和App都打不开&#xff0c;请求超时。系统架构&#xff1a; 整个系统托管在公有云…
阅读更多...
图解数据结构：盘点链表与栈和队列的那些血缘（单双链表模拟实现栈和队列）

图解数据结构：盘点链表与栈和队列的那些血缘（单双链表模拟实现栈和队列）

写在前面 Hello&#xff0c;各位盆友们&#xff0c;我是黄小黄。关于前一段时间为什么拖更这件事&#xff0c;这里给大家说一句抱歉。笔者前段时间忙于ddl和一些比赛相关的事件&#xff0c;当然还有些隐藏任务&#xff0c;所以博文更新就放缓了。  这里还需要做一下对以后博文…
阅读更多...
计算机原理一_计算机的组成、进程与线程

计算机原理一_计算机的组成、进程与线程

目录儿一、计算机组成二、进程与线程2.1 线程的切换2.2 CPU的并发控制2.2.1 关中断2.2.2 缓存一致性协议2.2.2.1 缓存Cache2.2.2.2 缓存行Cache Line2.2.2.3 缓存一致性拓展&#xff1a;超线程2.2.3 内存屏障2.2.3.1 CPU的乱序执行拓展1&#xff1a;java 的 this 溢出问题拓展2…
阅读更多...
Linux（一）：Linux基本结构

Linux（一）：Linux基本结构

一、Linux系统划分 linux系统分为用户区、内核区 1.1分区目标 保护数据和硬件安全&#xff0c;对系统进行分区也就是进程分区&#xff0c;当处于用户态&#xff0c;只能访问用户区&#xff0c;用户无法修改内核&#xff0c;保证硬件安全&#xff0c;操作系统不易损坏&#…
阅读更多...
【Qt】QMainWindow应用程序窗口类简单介绍

【Qt】QMainWindow应用程序窗口类简单介绍

QMainWindow介绍 QMainWindow是一个为用户提供主窗口程序的类&#xff0c;是许多应用程序的基础&#xff0c;包含的组件有&#xff1a; 菜单栏QMenuBar&#xff0c;一个主窗口最多只能有一个菜单栏&#xff1b;包含一个下拉菜单项的列表&#xff0c;这些菜单项由QAction动作类…
阅读更多...
【git版本控制】| git版本控制操作命令（全）

【git版本控制】| git版本控制操作命令（全）

文章目录一、简介二、工作模式1 集中式&#xff08;CVS、SVN&#xff09;2 分布式Git三、Git1 工作模式2 git工作流程3 工作区和版本库4 注意事项5 基本操作5.1 创建本地版本库5.2 初始化本地版本库5.3 .git目录的作用5.4 创建用户5.5 其他操作6 git分支7 常见警告8 免密登录9 …
阅读更多...
interface接口--GO面向对象编程思想

interface接口--GO面向对象编程思想

一、interface接口 interface 是GO语言的基础特性之一。可以理解为一种类型的规范或者约定。它跟java&#xff0c;C# 不太一样&#xff0c;不需要显示说明实现了某个接口&#xff0c;它没有继承或子类或“implements”关键字&#xff0c;只是通过约定的形式&#xff0c;隐式的…
阅读更多...
【C语言进阶】自定义类型：结构体，枚举，联合体

【C语言进阶】自定义类型：结构体，枚举，联合体

目录 1、结构体的声明 1.1 结构体基础知识 1.2 结构体的声明 1.3 特殊的声明 1.4 结构体的自引用 1.5 结构体变量的定义和初始化 1.6 结构体内存对齐 ​编辑1.7 修改默认对齐数 1.8 结构体传参 2. 位段 2.1 什么是位段 2.2 位段的内存分配 2.3 位段的跨平台问…
阅读更多...
【owt-server】代码结构及新增一个agent

【owt-server】代码结构及新增一个agent

owt server 官方 5.0 仓库：代码结构 manage console manage api portal sip portal 与agent 并列 agent又有很多种类。 启动脚本 启动一个新的agent 比如streaming-agent streaming-agent )cd ${OWT_HOME}/s
阅读更多...
分布式id

分布式id

分布式id一 什么是分布式系统唯一ID二 分布式系统唯一ID的特点三 分布式系统唯一ID的实现方案3.1 基于UUID3.2 基于数据库自增id3.3 基于数据库集群模式3.4 基于Redis模式3.5 基于雪花算法&#xff08;Snowflake&#xff09;模式3.6 百度&#xff08;uid-generator&#xff09;…
阅读更多...
Python爬虫数据到sqlite实例

Python爬虫数据到sqlite实例

参考链接:https://blog.csdn.net/qq_45775027/article/details/115319253最近需要使用到爬虫数据库&#xff0c;原文中作者有些没补齐&#xff0c;略作修改之后跑通了。主要修改&#xff1a;1.调整了数据获取的正则表达式&#xff1b;2. 改了一下数据库的table名和定义名字&…
阅读更多...
基于Java+SpringBoot+vue+element实现前后端分离牙科诊所管理系统详细设计

基于Java+SpringBoot+vue+element实现前后端分离牙科诊所管理系统详细设计

基于JavaSpringBootvueelement实现前后端分离牙科诊所管理系统详细设计 博主介绍&#xff1a;5年java开发经验&#xff0c;专注Java开发、定制、远程、文档编写指导等,csdn特邀作者、专注于Java技术领域 作者主页 超级帅帅吴 欢迎点赞 收藏 ⭐留言 文末获取源码联系方式 文章目…
阅读更多...
【Linux】虚拟地址空间 --- 虚拟地址、空间布局、内存描述符、写时拷贝、页表…

【Linux】虚拟地址空间 --- 虚拟地址、空间布局、内存描述符、写时拷贝、页表…

该吃吃&#xff0c;该喝喝&#xff0c;遇事儿别往心上隔&#x1f60e; 文章目录一、虚拟地址空间1.虚拟地址的引出&#xff08;看不到物理地址&#xff0c;只能看看虚拟地址喽&#xff09;2.虚拟地址空间布局&#xff08;五个段&#xff09;3.感性理解一下虚拟地址空间&#xf…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023