win setup kafka 3.6.2 Step-by-Step Guide

打印 上一主题 下一主题

主题 628|帖子 628|积分 1884

At the end of the document, some bugs are recorded
setup

from https://kafka.apache.org/downloads download .tgz binary package to local and extract
Prerequisites


  • Kafka Installed: Ensure Kafka is installed and running.
  • Java Installed: Kafka requires Java. Make sure having the JDK installed.
edit config file



  • edit config/server.properties file:
    broker.id=0
    log.dirs=/tmp/kafka-logs # or in window use D:\\tmp\\kafka-logs
    zookeeper.connect=localhost:2181
    listeners=PLAINTEXT://:9092
  • edit config/zookeeper.properties file:
    dataDir=/bigdata/zk # in win use D:\\bigdata\\zk
1. Start Kafka Server

Make sure Zookeeper and Kafka server are running.
Start Zookeeper:

  1. .\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties
复制代码
Start Kafka server:

  1. .\bin\windows\kafka-server-start.bat .\config\server.properties
复制代码
2. Create a Kafka Topic

Before producing and consuming messages, need a topic.
  1. .\bin\windows\kafka-topics.bat --create --topic test --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
复制代码
3. Set Up Kafka Producer

use the Kafka console producer to send messages to the topic.
Open a new Command Prompt and run:
  1. .\bin\windows\kafka-console-producer.bat --topic test --bootstrap-server localhost:9092
复制代码
Type messages in the console to send them to the Kafka topic.
4. Set Up Kafka Consumer

Open another Command Prompt to start the consumer that reads messages from the topic.
  1. .\bin\windows\kafka-console-consumer.bat --topic test --bootstrap-server localhost:9092 --from-beginning
复制代码
should see messages in the consumer console as type them in the producer console.
Connecting Kafka with Code
Here are examples in Java and Python.
Java Example

First, add Kafka client dependencies topom.xml if using Maven:
  1. <dependency>
  2.     <groupId>org.apache.kafka</groupId>
  3.     <artifactId>kafka-clients</artifactId>
  4.     <version>3.6.2</version>
  5. </dependency>
复制代码
Producer Example
  1. import org.apache.kafka.clients.producer.KafkaProducer;
  2. import org.apache.kafka.clients.producer.ProducerRecord;
  3. import java.util.Properties;
  4. public class KafkaProducerExample {
  5.     public static void main(String[] args) {
  6.         Properties props = new Properties();
  7.         props.put("bootstrap.servers", "localhost:9092");
  8.         props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  9.         props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  10.         KafkaProducer<String, String> producer = new KafkaProducer<>(props);
  11.         producer.send(new ProducerRecord<>("test", "key", "value"));
  12.         producer.close();
  13.     }
  14. }
复制代码
Consumer Example
  1. import org.apache.kafka.clients.consumer.ConsumerRecords;
  2. import org.apache.kafka.clients.consumer.KafkaConsumer;
  3. import java.time.Duration;
  4. import java.util.Collections;
  5. import java.util.Properties;
  6. public class KafkaConsumerExample {
  7.     public static void main(String[] args) {
  8.         Properties props = new Properties();
  9.         props.put("bootstrap.servers", "localhost:9092");
  10.         props.put("group.id", "test-group");
  11.         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  12.         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  13.         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
  14.         consumer.subscribe(Collections.singletonList("test"));
  15.         while (true) {
  16.             ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
  17.             records.forEach(record -> {
  18.                 System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
  19.             });
  20.         }
  21.     }
  22. }
复制代码
Python Example

First, install the Kafka Python client:
  1. pip install kafka-python
复制代码
Producer Example
  1. from kafka import KafkaProducer
  2. producer = KafkaProducer(bootstrap_servers='localhost:9092')
  3. producer.send('test', b'Hello, Kafka!')
  4. producer.close()
复制代码
Consumer Example
  1. from kafka import KafkaConsumer
  2. consumer = KafkaConsumer('test', bootstrap_servers='localhost:9092', auto_offset_reset='earliest')
  3. for message in consumer:
  4.     print(f"Key: {message.key}, Value: {message.value}")
复制代码
Note


  • Start Zookeeper and Kafka server: Ensure they are running correctly.
  • Create topics: Use Kafka commands to create the required topics.
some bug

‘wmic’ is not recognized as an internal or external command, operable program or batch file
click Environment Variables. In the section for system variables, find PATH (or any capitalization thereof). Add this entry to it:
%SystemRoot%\System32\Wbem
ERROR Exiting Kafka due to fatal exception during startup. (kafka.Kafka$) java.nio.file.InvalidPathException: Illegal char < > at index 2: D: mpdownloadkafkakafka_2.13-3.6.2log\meta.properties.tmp
Correct the Path Format:
Ensure that the path specified in configuration does not contain illegal characters or spaces. Paths in Windows should use double backslashes \ or a single forward slash /.
WARN [SocketServer listenerType=ZK_BROKER, nodeId=0] Unexpected error from /0:0:0:0:0:0:0:1 (channelId=0:0:0:0:0:0:0:1:9092-0:0:0:0:0:0:0:1:62710-1); closing connection (org.apache.kafka.common.network.Selector) org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = 1195725856 larger than 104857600)


  • Edit server.properties:
    Open the server.properties file in Kafka config directory and increase the max.request.size property. Add or modify the following lines:
    max.request.size=209715200 # Increase this value as needed, default is 104857600 (100MB)
    socket.request.max.bytes=209715200 # Ensure this matches or exceeds max.request.size
  • Edit consumer.properties and producer.properties (if applicable):
    If have consumer and producer configurations, ensure that these properties are set appropriately there as well:
    max.request.size=209715200

免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览

快速回复

您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

徐锦洪

金牌会员
这个人很懒什么都没写!

标签云

快速回复 返回顶部 返回列表