wsl下搭建 kafka raft 集群

修改配置
node.id=1
controller.quorum.voters=1@localhost:9093,2@localhost:8093,3@localhost:7093
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
inter.broker.listener.name=PLAINTEXT
advertised.listeners=PLAINTEXT://localhost:9092

生成集群ID

./bin/kafka-storage.sh random-uuid

格式化日志目录

./bin/kafka-storage.sh format -t Cba3BkapTFWyEKbpUkNJ_w -c ./config/kraft/server.properties
./bin/kafka-storage.sh format -t Cba3BkapTFWyEKbpUkNJ_w -c ./config/kraft/server2.properties
./bin/kafka-storage.sh format -t Cba3BkapTFWyEKbpUkNJ_w -c ./config/kraft/server3.properties

启动集群

bin/kafka-server-start.sh config/kraft/server.properties 2>&1 >/tmp/logs/kafka/server1.log &
bin/kafka-server-start.sh config/kraft/server2.properties 2>&1 >/tmp/logs/kafka/server2.log &
bin/kafka-server-start.sh config/kraft/server3.properties 2>&1 >/tmp/logs/kafka/server3.log &

./bin/kafka-topics.sh --create --topic test --partitions 1 --replication-factor 3 --bootstrap-server localhost:9092,localhost:8092,localhost:7092
./bin/kafka-topics.sh --list --bootstrap-server localhost:9092,localhost:8092,localhost:7092

生产者

#coding:utf-8

from kafka import KafkaProducer

kafka_servers = ['localhost:7092']
producer = KafkaProducer(bootstrap_servers=kafka_servers)

producer.send('test',b'ddd')

future = producer.send('test',b'123vvv')

result = future.get(timeout= 10)

print(result)

消费者

#coding:utf-8

from kafka import KafkaConsumer
from kafka.structs import TopicPartition

kafka_servers = ['localhost:9092','localhost:8092','localhost:7092']
consumer = KafkaConsumer( group_id='test2' ,bootstrap_servers=kafka_servers,enable_auto_commit=False )
consumer.subscribe( topics=('test',) )
# for msg in consumer:
#     print(msg)
# for r in consumer.assignment(): print(r)
tp = TopicPartition( topic='test',partition=0 )

def pl():
    offset_committed = consumer.committed( tp )
    offset_end = consumer.end_offsets( [tp] )
    print(offset_committed,offset_end)

pl()
res = consumer.poll(10,2)
for msgs in res.values():
    for msg in msgs:
        print(msg)
        # consumer.commit()
pl()