主线程Hibernate小于1000时无法生成消息

Question

public class Producer {
    private final KafkaProducer<String, String> producer;
    private final String topic;

    public Producer(String topic, String[] args) {
        //......
        //......
        producer = new KafkaProducer<>(props);
        this.topic = topic;
    }

    public void producerMsg() throws InterruptedException {
        String data = "Apache Storm is a free and open source distributed";
        data = data.replaceAll("[\\pP‘’“”]", "");
        String[] words = data.split(" ");
        Random _rand = new Random();

        Random rnd = new Random();
        int events = 10;
        for (long nEvents = 0; nEvents < events; nEvents++) {
            long runtime = new Date().getTime();
            int lastIPnum = rnd.nextInt(255);
            String ip = "192.168.2." + lastIPnum;
            String msg = words[_rand.nextInt(words.length)];
            try {
                producer.send(new ProducerRecord<>(topic, ip, msg));
                System.out.println("Sent message: (" + ip + ", " + msg + ")");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Producer producer = new Producer(Constants.TOPIC, args);
        producer.producerMsg();
        //If I write Thread.sleep(1000),It will not work!!!!!!!!!!!!!!!!!!!!
        Thread.sleep(2000);
    }
}

我很感激

Answer 1

你能展示一下你用来配置制作人的道具吗？我只是猜测有可能...

在producerMsg（）中，您使用的是异步方式来使用生产者，所以只有producer.send（），这意味着消息被放在一个内部缓冲区中，用于生成稍后发送的批处理。生产者有一个内部线程可以从缓冲区获取并发送批处理。也许只有1000 ms不足以达到这样的条件：生产者真正发送消息（请参见batch.size和linger.ms)，主应用程序结束，生产者在没有发送消息的情况下死亡。给它更多的时间（2000毫秒），它可以工作。顺便说一句，我没试过密码。

所以原因似乎是你的：