Java API中kafka生产者发送消息没有成功

Posted 2022-05-17 二木成林

异常

Java API中kafka生产者发送消息被阻塞，没有任何异常信息，发送消息也没有成功：

错误代码

在 API 中发送生产者消息的代码如下：

public class CustomProducer 
    public static void main(String[] args) 
        // 1.创建 Kafka 生产者的配置对象，所有关于生产者的配置都添加在 Properties 类中
        Properties prop = new Properties();

        // 2.给 Kafka 配置对象添加配置信息
        // 2.1 配置 bootstrap.servers
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.88.142:9092");
        // 2.2 配置 key.serializer 和 value.serializer
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 3.创建 Kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(prop);

        // 4.调用 send 方法，发送消息
        for (int i = 0; i < 5; i++) 
            kafkaProducer.send(new ProducerRecord<>("first", "zhangsan_" + i));
        

        // 5.关闭资源
        kafkaProducer.close();
    

解决

首先查看服务器的防火墙是否关闭，如果没有关闭则关闭防火墙。我使用的Linux系统是 CentOS7，所以采用查看防火墙状态的命令是：systemctl status firewalld.service。

如果防火墙是关闭的，仍然无法使用 API 发送生产者消息成功，那么修改 Kafka 安装目录下的 conf 目录下的 server.properties 配置文件，即取消 advertised.listeners 的注释，并且将 your.host.name 修改为你本地服务器的IP地址。例如：advertised.listeners=PLAINTEXT://192.168.88.142:9092。

修改之后重启Kafka
再度监听消费者，在代码中通过 API 发送消息成功。

注意，服务器中消费者脚本监听的 topic 必须和代码中指定发送消息的 topic 一致，否则也不能监听到。