package com.test.kafkaMessage;

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

import java.util.Properties;

public class SendOrderMessageTest {

    public KafkaProducer<String, String> producer;
    public final static String TOPIC = "strategy_pm_in";

    private SendOrderMessageTest() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "10.100.11.201:9092,10.100.11.202:9092,10.100.11.203:9092");//xxx服務(wù)器ip
        props.put("acks", "all");//所有follower都響應(yīng)了才認(rèn)為消息提交成功，即"committed"
        props.put("retries", 0);//retries = MAX 無限重試，直到你意識(shí)到出現(xiàn)了問題:)
        props.put("batch.size", 16384);//producer將試圖批處理消息記錄，以減少請(qǐng)求次數(shù).默認(rèn)的批量處理消息字節(jié)數(shù)
        //batch.size當(dāng)批量的數(shù)據(jù)大小達(dá)到設(shè)定值后，就會(huì)立即發(fā)送，不顧下面的linger.ms
        props.put("linger.ms", 1);//延遲1ms發(fā)送，這項(xiàng)設(shè)置將通過增加小的延遲來完成--即，不是立即發(fā)送一條記錄，producer將會(huì)等待給定的延遲時(shí)間以允許其他消息記錄發(fā)送，這些消息記錄可以批量處理
        props.put("buffer.memory", 33554432);//producer可以用來緩存數(shù)據(jù)的內(nèi)存大小。
        props.put("key.serializer",
                "org.apache.kafka.common.serialization.IntegerSerializer");
        props.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");

        producer = new KafkaProducer<String, String>(props);
    }

    public void sendKafkaMessageString() throws InterruptedException {

        int messageNo = 1;
        final int COUNT = 2;
        while(messageNo < COUNT) {
            String key = String.valueOf(messageNo);
            //String data = String.format("hello KafkaProducer message %s from hubo 06291018 ", key);
            String data = String.format("best test zzz |kk");
            Thread.sleep(1);  //1000代表1秒
            System.out.println(messageNo);
            try {
                producer.send(new ProducerRecord<String, String>(TOPIC, data));
            } catch (Exception e) {
                e.printStackTrace();
            }
            messageNo++;
        }
        producer.close();
    }

    public static void main(String[] args) throws InterruptedException {
        new SendStopOrderMessageTest().sendKafkaMessageString();
    }


}