kafka producer实例

2023-08-11 12:22:34

1. 定义要发送的消息User POJO

package lenmom.kafkaproducer;

public class User {

    public String name;

    public int age;

    public String address;

    public User(String name,int age,String address){

        this.name=name;

        this.age=age;

        this.address=address;

    }

    @Override

    public String toString() {

        return  "user:"+name +",age:"+age+",address:"+address;

    }

}

2. 定义序列化User POJO序列化器

package lenmom.kafkaproducer;

import org.apache.kafka.common.serialization.Serializer;

import org.codehaus.jackson.map.ObjectMapper;

import java.util.Map;

public class UserSerializer implements Serializer {

    private ObjectMapper objectMapper;

    public void configure(Map map, boolean b) {

        this.objectMapper = new ObjectMapper();

    }

    public byte[] serialize(String topic, Object object) {

        byte[] result = null;

        try {

            result = objectMapper.writeValueAsString(object).getBytes("utf-8");

        } catch (Exception e) {

              e.printStackTrace();

        }

        return result;

    }

    public void close() {

    }

}

此序列化器使用了org.codehaus.jackson.jackson-mapper-asl.jar来进行序列化为jason

3. 定义partitioner

package lenmom.kafkaproducer;

import org.apache.kafka.clients.producer.Partitioner;

import org.apache.kafka.common.Cluster;

import org.apache.kafka.common.PartitionInfo;

import java.util.List;

import java.util.Map;

import java.util.Random;

public class AgePartitioner implements Partitioner {

    private Random random;

    public int partition(String topic, Object keyObj, byte[] keyBytes, Object value,

                         byte[] valueBytes, Cluster cluster) {

        System.out.println("key:"+keyObj.toString()+",value: "+value.toString());

        Integer age = ((User)value).age;

        List<PartitionInfo> partitions = cluster.availablePartitionsForTopic(topic);

        int partitionCount = ((List) partitions).size();

        int auditPartition = partitionCount - 1;

        // 如果年龄大于20岁的，分发到kafka的最后一个分区，否则随机发送到前面几个分区中去

        return age< 20 ? random.nextInt(partitionCount - 1) : auditPartition;

    }

    public void close() {

    }

    public void configure(Map<String, ?> map) {

        this.random = new Random();

    }

}

4. 定义发送的消息统计拦截器

package lenmom.kafkaproducer;

import org.apache.kafka.clients.producer.ProducerInterceptor;

import org.apache.kafka.clients.producer.ProducerRecord;

import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

public class CounterIntecepter implements ProducerInterceptor {

    int count = 0;

    int failedCount = 0;

    /*

     * This method will be called before the message send to the broker

     * */

    public ProducerRecord onSend(ProducerRecord producerRecord) {

        count += 1;

        System.out.println("preparing to send value: " + producerRecord.value().toString());

        return producerRecord;

    }

    /*

     * This method would be called when the message was handled by the borker after the broker call back.

     */

    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {

        if (e != null) {

            failedCount++;

        }

    }

    public void close() {

        System.out.println("total: " + this.count + ",failed: " + this.failedCount);

    }

    public void configure(Map<String, ?> map) {

    }

}

5. Producer主类

package lenmom.kafkaproducer;

import org.apache.kafka.clients.producer.*;

import java.util.ArrayList;

import java.util.List;

import java.util.Properties;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.TimeUnit;

public class ProducerDemo {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        Properties props = new Properties();

        props.put("bootstrap.servers", "192.168.1.131:9092,192.168.1.193:9092,192.168.1.194:9092");

        props.put("acks", "all"); //

        props.put("retries", Integer.MAX_VALUE);

        props.put("batch.size", 323840);

        props.put("linger.ms", 100);

        props.put("buffer.memory", 33554432);

        props.put("max.block.ms", 3000);

        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

        props.put(ProducerConfig.RETRIES_CONFIG, 3);

        props.put("max.in.flight.requests.per.connection",1);

        testCustomSerializerProducer(props);

    }

    private static void testCustomSerializerProducer(Properties props) throws ExecutionException, InterruptedException {

        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "lenmom.kafkaproducer.UserSerializer");

        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        //设置消息拦截器

        List<String> interceptors= new ArrayList<String>();

        interceptors.add("lenmom.kafkaproducer.CounterIntecepter");

        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,interceptors);

        //设置自定义分区 
        props.put("partitioner.class", "lenmom.kafkaproducer.AgePartitioner");

        final Producer<String, User> producer = new KafkaProducer<String, User>(props);

        for (int i = 0; i < 100; i++) {

            User user = new User("lenmom" + i, i, "address" + i);

            ProducerRecord<String, User> record= new ProducerRecord<String, User>("lenmom", user);

            producer.send(record, new Callback() {

                public void onCompletion(RecordMetadata recordMetadata, Exception e) {

                    if (e != null) {

                        producer.close(0,TimeUnit.MILLISECONDS);

                    } else {

                        System.out.println("send success");

                    }

                }

            });

        }

        producer.close();

    }

}

注意: KafkaProducer是线程安全的，也就是说可以在多个线程*享一个KafkaProducer实例.

6. pom.xml文件

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0"

         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0</modelVersion>

    <groupId>lenmom</groupId>

    <artifactId>kafkaproducer</artifactId>

    <version>1.0-SNAPSHOT</version>

    <dependencies>

        <dependency>

            <groupId>org.apache.kafka</groupId>

            <artifactId>kafka-clients</artifactId>

            <version>0.10.1.0</version>

        </dependency>

        <dependency>

            <groupId>org.codehaus.jackson</groupId>

            <artifactId>jackson-mapper-asl</artifactId>

            <version>1.9.13</version>

        </dependency>

    </dependencies>

</project>

运行效果：

Producer执行效果

启动consumer:

$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.101: --topic lenmom --from-beginning

接收到的消息：

7. 配置参数说明

7.1 消息无丢失配置

7.1.1 Producer配置

7.1.1.1

#使得内存缓冲区被填满时producer 处于阻塞状态并停止接收新的消息而不是抛出异常；否则producer 生产速度过快会耗尽缓冲区
block.on.buffer.full ＝true

7.1.1.2
#设置acks 为all 很容易理解，即必须要等到所有fo llower 都响应了发送消息才能认为提交成功，这是pro ducer 端最强程度的持久化保证。
acks = all or 1

7.1.1.3

#重试次数设置为Integer.MAX_VALUE,设置成MAX_VALUE 纵然有些极端，但其实想表达的是producer 要开启无限重试。
#用户不必担心pro ducer 会重试那些肯定无法恢复的错误，当前producer 只会重试那些可恢复的异常
#情况，所以放心地设置一个比较大的值通常能很好地保证消息不丢失。
retries = Integer.MAX_VALUE

7.1.1.4
#设置该参数为1 主要是为工防止topic 同分区下的消息乱序问题。这个参数的实际效果其
#实限制了producer 在单个broker 连接上能够发送的未响应请求的数量。因此，如果设置成l ,
#则producer 在某个broker 发送响应之前将无法再给该broker 发送PRODUCE 请求
max.in.flight.requests.per.connection= 1

7.1.1.5

使用带有回调机制的send
不要使用KafkaPro ducer 中单参数的send 方法，因为该se nd 调用仅仅是把消息发出而不会理会消息发送的结果。如果消息发送失败，该方法不会得到任何通知，故可能造成数据的丢失。
实际环境中一定要使用带回调机制的send 版本，即KafkaProducer.send(record, callback）。

7.1.1.6

Callback 逻辑中显式立即关闭producer
在Callback 的失败处理逻辑中显式调用KafkaProducer .close(O ）。这样做的目的是为了处理消息的乱序问题。
若不使用close(O），默认情况下producer 会被允许将未完成的消息发送出去，这样就有可能造成消息乱序。

7.1.2 Broker消息无丢失设置

7.1.2.1

#broker关闭unclean leader 选举，即不允许非ISR 中的副本被选举为leader ，从而避免broker 端因日志水位截断而造成的消息丢失。
unclean.leader.election .enable= false

7.1.2.2

#broker 设置成3 主要是参考了Hadoop 及业界通用的三备份原则，其实这里想强调的是一定要使用多个副本来保存分区的消息。
replication.factor = 3

7.1.2.3

#broker 用于控制某条消息至少被写入到ISR 中的多少个副本才算成功，设置成大于1 是为了提升producer 端发送语义的持久性。
#记住只有在producer 端acks 被设置成all 或一l 时，这个参数才有意义。在实际使用时，不要使用默认值。
min.insync .replicas = 2

7.1.2.4

#btoker 若两者相等，那么只要有一个副本挂掉，分区就无法正常工作，虽然有很高的持久性但可用性被极大地降低了。推荐配置成replication.factor= min.insyn.replicas + l
replication.factor > min.insync.replicas

7.1.2.5

#broker 禁用自动提交
enable.auto.commit= false

7.2 压缩参数设置

producer使用压缩算法，可以提高kafka的吞吐量，目前支持的压缩算法有GZIP 、Snappy、LZ4和Facebook 公司于2016 年8 月底开源了新的压缩算法Zstandard。

配置方法为，在给producer的配置构造函数中加入对应的配置项，如：

props . put (” compressiont.typ”,” snappy");

／／或者

props.put(ProducerConfig . COMPRESSION TYPE CONFIG ,”snappy”);

使用压缩算法，对集群的CPU有要求，建议的生产集群配置为：

linux
CPU 24 核
内存32GB 。
磁盘lTB 7200 转SAS 盘两块
带宽lGb/s
ulimit -n 1000000 。
Socket Buffer 至少64KB一一适用于跨机房网络传输。

码农公寓

相关文章