kafka学习(二)

创建kafka生产者

要往kafka写入消息,首先要创建一个生产者对象,并设置一些熟悉。kafka生产者有3个必选的属性。

1.bootstrap.servers 该属性指定broker的地址清单,地址的格式为host:port 一般建议至少要提供两个broker的信息,一旦其中一个宏机,生产者仍然能够连接到集群中。

2.key.serialier broker希望收到的消息的键和值都是字节数组,生产者接口允许使用参数化类型,因此可以把java对象作为键和值发送给broker。

3.value.serializer 与key.serializer一样,value.serializer指定的类将值序列化。

构建kafkaProduce对象步骤

1.构建一个properties对象

2.将上诉3个参数按照格式put到properties中。

3.new kafkaProduce对象,将properties放入kafkaproduce构造函数

kafka生产者发送的3种方式

1.发送并忘记(fire-and-forget),我们把消息发送给服务器,但不关心它是否正常到达,大多数情况会正常到达,因为kafka是高可用的,而且生产者会自动尝试重发。不过这种方式会存在丢失一些消息。

2.同步发送,我们使用send()方法发送消息,它会返回一个futured对象,调用get()方法进行等待,就可以知道消息时候发送成功。

2.异步发送,我们调用send(),并指定一个回调函数,服务器在返回响应时调用该函数。

发送消息到kafka过程

最简单的一个代码示例

ProducerRecord<String,String> record = new ProducerRecord<>("test","test","test")

try{

producer.send(record);

}catch(Exception e){

e.printStackTrace()

}

1.生产者的send()方法将producerRecord对象作为参数,所以我们先创建一个producerRecord对象。

2.执行send()方法先是放进缓存区,然后使用单独的线程发送到服务器端,send()会返回一个包含RecordMetadata的Future对象。

3.我们忽略发送至服务器的异常,但是在发送之前,生产者还是有可能发生其他异常。这些异常有可能是serializationException(说明序列化消息失败),bufferExhaustedException或TimeoutException(说明缓存区已经满了) ,又或者是InterrupException(说明发送线程被中断)

同步发送消息

1.首先producer.send()方法先返回一个future对象,然后调用future.get()方法来等待kafka响应。如果没有发生异常就会得到一个recordMetadata对象,可以用它来获取消息的偏移量。

2.如果在发送数据之前或者已经超过重发的次数,那么就会抛异常。

kafkaProducer一般会发送两类错误。其中一种就是重试错误,这类错误可以通过重发消息来解决,对于连接错误,可以通过再次连接来解决,无主(no leader) 错误则可以通过重新分区选举来解决

还有一类就是消息太大比kafka配置还大。

异步发送消息

1.为了使用回调,需要一个实现了producer.callback接口

2.如果kafka返回一个错误,onCompletion方法会抛出一个非空异常

生产者的配置

1.acks 参数指定了必须有多少个分区副本收到消息,生产者才会认为消息写入是成功的。这个参数对消息丢失的可能性有重要影响,如果acks=0 生产者在成功写入消息之前不会等待任何来自服务器的响应。也就是说,如果当中出现了问题,导致服务器没有收到消息,那么生产者就无从得知,消息也就丢了。这样设置可以达到很高的吞吐量。

acks=1 只要集群的首领节点收到消息,生产者就会收到一个来自服务成功响应。如果消息无法到达首领节点,生产者就会收到一个错误的响应,为了避免数据丢失,生产者会重发消息。

acks=all 只要到所以参与复制的节点全部收到消息时,生产者才会收到一个来自服务器成功的消息,这种模式是最安全的。

2.buffer.memory 设置生产者内存缓冲器的大小,生产者用它缓冲要发送到服务器的消息。如果空间不足一般send()阻塞或者抛异常。

3.compressionType 默认情况下,消息发送时不会被压缩。该参数可以设置为snappy,gzip,lz4 它指定消息发送给broker之前使用哪一种压缩算法进行压缩。

4.retries 决定生产者可以重发消息的次数,如果达到重发次数,生产放弃发送并返回错误。

5.batch.size 当有多个消息需要被发送到同一个分区时,生产者会把它们放在同一个批次里。该参数指定一个批次可以使用的内存大小,安装字节数计算而不是消息个数。

6.linger.ms 该参数指定了生产者在发送批次之前等待更多消息加入批次的时间,批次会在linger.ms达到上限时把批次发送出去。

7.client.id 该参数是任意字符串,服务器会用它来识别消息的来源。

8.max.in.fight.requests.per.connection 该参数指定了生产者在收到服务器响应之前可以发送多少个消息。它的值越高,占用的内存越大。

9.timeout.ms request.timeout.ms和metadata.fetch.timeout.ms

request.timeout.ms 指定了生产者在发送数据时等待服务器返回响应的时间

metadata.fetch.timeout.ms 指定了生产在获取元数据时等待服务器返回响应的时间,如果超时了那么生产者要么重试发送数据,要么返回错误。

timeout.ms 指定了broker等待同步副本返回消息确认的时间,与acks的配置相匹配。如果指定时间没有收到副本的确认,那么broker就会返回一个错误。

10.max.request.ms 该参数指定了调用send()方法或使用partitionsFor()方法获取元数据时生产者的阻塞时间。

11.max.request.size 该参数用于控制生产者发送的请求大小。

12.receive.buffer.bytes和send.buffer.bytes

这两个参数分别指定了TCP socket接受和发送数据包的缓存区大小。

提一下: kafka可以保证一个分区的消息时有序的。

序列化器

创建一个生产者对象必须指定序列化器,kafka提供了整型和字符数组序列化器,不过它们还不足满足大部分需求。

自定义序列化器

一般实现serializer接口来实现序列化,但是这一般都用序列化框架,比如,Avro,thrift,protobuf

kafka使用apache avro序列化

apache avro 是一种与编程语言无关的序列化格式,arvo目的是提供一种共享数据文件的方式。

avro数据通过与语言无关的schema来定义。schema通过JSON来描述,数据被序列化成二进制文件或JSON文件,不过一般会只用二进制文件。

在kafka中使用Avro

生产者配置的两个配置参数key.serializer 和value.serializer 在properties中将这两个参数设置为key ,vaue值为io.conflunt.kafka.serializers.kafkaAvroSerializer,还要设置schema.registry.url参数,值为scheme的存储位置,其他和上述生产者对象一样。

分区

我们知道producerRecord对象包含了目标主题,键和值。kafka消息是一个个键值对,键可以设置为null。

键有两个作用

1.可以作为消息的附加消息

2.可以用来决定消息该被写到主题的那个分区。

如果键为null,并且使用默认的分区器,那么记录将被随机地发送到主题内各个可用的分区上。分区器使用轮询算法将消息均衡地分布到各个分区上。

实现自定义分区

这里实现Partition接口来自定义分区 这个接口包含configure partition和close方法。规则由自己去定。

原文地址:https://www.cnblogs.com/Seeasunnyday/p/9230584.html

时间: 2024-11-09 04:36:14

kafka学习(二)的相关文章

kafka学习(二)-zookeeper集群搭建

zookeeper概念 ZooKeeper是一个分布式的,开放源码的分布式应用程序协调服务,它包含一个简单的原语集,分布式应用程序可以基于它实现同步服务,配置维护和命名 服务等.Zookeeper是hadoop的一个子项目,其发展历程无需赘述.在分布式应用中,由于工程师不能很好地使用锁机制,以及基于消息的协调机制 不适合在某些应用中使用,因此需要有一种可靠的.可扩展的.分布式的.可配置的协调机制来统一系统的状态.Zookeeper的目的就在于此. 1.角色 Zookeeper中的角色主要有以下三

kafka学习之路(二)——提高

kafka学习之路(二)--提高 消息发送流程 因为Kafka内在就是分布式的,一个Kafka集群通常包括多个代理.为了均衡负载,将话题分成多个分区,每个代理存储一或多个分区.多个生产者和消费者能够同时生产和获取消息.     过程: 1.Producer根据指定的partition方法(round-robin.hash等),将消息发布到指定topic的partition里面 2.kafka集群接收到Producer发过来的消息后,将其持久化到硬盘,并保留消息指定时长(可配置),而不关注消息是否

kafka 学习(二)

kafka 学习(二) 一.Kafka的架构 如上图所示,一个典型的Kafka集群中包含若干Producer(可以是web前端产生的Page View,或者是服务器日志,系统CPU.Memory等),若干broker(Kafka支持水平扩展,一般broker数量越多,集群吞吐率越高),若干Consumer Group,以及一个Zookeeper集群.Kafka通过Zookeeper管理集群配置,选举leader,以及在Consumer Group发生变化时进行rebalance.Producer

kafka学习笔记:知识点整理

一.为什么需要消息系统 1.解耦: 允许你独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束. 2.冗余: 消息队列把数据进行持久化直到它们已经被完全处理,通过这一方式规避了数据丢失风险.许多消息队列所采用的"插入-获取-删除"范式中,在把一个消息从队列中删除之前,需要你的处理系统明确的指出该消息已经被处理完毕,从而确保你的数据被安全的保存直到你使用完毕. 3.扩展性: 因为消息队列解耦了你的处理过程,所以增大消息入队和处理的频率是很容易的,只要另外增加处理过程即可. 4.

[Big Data - Kafka] kafka学习笔记:知识点整理

一.为什么需要消息系统 1.解耦: 允许你独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束. 2.冗余: 消息队列把数据进行持久化直到它们已经被完全处理,通过这一方式规避了数据丢失风险.许多消息队列所采用的"插入-获取-删除"范式中,在把一个消息从队列中删除之前,需要你的处理系统明确的指出该消息已经被处理完毕,从而确保你的数据被安全的保存直到你使用完毕. 3.扩展性: 因为消息队列解耦了你的处理过程,所以增大消息入队和处理的频率是很容易的,只要另外增加处理过程即可. 4.

kafka学习之路

https://www.cnblogs.com/qingyunzong/p/9004509.html 目录 一.简介 1.1 概述 1.2 消息系统介绍 1.3 点对点消息传递模式 1.4 发布-订阅消息传递模式 二.Kafka的优点 2.1 解耦 2.2 冗余(副本) 2.3 扩展性 2.4 灵活性&峰值处理能力 2.5 可恢复性 2.6 顺序保证 2.7 缓冲 2.8 异步通信 三.常用Message Queue对比 3.1 RabbitMQ 3.2 Redis 3.3 ZeroMQ 3.4

[Python 学习] 二、在Linux平台上使用Python

这一节,主要介绍在Linux平台上如何使用Python 1. Python安装. 现在大部分的发行版本都是自带Python的,所以可以不用安装.如果要安装的话,可以使用对应的系统安装指令. Fedora系统:先以root登入,运行 yum install python Ubuntu系统:在root组的用户, 运行 sudo apt-get install python 2. 使用的Python的脚本 Linux是一个以文件为单位的系统,那么我们使用的Python是哪一个文件呢? 这个可以通过指令

OpenCV for Python 学习 (二 事件与回调函数)

今天主要看了OpenCV中的事件以及回调函数,这么说可能不准确,主要是下面这两个函数(OpenCV中还有很多这些函数,可以在 http://docs.opencv.org/trunk/modules/highgui/doc/user_interface.html 找到,就不一一列举了),然后自己做了一个简单的绘图程序 函数如下: cv2.setMouseCallback(windowName, onMouse[, param]) cv2.createTrackbar(trackbarName,

kafka入门二:Kafka的设计思想、理念

本节主要从整体角度介绍Kafka的设计思想,其中的每个理念都可以深入研究,以后我可能会发专题文章做深入介绍,在这里只做较概括的描述以便大家更好的理解Kafka的独特之处.本节主要涉及到如下主要内容: Kafka设计基本思想 Kafka中的数据压缩 Kafka消息转运过程中的可靠性 Kafka集群镜像复制 Kafka 备份机制 一.kafka由来 由于对JMS日常管理的过度开支和传统JMS可扩展性方面的局限,LinkedIn(www.linkedin.com)开发了Kafka以满足他们对实时数据流