windows本地简单调试kafka_本地使用kafka

使用的kafka是3.4.0版本

一、官网地址：​​​​​​Apache Kafka

 二、下载后解压扔到自己相应的目录下，因为简单的调试自己的脚本，所以没修改配置文件

 三、启动

开三个黑窗口，到kafka安装路径下执行，进入路径不对执行也会报错

第一个黑窗口启动zk

.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

第二个黑窗口启动kafka

.\bin\windows\kafka-server-start.bat .\config\server.properties

第三个黑窗口，用来验证

1.创建topic     

./kafka-topics.bat --bootstrap-server localhost:9092 --topic test -create --partitions 1 --replication-factor 1

2.查看信息     

./kafka-topics.sh --bootstrap-server localhost:9092 --list

3.删除topic

./kafka-topics.sh --delete --bootstrap-server localhost:9092 --topic test

#------------------------------------下面是简单代码调试------------------------------------

python

#也可以尝试在生产的接入点后加入这个参数选择自己对应的版本api_version=(2, 6, 0)
 
 
#生产
from kafka3 import KafkaProducer
producer = KafkaProducer(bootstrap_servers='接入点:9092')
for i in range(100):
    producer.send('test', b'some_message_bytes')
 
 
 
#消费
from kafka3 import KafkaConsumer
consumer = KafkaConsumer('test', group_id='python',bootstrap_servers='接入点:9092')
for msg in consumer:
    print (msg)

java

<dependencies>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>3.4.0</version>
    </dependency>
</dependencies>

package org.example;
 
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
 
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
 
import java.util.Collections;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;
 
public class test {
 
    //主题名称
    private final static String TOPIC_NAME = "test";
 
    public static void main(String[] args) throws ExecutionException, InterruptedException {
 
 
        Properties properties = new Properties();
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
 
        // 创建管理者
        AdminClient adminClient = AdminClient.create(properties);
 
        try {
            // 创建topic前，可以先检查topic是否存在，如果已经存在，则不用再创建了
            Set<String> topics = adminClient.listTopics().names().get();
            if (topics.contains("yangtopic")) {
                return;
            }
 
            // 创建topic
            NewTopic newTopic = new NewTopic("yangtopic", 1, (short) 1);
            CreateTopicsResult result = adminClient.createTopics(Collections.singletonList(newTopic));
            result.all().get();
        } finally {
            adminClient.close();
        }
 
 
 
        //1. 设置参数
        Properties props = new Properties();
        //指定服务器配置【ip：端口】
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
 
        //把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把发送消息的value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 
        //2. 创建生产消息的客户端，传入参数
        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        //3. 创建消息；key：作用是决定了往哪个分区上发，value：具体要发送的消息内容
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(
                TOPIC_NAME,"myKeyValue", "helloKafka");
        //4. 发送消息，得到消息发送的元数据并输出
        RecordMetadata metadata = producer.send(producerRecord).get();
        System.out.println("同步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-"
                + metadata.partition() + "|offset-" + metadata.offset());
    
 
 
    }
}

#-----------------------------------------调试遇到的问题-------------------------------------------------

在windows本地用Java删除topic，导致kafka的服务停了zk还正常运行再重启kafka也无效，可以尝试把kafka和zk的日志文件以及数据全都删掉，再重新启用。

但是同样代码我在linux调用docker里面的Kafka没出问题。

ERROR Shutdown broker because all log dirs in D:\tmp\kafka-logs have failed (kafka.log.LogManager)

我没改配置文件所以是在这个位置全删掉，里面的topic数据也会被删掉