Kafka生產者-向Kafka中寫入數據

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

import java.util.Properties;

/**
 * @Description:
 */
public class KafkaProducerTest {
    public static void ProducerTest(String data){
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.108.137:30019");//指定broker的地址清單
        props.put("acks", "all");//所有的partition都收到消息，生產者才會收到一個服務器的成功響應
        props.put("retries", 0);//生產者從服務器收到臨時性錯誤時，生產者重發消息的次數
        props.put("batch.size", 16384);//發送到同一個partition的消息會被先存儲在batch中，該參數指定一個batch可以使用的內存大小
        props.put("linger.ms", 1);//生產者在發送消息前等待linger.ms，為了等待更多的消息加入到batch中
        props.put("buffer.memory", 33554432);//設置生產者內緩存區域的大小，生產者用它緩沖要發送到服務器的消息
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");//必須是一個實現org.apache.kafka.common.serialization.Serializer接口的類，將key序列化成字節數組。注意：key.serializer必須被設置，即使消息中沒有指定key
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");//value序列化成字節數組
        Producer<String, String> producer = new KafkaProducer(props);
        producer.send(new ProducerRecord<>("test", data));//topic:key value
        producer.close();
    }
    public static void main(String[] args) {
        String data="{\"entitycode\":\"test\""}";
        ProducerTest(data);
    }


}