【kafka】Kafka的JavaAPI操作（Streams API开发 生产者 开发者）

目录

创建maven工程并添加jar包

生产者代码

生产者代码

自定义分区条件

生产者生产数据到kafka集群，数据到分区的方式

消费者代码

自动提交offset

手动提交offset

消费完每个分区之后手动提交offset

指定分区数据进行消费

自动提交offset和手动提交offset的区别：

Consumer消费数据的流程：

Streams API开发

创建topic testt01 testt02：

开发StreamAPI

生产数据

消费数据

• 创建maven工程并添加jar包

<dependencies>
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>1.0.0</version>
</dependency>   
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <version>1.0.0</version>
</dependency>

</dependencies>

<build>
    <plugins>
        <!-- java编译插件 -->
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.2</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
                <encoding>UTF-8</encoding>
            </configuration>
        </plugin>
    </plugins>
</build>

• 生产者代码

生产者代码

 //配置kafka集群环境（设置）
Properties props = new Properties();
//kafka服务器地址
props.put("bootstrap.servers", "node01:9092,node02:9092,node03:9092");
//消息确认机制
props.put("acks", "all");
//重试机制
props.put("retries", 0);
//批量发送的大小
props.put("batch.size", 16384);
//消息延迟
props.put("linger.ms", 1);
//批量的缓冲区大小
props.put("buffer.memory", 33554432);
//kafka   key 和value的序列化
props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");

//自定义分区条件
//props.put("partitioner.class", "Producer.ProducerPartition");
        
//2、实例一个生产者对象
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(props);

for (int i = 0; i < 9; i++) {
//3、通过生产者对象将数据发送到kafka集群
       //1、没有指定分区编号，没有指定key,时采用轮询方式存户数据
       ProducerRecord producerRecord = new ProducerRecord("18BD12","bbbb___"+i);
       
       //2、没有指定分区编号，指定key时，数据分发策略为对key求取hash值，这个值与分区数量取余，于数就是分区编号。
       //ProducerRecord producerRecord = new ProducerRecord("18BD12","test","aaaa___"+i);
            
       //3、指定分区编号，所有数据输入到指定的分区内
       //ProducerRecord producerRecord = new ProducerRecord("18BD12",1,"test","aaaa___"+i);

       //4、定义分区策略
       //ProducerRecord producerRecord = new ProducerRecord("18BD12","test","aaaa___"+i);

       kafkaProducer.send(producerRecord);
}
//4、关闭成产者
kafkaProducer.close();

自定义分区条件

public class ProducerPartition implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] key