java kafka 之生产者

小灰灰 2022-05-08 10:24 158阅读 0赞

**首先你得保证你的kafka服务已经起到起来**

## maven ##

<dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka_2.11</artifactId>
                <version>1.0.1</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
                <version>1.0.1</version>
            </dependency>

## java代码 ##

import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    import java.util.Date;
    import java.util.Properties;
    
    /** * @author lf * @描述 生产者 * @date 2018-08-30 */
    public class Producer { 
    
        public static Properties init(){ 
            Properties properties = new Properties();
            //kafka服务器的ip和端口
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092"); //"bootstrap.servers"
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); //"key.serializer"
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());//"value.serializer"
            //生产者发送的数据需要等待主分片和其副本都保存才发回确认消息
            //properties.put("acks", "all");
            /* Server完成 producer request 前需要确认的数量。 acks=0时，producer不会等待确认，直接添加到socket等待发送； 性能最高 acks=1时，等待leader写到local log就行； acks=all或acks=-1时，等待ISR中所有副本确认 最可靠 （注意：确认都是 broker 接收到消息放入内存就直接返回确认， 不是需要等待数据写入磁盘后才返回确认，这也是kafka快的原因） */
            return properties;
        }
    
         public static void main(String args[]) { 
    
             ProducerRecord<String , String> record = null;
             KafkaProducer<String,String> producer = null;
            Properties configs = init();
             producer = new KafkaProducer<String, String>(configs);
             for (int i = 0; i < 5; i++) { 
                 // 指定key,发送的消息都会落到一个分区上去，不指定则随机分配 part是topic名字
                 record = new ProducerRecord<String, String>("part", "我是生产者"+new Date());
                 //发送消息
                 producer.send(record);
                 //用底下的发送方式，可以知道数据发送成功
                 /*producer.send(record, new Callback() { public void onCompletion(RecordMetadata recordMetadata, Exception e) { if (null != e){ System.out.println("send error" + e.getMessage()); }else { System.out.println(String.format("offset:%s,partition:%s",recordMetadata.offset(),recordMetadata.partition())); } } });*/
             }
             producer.close();
          }
    }

此处只是最简单的生产消息，注意的话是，用带回调方法的生产模式，可以记录丢失的数据，如数据很重要，你可以保存起来，再次发送