kafka使用

时间:2023-12-04 20:47:20

0:

  ./sbt update
  ./sbt package
  ./sbt assembly-package-dependency

1: 启动ZK:

  通过kafka的命令启动:bin/zookeeper-server-start.sh config/zookeeper.properties

  或者直接启动zk集群:zkServer.sh start

2:启动kafka的服务器:

  1. 修改配置文件config/server.properties,host.name=115.28.23.38,将host.name修改,这个host.name是kafka的server的机器名字,会注册到zk中。
  2. 修改zookeeper.connect=hostname:port/kafka,注意要事先在zk中创建/kafka节点,否则回报出错误:java.lang.IllegalArgumentException: Path length must be > 0
  3. 修改num.partitions、log.dirs、num.network.threads、num.io.threads
  4. 修改运行文件 bin/kafka-server-start.sh ,export KAFKA_HEAP_OPTS="-Xmx256M -Xms256M",调整jvm堆大小为256m。
  5. 修改config/log4j.properties的log输出路径
  6. 运行:bin/kafka-server-start.sh config/server.properties,生成环境需要后台启动方式:nohup  &

3:创建一个topic:

  bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

4:绑定topic,生产消息:

  bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

5:展示当前的topic:

  bin/kafka-topics.sh --list --zookeeper localhost:2181

6:消费消息:

  bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

7:启动kafka

  sh bin/kafka-server-start.sh -daemon config/server.properties

编写客户端生产、消费的代码: