这次的笔记主要记录一下kafka的生产者的使用和一些重要的参数。

文中主要截图均来自kafka权威指南

主要涉及到两个类KafkaProducer和ProducerRecord.

总览

生产者的主要架构如下：

首先创建了一个ProducerRecord

进行序列化 kv变为ByteArray

进入Partitioner 如果之前指定了分区那这一步什么都不会做

接着将Record放入要被发送到的同样的topic和分区的batch中

另一个单独的thread会进行发送操作

发送成功会返回RecordMetadata(包含topic 分区偏移量)

失败的话会返回错误 producer会重试几次(retries配置)直到成功或者放弃返回错误

构造Kafka Producer

示例代码如下:

Properties properties = new Properties();

properties.put("bootstrap.servers", "192.168.56.101:9092");

properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

properties.put("metadata.fetch.timeout.ms", 5000);

KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

ProducerRecord<String, String> record = new ProducerRecord<String, String>("CustomerCountry",

		"Precision Products", "France");

Future<RecordMetadata> f = producer.send(record);

返回的是一个Future,有以下处理方式:

不在乎异常可以不处理
直接get拿到结果同步的获取方式

另外send方法也支持传入回调的方式进行异步处理

bootstrap.servers

指定了要连接到的broker,可以用,分隔写入多个，连接broker cluster时不需要写入全部的broker地址，broker本身会同步其他broker的信息，但最少放两个以免写一个对应的broker挂了。

在实际使用中，可能会出现org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.这个错误，请求本身是发送成功的，检查zk topic也建立了，但就是获取不到metadata。

这是因为填入的broker服务器和kafka在zk上注册的host不一致导致的(在zk上可以通过get /brokers/ids/{id}查看)，可以通过修改kafka的server.properties的advertised.host.name和advertised.port改正。

但是要注意hosts文件会影响，比如配置了自己ip到一个域名上导致失败，注意代码在运行时log打出的实际的值。

例如电脑上装了银联控件，结果把localhost映射到了一个奇怪的地址：

例子中已经改为了ip。

另外实验环境也迁移至了虚拟机上，上一节用bash on windows搭建好的环境不知道为什么无论怎么改都获取不到metadata

key.serialization

用来指定key的序列化器(下面会提到)

配置Producers

文档：http://kafka.apache.org/documentation.html#producerconfigs

因为用的是0.9.0.1，一些参数在最新版已经被弃用了，自己使用多注意下。

这些的默认配置在ProducerConfig的static块中定义：

static {

    CONFIG = new ConfigDef().define(BOOTSTRAP_SERVERS_CONFIG, Type.LIST, Importance.HIGH, CommonClientConfigs.BOOSTRAP_SERVERS_DOC)

                            .define(BUFFER_MEMORY_CONFIG, Type.LONG, 32 * 1024 * 1024L, atLeast(0L), Importance.HIGH, BUFFER_MEMORY_DOC)

                            .define(RETRIES_CONFIG, Type.INT, 0, between(0, Integer.MAX_VALUE), Importance.HIGH, RETRIES_DOC)

                            .define(ACKS_CONFIG,

                                    Type.STRING,

                                    "1",

                                    in("all", "-1", "0", "1"),

                                    Importance.HIGH,

                                    ACKS_DOC)

    ... ...

这里列举一下比较重要的参数：

ACKS

指定一次写成功需要多少个分区分片必须收到record

0

不会等待broker的返回不会知道是否成功高吞吐
1

接受leader的比0安全但也会有还来不及复制leader就挂掉等导致失败的情况
all

所有的复本都接收到消息并回应最安全的模式但相对的延迟就比较惨了

默认值为1

BUFFER.MEMORY

缓存等待发送的消息

如果超过了会在抛错前等待max.block.ms秒

默认32M

COMPRESSION.TYPE

默认是none 不压缩

可以是snappy gzip lz4

snappy用较少的cpu完成还行的压缩性能和网络带宽的权衡

gzip是使用更多的cpu和时间但是压缩更好网络带宽是主要限制时使用

RETRIES

重试次数默认是0

默认间隔100ms (通过retry.backoff.ms控制)

不是所有的错误都会重试(RetriableException)

不要在业务里做重试

BATCH.SIZE

多个record被发送到同一个分区的时候 producer会进行批处理

内存默认16k

满了就发但不意味着只有满了才会发(见下一个参数)

设太大不会导致他的延迟只不过会用更多的内存

设太小就惨了会导致过于频繁

LINGER.MS

在发送当前batch时等多久额外的信息

默认是0

producer会在batch满了或者这个时间到了发送默认是0就意味着即时发送

大于0会导致延迟但是增加吞吐量

CLIENT.ID

任何字符串自己定义用来打log和metrics

MAX.IN.FLIGHT.REQUESTS.PER.CONNECTION

可以设置可以发送多少个请求不等回应也就是允许同时有多少条消息在空中飞

默认是5

设大了会增加内存使用但是提高吞吐

设过大了会导致降低吞吐因为batch变得低效了

设置成1可以保证消息即使是在重试下也有序(下面顺序保证中提到)

TIMEOUT.MS, REQUEST.TIMEOUT.MS, METADATA.FETCH.TIMEOUT.MS

第1和第3在0.11.0.1版本中被移除了。。可以参考旧文档

第2 从服务端等待多久回应(默认30s)

MAX.BLOCK.MS

在send()和显式通过partitionsFor()请求元数据中间等待的时间

在send buffer满了以及metadata没返回阻塞

MAX.REQUEST.SIZE

一次请求的数据大小限制(限制最大的消息以及批处理的大小)

默认1M 意味着可以同时发1000条1K的消息或者发一条1M的消息

注意上一节broker里也有自己的配置最好两个配置相匹配

RECEIVE.BUFFER.BYTES SEND.BUFFER.BYTES

前者默认32K

后者默认128K

TCP发送和接受的buffer 用于socket写入和读取数据

设置为-1是用OS默认的设置

不同数据中心间可以适当提高点这个值因为网络环境会更差

顺序保证

kafka保证在一个分区里消息的顺序

上述可以发现如果设置了max.in.flights.request.pre.connection大于1(默认5)

并且retries大于0(默认0)那就完蛋了

可能写给同一个分区第一条消息写入失败进入retry

但第二条写入成功

建议设置per.connection为1 但这会严重限制吞吐只用于顺序很重要的场合

序列化

自定义可以自己实现org.apache.kafka.common.serialization.Serializer接口。

使用Apache Avro

优点:

语言中立的序列化格式
兼容性

用于写和读的schema必须兼容

反序列化器必须能够访问用于写数据的schema

保存每个record的schema开销很大(通常是两倍于record)

可以将schema保存在一个注册器中管理只需要在消息中写入对应的id

Schema Registry - 有很多开源的选项可以使用confluent platform

文档: http://docs.confluent.io/current/schema-registry/docs/index.html

这个可以单独记一个笔记，等实践后再做整理。

分区

如果key为null并且默认的分区器被使用会用round-robin来平衡

有key 默认的分区器会hash key(自带的hash算法) 因为要保证同一个key总是写入同样的分区他会使用所有的分区(不管是否可用) 写到不可用的会报错

如果分区数改变了映射关系就会改变同样的key在之后可能会到其他分区中

所以预先确定分区的数量在这种i情况下比较重要

巴特西

kafka学习笔记2:生产者

总览