kafka生产者代码

1、maven包

		
            org.apache.kafka
            kafka-clients
            2.4.1
        

2、生产者代码

package cn.itcast.kafka;


import org.apache.kafka.clients.Metadata;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;


public class KafkaProducerTest {


    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //1、创建用于连接kafka的properties配置
        Properties conf = new Properties();
        //kafka服务器地址
        conf.put("bootstrap.servers", "192.168.30.1:9021");
        //表示当生产者数据到kafka中，kafka中会以什么样的策返回
        conf.put("acks", "all");
        //kafka中的消息是以key、value键值对储存的，而且生产者生产的消息是需要在网络上传导的，这里指定的是StringSerializer方式，也就是字符串方式发送
        conf.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //同上
        conf.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        //2、创建一个生产者对象kafkaProducer
        KafkaProducer kafkaProducer = new KafkaProducer<>(conf);

        //3、发送1-100的消息到指定的topic
        for (int i = 0; i < 100; i++) {
            //构建一条消息，直接new ProducerRecord
            ProducerRecord producerRecord = new ProducerRecord<>("test", null, i + "");


            //使用同步等待的方式发送消息
            


            
            kafkaProducer.send(producerRecord, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    //1、判断消息是否发送成功
                    if(e == null){
                        //发送成功
                        //主题
                        String topic = recordMetadata.topic();
                        //分区
                        int partition = recordMetadata.partition();
                        //偏移量
                        long offset = recordMetadata.offset();
                        System.out.println("topic:" + topic + "分区id:" + partition + "偏移量:" + offset);
                    }else {
                        //发送出现错误
                        System.out.println("生产消息出现异常");
                        //打印异常消息
                        System.out.println(e.getMessage());
                        //打印调用栈
                        System.out.println(e.getStackTrace());
                    }
                }
            });
        }
        //4、关闭生产者
        kafkaProducer.close();
    }
}