Kafka新旧版本命令问题

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)

在这里插入图片描述

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用来启动生产者。输入信息即可被消费者监听到。

在这里插入图片描述

然后使用/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命令生成消费者。我们发现消费者已经成功的消费到信息。

在这里插入图片描述

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();
    }

}

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

在这里插入图片描述

可以成功消费到数据。

4、总结

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

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

# kafka # java # 大数据 # 分布式 # hadoop
Logo
CSDN学习社区

CSDN联合极客时间，共同打造面向开发者的精品内容学习社区，助力成长！

更多推荐

嵌入式作业（七）：基于Ardunio的STM32串口通信

嵌入式作业（七）0作业要求1Ardunio 完成STM32的串口通信（1）安装Ardunio IDE（2）stm32串口通信2关于 stduino IDE0作业要求安装 Ardunio IDE 和相关软件支持库，在Ardunio 完成STM32板子的串口通信程序：（1）持续向串口输出“Hello world！”；（2）当接收到“stop!”时，停止输出。网上有一个国人版的MCU集成开发平台， st

avatar CSDN学习社区

JDBC详解

JDBC文章目录JDBC什么是JDBC?JDBC驱动程序:Java使用JDBC访问数据库的步骤:设置classpath:Oracle连接字符串的书写格式:简单的例子:常用数据库的驱动程序及JDBC URL:Oracle数据库:SQL Server数据库MySQL数据库Access数据库PreparedStatement接口:JNDI-数据源（Data Source）与连接池（Connection

avatar CSDN学习社区

“模式识别与机器学习”学习笔记no2.再谈感知机

接**上篇：上篇主要进行了PLA，Pocket算法的理论过程分析和在给定数据集上利用pocket算法对数据集进行分类学习，得到错分数量最少的分类面。上篇中pocket算法的过程已经进行了编程和测试，框架已经建立了起来，这一篇主要上篇中没有提到或涉及不深的几个问题。1.数据集的构造。上篇是直接使用了题目给的向量，这次来根据正态分布来产生数据集。np.random.normal函数可以根据均值和方差生

avatar CSDN学习社区