Kafka生产者
架构图:
必选属性:
bootstrap.servers: broker的地址清单(host:port)
key.serializer: 键的序列化器(ByteArraySerializer[这个只做很少的事情], StringSerializer, IntegerSerializer, 自定义序列化器)
value.serializer: 值的序列化器(同上)
创建Kafka生产者:
1. 新建一个Properties对象;
2. 因为我们打算把键和值定义成字符串类型, 所以使用内置的StringSerializer;
3. 在这里我们创建了一个新的生产者对象, 并为键和值设置了恰当的类型, 然后把Properties对象传给它。
private Properties kafkaProps = new Properties();
kafkaProps.put("bootstrap.servers", "broker1:9092, broker2:9092");
kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> producer = new kafkaProducer<String, String>(kafkaProps);
发送消息:
1.同步发送消息
ProducerRecord<String, String> record = new ProducerRecord<>("topic", "key", "value");
try {
producer.send(record).get();
} catch (Exception e) {
e.printStackTrace();
}
2.异步发送
private class DemoProducerCallback implements Callback {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e != null) {
e.printStackTrace();
}
}
}
ProducerRecord<String, String> record = new ProducerRecord<>("topic", "key", "value");
producer.send(record, new DemoProducerCallback());
可配置参数:
1.acks: 有多少个分区副本收到消息生产者才会认为消息写入是成功的;
2.buffer.memory: 设置生产者内存缓冲区的大小;
3.compression.type: 指定消息发送时使用哪一种压缩算法进行压缩(snappy, gzip, lz4);
4.retries: 生产者可以重发消息的次数;
5.batch.size: 同一批次发送到同一分区使用的内存大小;
6.linger.ms: 同批次等待时间;
7.client.id: 任意字符串, 识别消息的来源;
8.max.in.flight.requests.per.connection: 生产者在收到服务器的响应之前可以发送多少个消息;
9.timeout.ms, request.timeout.ms 和 metadata.fetch.timeout.ms:
timeout.ms: 等待同步副本返回消息确认的时间;
request.timeout.ms: 生产者在发送数据时等待服务器返回响应的时间;
metadata.fetch.timeout.ms: 生产者在获取元数据时等待服务器返回响应的时间;
10.max.block.ms: 获取元数据时的阻塞时间;
11.max.request.size: 生产者发送请求的大小;
12.receive.buffer.bytes 和 send.buffer.bytes: TCP socket 接收和发送数据宝的缓冲区大小;
序列化器:
主要实现 org.apache.kafka.common.serialization.Serializer 的 byte[] serialize(String topic, Customer data) 方法
分区器:
主要实现 org.apache.kafka.clients.producer.Partitioner 的 int partition(String topic, Object key,byte[] keyBytes, Object value,byte[] valueBytes, Cluster cluster) 方法
