Kafka新旧版本命令问题

迈不过友情╰ 2023-07-24 14:41 29阅读 0赞

### Kafka新旧版本命令问题 ###

*  1、异常汇总
 *  2、问题分析
 *  3、问题解决
 *  4、使用java生产数据
 *  4、总结

# 1、异常汇总 #

输入如下命令生成一个生产者`/usr/local/chen/kafka/bin/kafka-console-producer.sh --broker-list 192.168.43.175:2181,192.168.43.176:2181,192.168.43.177:2181 --topic chentest7`。出现以下异常。

>[2020-04-12 15:27:51,725] WARN [Producer clientId=console-producer] Bootstrap broker 192.168.43.175:2181 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
    [2020-04-12 15:27:51,830] WARN [Producer clientId=console-producer] Bootstrap broker 192.168.43.177:2181 (id: -3 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
    [2020-04-12 15:27:51,932] WARN [Producer clientId=console-producer] Bootstrap broker 192.168.43.177:2181 (id: -3 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
    [2020-04-12 15:27:52,036] WARN [Producer clientId=console-producer] Bootstrap broker 192.168.43.175:2181 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
    [2020-04-12 15:27:52,143] WARN [Producer clientId=console-producer] Bootstrap broker 192.168.43.176:2181 (id: -2 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
    [2020-04-12 15:27:52,245] WARN [Producer clientId=console-producer] Bootstrap broker 192.168.43.176:2181 (id: -2 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNTEwNTk3_size_16_color_FFFFFF_t_70]

# 2、问题分析 #

这是由于新旧版本`kafka`命令不同。

# 3、问题解决 #

使用新版本的`kafka`启动命令即可。`/usr/local/chen/kafka/bin/kafka-console-producer.sh --broker-list 192.168.43.175:9092,192.168.43.176:9092,192.168.43.177:9092 --topic chentest6`用来启动生产者。输入信息即可被消费者监听到。

![在这里插入图片描述][20200412155956453.png]

然后使用`/usr/local/chen/kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.43.175:9092,192.168.43.176:9092,192.168.43.177:9092 --topic chentest6 --from-beginning`命令生成消费者。我们发现消费者已经成功的消费到信息。

![在这里插入图片描述][20200412160005436.png]

# 4、使用java生产数据 #

简单的定义一个类，配置`kafka`的相关信息，用来生成数据。

import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    
    /** * @author Chen * @version 1.0 * @date 2020/4/10 8:03 * @description: */
    public class ProducerDemo { 
        private final KafkaProducer<String, String> producer;
    
    // public final static String TOPIC = "access-log";
        public final static String TOPIC = "chentest6";
    
        private ProducerDemo() { 
            Properties props = new Properties();
            props.put("zookeeper.connect", "192.168.43.175:2181,192.168.43.176:2181,192.168.43.177:2181");
            props.put("acks", "all");//所有follower都响应了才认为消息提交成功，即"committed"
            props.put("retries", 0);//retries = MAX 无限重试，直到你意识到出现了问题:)
            props.put("batch.size", 16384);//producer将试图批处理消息记录，以减少请求次数.默认的批量处理消息字节数
            //batch.size当批量的数据大小达到设定值后，就会立即发送，不顾下面的linger.ms
            props.put("linger.ms", 1);//延迟1ms发送，这项设置将通过增加小的延迟来完成--即，不是立即发送一条记录，producer将会等待给定的延迟时间以允许其他消息记录发送，这些消息记录可以批量处理
            props.put("buffer.memory", 33554432);//producer可以用来缓存数据的内存大小。
            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 produce() { 
            int messageNo = 1;
            final int COUNT = 5;
    
            while (messageNo < COUNT) { 
                String key = String.valueOf(messageNo);
                String data = String.format("hello KafkaProducer message %s from hubo chen ", key);
    
                try { 
                    producer.send(new ProducerRecord<String, String>(TOPIC, data));
                } catch (Exception e) { 
                    e.printStackTrace();
                }
    
                messageNo++;
            }
    
            producer.close();
        }
    
        public static void main(String[] args) { 
            ProducerDemo producerDemo = new ProducerDemo();
            producerDemo.produce();
        }
    
    }

运行后不出意外出现了异常。

Exception in thread "main" org.apache.kafka.common.config.ConfigException: Missing required configuration "bootstrap.servers" which has no default value.
    	at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:463)
    	at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:453)
    	at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:62)
    	at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:75)
    	at org.apache.kafka.clients.producer.ProducerConfig.<init>(ProducerConfig.java:359)
    	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:287)
    	at ProducerDemo.<init>(ProducerDemo.java:35)
    	at ProducerDemo.main(ProducerDemo.java:59)

同样是版本问题，在新版本中，使用`props.put("bootstrap.servers", "192.168.43.175:9092,192.168.43.176:9092,192.168.43.177:9092");`配置服务器。修改代码如下：

import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    
    /** * @author Chen * @version 1.0 * @date 2020/4/10 8:03 * @description: */
    public class ProducerDemo { 
        private final KafkaProducer<String, String> producer;
    
    // public final static String TOPIC = "access-log";
        public final static String TOPIC = "chentest6";
    
        private ProducerDemo() { 
            Properties props = new Properties();
            props.put("bootstrap.servers", "192.168.43.175:9092,192.168.43.176:9092,192.168.43.177:9092");//xxx服务器ip
    // props.put("bootstrap.servers", "192.168.43.120:9092");//xxx服务器ip
    // props.put("bootstrap.servers", "192.168.43.175:9092");//xxx服务器ip
            props.put("acks", "all");//所有follower都响应了才认为消息提交成功，即"committed"
            props.put("retries", 0);//retries = MAX 无限重试，直到你意识到出现了问题:)
            props.put("batch.size", 16384);//producer将试图批处理消息记录，以减少请求次数.默认的批量处理消息字节数
            //batch.size当批量的数据大小达到设定值后，就会立即发送，不顾下面的linger.ms
            props.put("linger.ms", 1);//延迟1ms发送，这项设置将通过增加小的延迟来完成--即，不是立即发送一条记录，producer将会等待给定的延迟时间以允许其他消息记录发送，这些消息记录可以批量处理
            props.put("buffer.memory", 33554432);//producer可以用来缓存数据的内存大小。
            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 produce() { 
            int messageNo = 1;
            final int COUNT = 5;
    
            while (messageNo < COUNT) { 
                String key = String.valueOf(messageNo);
                String data = String.format("hello KafkaProducer message %s from hubo chen ", key);
    
                try { 
                    producer.send(new ProducerRecord<String, String>(TOPIC, data));
                } catch (Exception e) { 
                    e.printStackTrace();
                }
    
                messageNo++;
            }
    
            producer.close();
        }
    
        public static void main(String[] args) { 
            ProducerDemo producerDemo = new ProducerDemo();
            producerDemo.produce();
        }
    
    }

直接运行，查看服务器消费者信息。

![在这里插入图片描述][20200412160018744.png]

可以成功消费到数据。

# 4、总结 #

软件的更新修改了BUG，或许也添加了新的功能，但是对使用者来说有点不太友好，要是自动检测到某个命令是旧命令，然后提示用户输入一个新的命令，是不是想想都激动。算了当前幻想，好好看官方文档吧！

点个赞再走呗！欢迎留言哦！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNTEwNTk3_size_16_color_FFFFFF_t_70]: /images/20230528/b4d73616eca842558db2b641cb45d67b.png
[20200412155956453.png]: /images/20230528/daad66cf7eb749679b792c7853270f9c.png
[20200412160005436.png]: /images/20230528/50f2a61e9c36437d8d62baab8dc730c4.png
[20200412160018744.png]: /images/20230528/ed78e4a15e6f4ec2906c68ddd793c457.png