生产者

发送消息：

Properties props = new Properties();
//kafka 集群，broker-list
props.put("bootstrap.servers", "172.24.211.140:9092");
props.put("key.serializer",
        "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer",
        "org.apache.kafka.common.serialization.StringSerializer");
Producer<String,  String> producer  =  new
        KafkaProducer<>(props);
for (int i = 0; i < 10; i++) {
    var record =
            new ProducerRecord<>("test", "Precision Products",
                    "France");
    producer.send(record, new Callback() {
        @Override
        public void onCompletion(RecordMetadata metadata, Exception exception) {
            System.out.println(metadata);
        }
    });

}
producer.close();

配置

• acks
  • 定了必须要有多少个分区副本收到消息，生产者才会认为消息写入是成功的
  • acks=0 ，生产者在成功写入消息之前不会等待任何来自服务器的响应 当 broker 故障时有可能 丢失数据
  • acks=1 ，只要集群的首领节点收到消息，生产者就会收到一个来自服务器的成功响应 如果在 follower同步成功之前 leader 故障，那么将会丢失数据
  • acks=all ，只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应 如果在 follower 同步完成后，broker 发送 ack 之前，leader 发生故障，那么会造成 数据重复
• buffer.memory
  • 设置生产者内存缓冲区的大小
• compression.type
  • 设置消息压缩算法
• retries
  • 决定了生产者可以重发消息的次数，如果达到这个次数，生产者会放弃重试并返回错误
• batch.size
  • 指定了一个批次可以使用的内存大小
• linger.ms
  • KafkaProducer 会在批次填满或 linger.ms 达到上限时把批次发送出去。默认情况下，只要有可用的线程，生产者就会把消息发送出去
• client.id
• max.in.flight.requests.per.connection
  • 指定了生产者在收到服务器响应之前可以发送多少个消息
• timeout.ms、request.timeout.ms 和 metadata.fetch.timeout.ms
• max.block.ms
  • 调用send时最长的阻塞时间
• max.request.siz
• receive.buffer.bytes 和 send.buffer.bytes
  • 分别指定了 TCP socket 接收和发送数据包的缓冲区大小

顺序保证

• 将max.in.flight.requests.per.connection设置为1

保证顺序的方法就是：

1. 每个主题只分为一个区
2. 每次发送的消息发送到同一个分区

序列化器

• 自定义序列化器：实现Serializer接口
  • 不推荐使用
• 其他序列化
  • avro：一种将shcema嵌入在数据里的序列化方式

分区策略

分区的原因：

• 方便扩展
• 提高并发

分区原则：

• 指明 partition 的情况下，直接将指明的值直接作为 partiton 值
• 没有指明 partition 值但有 key 的情况下，将 key 的 hash 值与 topic 的 partition数进行取余得到 partition 值
• 否则就是随机取一个值 然后再这个值的基础上进行轮询

自定义分区器：

实现Partitioner接口

数据可靠性保证

• Exactly Once

将服务器的 ACK 级别设置为-1，可以保证 Producer 到 Server 之间不会丢失数据，即 AtLeast Once 语义

At Least Once + 幂等性 = Exactly Once

另外一个需要注意的是要注册发送回调，发生不成功，需要客户端进行重试

生产者幂等

生产者可以指定 enable.idempotence 来实现生产幂等，其原理是 broker 引入了ProducerID和SequenceNumber，每个新的Producer初始化时，会被分配一个唯一的ProducerID，同时生产者为每条消息生成SequenceNumber，broker 端通过这两个字段来判断数据之前是否发送过，从而实现幂等

这种幂等判断只能保证某个主题的一个分区上不出现重复消息

事务

能保证多条消息原子性地写入到目标分区，同时也能保证 Consumer 只能看到事务成功提交的消息

producer.beginTransaction(); 
producer.send(record1); 
producer.send(record2); 
producer.commitTransaction();

是一种类似 2PC 的实现