1. kafka安装与配置

  • 官网下载安装解压缩:http://kafka.apache.org/downloads
    在这里插入图片描述
  • 下载解压启动
启动命令:bin/kafka-server-start.sh  -daemon config/server.properties

关闭命令:./bin/kafka-server-stop.sh

server.properties配置中需要关注以下几个参数:
broker.id=0 表示broker的编号,如果集群中有多个broker,则每个broker的编号需要设置
的不同
listeners=PLAINTEXT://:9092 brokder对外提供的服务入口地址
log.dirs=/tmp/kafka/log 设置存放消息日志文件的地址
zookeeper.connect=localhost:2181 Kafka所需Zookeeper集群地址,教学中Zookeeper和
Kafka都安装本机

在这里插入图片描述

2. Kafka测试消息生产与消费

  • 首先创建一个主题
    命令如下:
bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic heima --partitions 2 --replication-factor 1

--zookeeper:指定了Kafka所连接的Zookeeper服务地址
--topic:指定了所要创建主题的名称
--partitions:指定了分区个数
--replication-factor:指定了副本因子
--create:创建主题的动作指令
  • 报错
    在这里插入图片描述
  • 解决
    解决方案:
    在较新版本(2.2 及更高版本)的 Kafka 不再需要 ZooKeeper 连接字符串,即- -zookeeper localhost:2181。使用 Kafka Broker的 --bootstrap-server localhost:9092来替代- -zookeeper localhost:2181。
bin/kafka-topics.sh --bootstrap-server localhost:9092  --create --topic heima --partitions 2 --replication-factor 1

在这里插入图片描述

  • 展示所有主题
    bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
    在这里插入图片描述
  • 查看主题详情
    bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic heima
    在这里插入图片描述
  • 启动消费端接收消息
    命令:bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic heima

–bootstrap-server 指定了连接Kafka集群的地址
–topic 指定了消费端订阅的主题

itcast@Server-node:/mnt/d/kafka_2.12-2.2.1$ bin/kafka-console-consumer.sh -- bootstrap-server localhost:9092 --topic heima Hello,Kafka!
  • 生产端发送消息
    命令:bin/kafka-console-producer.sh --broker-list localhost:9092 --topic heima

–broker-list 指定了连接的Kafka集群的地址
–topic 指定了发送消息时的主题

itcast@Server-node:/mnt/d/kafka_2.12-2.2.1$ bin/kafka-console-producer.sh -- broker-list localhost:9092 --topic heima >Hello,Kafka!
  • 生产者在linux输入发送消息的内容,消费者可以查收到

在这里插入图片描述
在这里插入图片描述

3. java连接Kafka

  • Producer
package com.hisign.xzxt2.kafka.learn.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class ProducerDemo {


    public static void main(String[] args) {
        Properties properties = new Properties();
        //broker的地址清单,建议至少填写两个,避免宕机
        properties.put("bootstrap.servers", "192.168.42.160:9092");
        //acks指定必须有多少个分区副本接收消息,生产者才认为消息写入成功,用户检测数据丢失的可能性
        //acks=0:生产者在成功写入消息之前不会等待任何来自服务器的响应。无法监控数据是否发送成功,但可以以网络能够支持的最大速度发送消息,达到很高的吞吐量。
        //acks=1:只要集群的首领节点收到消息,生产者就会收到来自服务器的成功响应。
        //acks=all:只有所有参与复制的节点全部收到消息时,生产者才会收到来自服务器的成功响应。这种模式是最安全的,
        properties.put("acks", "all");
        //retries:生产者从服务器收到的错误有可能是临时性的错误的次数
        properties.put("retries", 0);
        //batch.size:该参数指定了一个批次可以使用的内存大小,按照字节数计算(而不是消息个数)
        properties.put("batch.size", 16384);
        //linger.ms:该参数指定了生产者在发送批次之前等待更多消息加入批次的时间,增加延迟,提高吞吐量
        properties.put("linger.ms", 1);
        //buffer.memory该参数用来设置生产者内存缓冲区的大小,生产者用它缓冲要发送到服务器的消息。
        properties.put("buffer.memory", 33554432);
        //key和value的序列化
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = new KafkaProducer<String, String>(properties);
        try {
            //producer = new KafkaProducer<>(properties);
            for (int i = 0; i < 10; i++) {
                String values = "value" + i + "\t";
                producer.send(new ProducerRecord<String, String>("heima", "key" + Integer.toString(i), values));
                Thread.sleep(500);
                System.out.println("Sent:" + values);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
}


  • Consumer
package com.hisign.xzxt2.kafka.learn.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class ConsumerDemo {

    public static void main(String[] args) throws Exception {
        String topicName = "heima";
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.42.160:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        //Kafka Consumer subscribes list of topics here.
        consumer.subscribe(Arrays.asList(topicName));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
                // print the offset,key and value for the consumer records.
                System.out.printf("offset = %d, key = %s, value = %s\n",
                        record.offset(), record.key(), record.value());
        }
    }

}


  • 报错
connection to node -1 (/192.168.42.161:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
  • 解决
    网上好多是修改server.properties文件中绑定的ip地址,但是我的修改完后还是报错
    最后检查了一下防火墙的设置,是开启状态的,最后关闭防火墙,成功连接

在这里插入图片描述
在这里插入图片描述

Logo

为开发者提供学习成长、分享交流、生态实践、资源工具等服务,帮助开发者快速成长。

更多推荐