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地址，但是我的修改完后还是报错
  最后检查了一下防火墙的设置，是开启状态的，最后关闭防火墙，成功连接