首先是启动一个生产者
final String kafkazk="localhost:9092";
String topic="testAPI";
Properties properties = new Properties() {{
put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkazk);
put(ProducerConfig.ACKS_CONFIG, "all");
put(ProducerConfig.RETRIES_CONFIG, 0);
put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
put(ProducerConfig.LINGER_MS_CONFIG, 1);
put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
}};
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
for(int i=0;i<10000000;i++){
producer.send(new ProducerRecord<>(topic, UUID.randomUUID().toString(), String.valueOf(i)));
}
}
启动配置参数如下所示：
2018-06-19 17:34:23,636 - ProducerConfig values:
acks = all
batch.size = 16384
block.on.buffer.full = false
bootstrap.servers = [localhost:9092]
buffer.memory = 33554432
client.id =
compression.type = none
connections.max.idle.ms = 540000
interceptor.classes = null
key.serializer = class org.apache.kafka.common.serialization.StringSerializer
linger.ms = 1
max.block.ms = 60000
max.in.flight.requests.per.connection = 5
max.request.size = 1048576
metadata.fetch.timeout.ms = 60000
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.sample.window.ms = 30000
partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
receive.buffer.bytes = 32768
reconnect.backoff.ms = 50
request.timeout.ms = 30000
retries = 0
retry.backoff.ms = 100
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.mechanism = GSSAPI
security.protocol = PLAINTEXT
send.buffer.bytes = 131072
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
timeout.ms = 30000
value.serializer = class org.apache.kafka.common.serialization.StringSerializer
acks参数：
在考虑请求完成之前，生产者要求leader收到的确认数量，这将控制发送的记录的持久性。
acks=0如果设置为零，则生产者不会等待来自服务器的任何确认。该记录将被立即添加到套接字缓冲区并被视为已发送。在这种情况下，retries不能保证服务器已经收到记录，并且配置不会生效（因为客户端通常不会知道任何故障）。为每个记录返回的偏移量将始终设置为-1。
acks=1这意味着领导者会将记录写入其本地日志中，但会在未等待所有追随者完全确认的情况下作出响应。在这种情况下，如果领导者在承认记录后但在追随者复制之前立即失败，那么记录将会丢失。
acks=all这意味着领导者将等待全套的同步副本确认记录。这保证只要至少有一个同步副本保持活动状态，记录就不会丢失。这是最强有力的保证。这相当于acks = -1设置。
batch.size
只要有多个记录被发送到同一个分区，生产者就会尝试将记录一起分成更少的请求。这有助于客户端和服务器的性能。该配置以字节为单位控制默认的批量大小。不会尝试批量大于此大小的记录。发送给brokers的请求将包含多个批次，每个分区有一个可用于发送数据的分区。
小批量大小将使批次不太常见，并可能降低吞吐量（批量大小为零将完全禁用批次）。一个非常大的批量大小可能会更浪费一点使用内存，因为我们将始终为预期的额外记录分配指定批量大小的缓冲区。
block.on.buffer.full
当我们的内存缓冲区耗尽时，我们必须停止接受新记录（块）或抛出错误。默认情况下，此设置为false，生产者不再抛出BufferExhaustException，而是使用该max.block.ms值来阻塞，之后将抛出TimeoutException。将此属性设置为true会将其设置max.block.ms为Long.MAX_VALUE。此外，如果此属性设置为true，则参数metadata.fetch.timeout.ms不再有效。
此参数已被弃用，并将在未来版本中删除，应该使用参数max.block.ms。
bootstrap.servers
用于建立到Kafka集群的初始连接的主机/端口对列表。客户端将使用所有服务器，而不管在此指定哪些服务器用于引导 - 此列表仅影响用于发现全套服务器的初始主机。该列表应该以表格的形式出现host1:port1,host2:port2,...。由于这些服务器仅用于初始连接以发现完整的群集成员资格（可能会动态更改），因此此列表不需要包含全套服务器（但您可能需要多个服务器） 。
buffer.memory
生产者可用于缓冲等待发送到服务器的记录的总内存字节数。如果记录的发送速度比发送到服务器的速度快，那么生产者将会阻止max.block.ms它，然后它会抛出异常。
这个设置应该大致对应于生产者将使用的总内存，但不是硬性限制，因为不是所有生产者使用的内存都用于缓冲。一些额外的内存将用于压缩（如果启用了压缩功能）以及维护正在进行的请求。
client.id
发出请求时传递给服务器的id字符串。这样做的目的是通过允许将逻辑应用程序名称包含在服务器端请求日志中，从而能够跟踪ip / port之外的请求源，如果不手动指定，代码中会自动生成一个id。
compression.type
指定给定主题的最终压缩类型。该配置接受标准压缩编解码器（'gzip'，'snappy'，'lz4'）。它另外接受'未压缩'，这相当于没有压缩，这意味着保留制片人设置的原始压缩编解码器，也可以修改源码，自定义压缩类型。
connections.max.idle.ms
在此配置指定的毫秒数后关闭空闲连接。
interceptor.classes
用作拦截器的类的列表。通过实现ProducerInterceptor接口，您可以在生产者发布到Kafka集群之前拦截（并可能会改变）生产者收到的记录。默认情况下，没有拦截器，可自定义拦截器。
key.serializer
用于实现Serializer接口的密钥的串行器类。
linger.ms
生产者将在请求传输之间到达的任何记录归入单个批处理请求。通常情况下，这只会在记录到达速度快于发送时才发生。但是，在某些情况下，即使在中等负载下，客户端也可能希望减少请求的数量。此设置通过添加少量人工延迟来实现此目的 - 即不是立即发送记录，而是生产者将等待达到给定延迟以允许发送其他记录，以便发送可以一起批量发送。这可以被认为与TCP中的Nagle算法类似。这个设置给出了批量延迟的上限：一旦我们得到batch.size值得记录的分区，它将被立即发送而不管这个设置如何，但是如果我们为这个分区累积的字节数少于这个数字，我们将在指定的时间内“等待”，等待更多的记录出现。该设置默认为0（即无延迟）。linger.ms=5例如，设置可以减少发送请求的数量，但会对在无效负载中发送的记录添加高达5毫秒的延迟。
max.block.ms
配置控制了KafkaProducer.send()并将KafkaProducer.partitionsFor()被阻塞多长时间。由于缓冲区已满或元数据不可用，这些方法可能会被阻塞止。用户提供的序列化程序或分区程序中的阻塞将不计入此超时。
max.in.flight.requests.per.connection
在阻塞之前，客户端将在单个连接上发送的最大数量的未确认请求。请注意，如果此设置设置为大于1并且发送失败，则由于重试（即，如果重试已启用），可能会重新排序消息。
max.request.size
请求的最大大小（以字节为单位）。这实际上也是最大记录大小的上限。请注意，服务器在记录大小上有自己的上限，这可能与此不同。此设置将限制生产者在单个请求中发送的记录批次数，以避免发送大量请求。
metadata.fetch.timeout.ms
首次将数据发送到主题时，我们必须获取有关该主题的元数据，以了解哪些服务器托管主题的分区。此配置指定在将异常抛回到客户端之前，此获取成功的最长时间（以毫秒为单位）。
metadata.max.age.ms
以毫秒为单位的时间段之后，即使我们没有看到任何分区领导改变以主动发现任何新代理或分区，我们强制更新元数据。
metric*
以metric开头的是指标相关的，后面会讨论。
partitioner.class
实现Partitioner接口的分区器类。默认使用DefaultPartitioner来进行分区。
receive.buffer.bytes
读取数据时使用的TCP接收缓冲区（SO_RCVBUF）的大小。如果该值为-1，则将使用操作系统默认值。
reconnect.backoff.ms
尝试重新连接到给定主机之前等待的时间量。这避免了在紧密循环中重复连接到主机。该退避适用于消费者发送给代理的所有请求。
request.timeout.ms
该配置控制客户端等待请求响应的最长时间。如果在超时过去之前未收到响应，客户端将在必要时重新发送请求，或者如果重试耗尽，请求失败。
retries
生产者发送失败后的重试次数，默认0
retry.backoff.ms
尝试重试对给定主题分区的失败请求之前等待的时间量。这可以避免在某些故障情况下重复发送请求。
sasl与ssl
与kafka安全相关