kafka怎么安装和配置？从下载到启动的完整教程

tar -zxvf kafka_2.12-2.4.1 -c /usr/local/

# 每个broker节点的id，集群架构下要唯一
broker.id = 0
# kafka对外提供的ip和端口号
#listeners=plaintext://:9092 # 这是系统默认的
# 这里的ip要使用内网的，也就是ifconfig查看出来的
listeners=plaintext://内网ip:9092
# 外部代理地址，比如java代码使用客户端操作连接kafka服务器
advertised.listeners=plaintext://外网ip:9092
# 消息存储的目录
log.dirs=/usr/local/kafka_2.12-2.4.1/data
# zookeeper地址
zookeeper.connect=localhost:2181
# 消息保留的时间，单位为小时，默认为7天
log.retention.hours=168

./bin/kafka-server-start.sh config/server.properties &
#或者
./bin/kafka-server-start.sh -daemon config/server.properties  

jps或者 ps -ef | grep kafka

# 把元信息数据存储到zk中，真正队列的数据还是存储在kafka的broker中 
# --partitions为分区数量，默认也是1个分区 --topic test为指定topic的名称
./bin/kafka‐topics.sh ‐‐create ‐‐zookeeper localhost:2181 ‐‐replication‐factor 1 ‐‐partitions 1 ‐‐topic test
# 查询有哪些topic
./bin/kafka-topics.sh --list --zookeeper localhost:2181
# 删除一个topic
./bin/kafka‐topics.sh ‐‐delete ‐‐topic test ‐‐zookeeper localhost:2181
# 查看topic的情况
./bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test

# 这里的ip为内网ip
./bin/kafka-console-producer.sh --broker-list ip:9092 --topic test

# 这里的ip为内网ip，该命令只能消费该消费者启动成功之后生产者发送的消息
./bin/kafka-console-consumer.sh --bootstrap-server ip:9092 --topic test
# 消费生产者所有的消息，只需要在上面的命令加上 --from-beginning
# kafka消费完之前的消息不会立马删除，还会在磁盘的文件里面存在，默认保留一周，这是与传统消息中间件不同之一
./bin/kafka-console-consumer.sh --bootstrap-server ip:9092 --from-beginning --topic test
# 消费多个主题的消息，加上 --whitelist "topic1|topic2"，每个主题中间用管道符 相隔
 ./bin/kafka‐console‐consumer.sh ‐‐bootstrap‐server ip:9092 ‐‐whitelist "test|test‐2"

# 通过 --consumer-property group.id=testgroup 来进行指定消费者组名
./bin/kafka‐console‐consumer.sh ‐‐bootstrap‐server ip:9092 ‐‐consumer‐property group.id=testgroup ‐‐topic test

# 通过 --consumer-property group.id=testgroup 来进行指定消费者组名
./bin/kafka‐console‐consumer.sh ‐‐bootstrap‐server ip:9092 ‐‐consumer‐property group.id=testgroup2 ‐‐topic test

./bin/kafka-consumer-groups.sh --bootstrap-server ip:9092 --list

./bin/kafka-consumer-groups.sh --bootstrap-server ip:9092 --describe --group testgroup

##### --partitions 后面的参数为分区的数量
./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 2 --topic test1

./bin/kafka‐topics.sh ‐alter ‐‐partitions 3 ‐‐zookeeper localhost:2181 ‐‐topic test
# 重新查看该topic的情况
./bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test

# 每个broker节点的id，集群架构下要唯一
broker.id = 1
# kafka对外提供的ip和端口号
#listeners=plaintext://:9092 # 这是系统默认的
# 这里的ip要使用内网的，也就是ifconfig查看出来的
listeners=plaintext://ip:9093
# 外部代理地址，比如java代码使用客户端操作连接kafka服务器
advertised.listeners=plaintext://外网ip:9093
# 消息存储的目录
log.dirs=/usr/local/kafka_2.12-2.4.1/data2
# zookeeper地址
zookeeper.connect=localhost:2181

# 每个broker节点的id，集群架构下要唯一
broker.id = 2
# kafka对外提供的ip和端口号
#listeners=plaintext://:9092 # 这是系统默认的
# 这里的ip要使用内网的，也就是ifconfig查看出来的
listeners=plaintext://ip:9094
# 外部代理地址，比如java代码使用客户端操作连接kafka服务器
advertised.listeners=plaintext://外网ip:9094
# 消息存储的目录
log.dirs=/usr/local/kafka_2.12-2.4.1/data3
# zookeeper地址
zookeeper.connect=localhost:2181

# --replication-factor 3 指定副本为3个，每个分区都分别对应3个副本
# --partitions 2 分区为2个
./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 2 --topic my-replication-topic

<dependency>
    <groupid>org.apache.kafka</groupid>
    <artifactid>kafka-clients</artifactid>
    <version>2.4.1</version>
</dependency>

public class producer {
    
    public static void main(string[] args) throws executionexception, interruptedexception {
        properties properties = new properties();
        // 如果当前连接的kafka配置了集群，只需要注册一个就可以，他会自动把当前集群里面所有的节点注册进来。
        // 一般情况下为了高可用，还是在这里多注册两个节点，以防某一个节点挂掉了，其他的也注册不上。
        properties.setproperty(producerconfig.bootstrap_servers_config,"ip:9092,ip:9093,ip:9094");
        
        // key 序列化
        properties.setproperty(producerconfig.key_serializer_class_config, stringserializer.class.getname());
        properties.setproperty(producerconfig.value_serializer_class_config, stringserializer.class.getname());
        
        // 宕机的情况下重试几次，默认为3此
        //        properties.setproperty(producerconfig.retries_config,"3");
        // 每次重试间隔时间，单位毫秒
        //        properties.setproperty(producerconfig.retry_backoff_ms_config,"100");
        //生产者每次批量发送消息的大小，默认16kb，一定要配合下面的 linger.ms 参数使用
        //        properties.setproperty(producerconfig.batch_size_config,"10");
        // 生产者缓存数据到一定时间发送，默认时间为0毫秒（单位毫秒），kafka作者建议8-10毫秒
        //        properties.setproperty(producerconfig.linger_ms_config,"0");
        // 压缩的类型，一共四种类型(none、gzip、snappy、lz4)，一般情况下用第三个，使用压缩的会降低性能
        //        properties.setproperty(producerconfig.compression_type_config, "snappy");
        //client.id，此参数也可以不配置，代表消息从哪里来（在发出请求时传递给服务器的id字符串。
        // 这样做的目的是通过允许在服务器端请求日志中包含逻辑应用程序名称，从而能够跟踪ip/端口之外的请求源。）
        //        properties.setproperty(producerconfig.client_id_config, "");
        // 在阻塞之前，客户机在单个连接上发送的未确认请求的最大数目。请注意，如果此设置设置为大于1且发送失败，则由于重试(即，如果启用重试)
        //        properties.setproperty(producerconfig.max_in_flight_requests_per_connection, "");
        
        // 自定义分区管理器（分区就是存储数据所在的文件夹）
        //        properties.setproperty("partitioner.class",mypartitioner.class.getname());
        kafkaproducer<string, string> kafkaproducer = new kafkaproducer<string, string>(properties);
        
        for (int i = 11; i <= 20; i++) {
//                        producerrecord<string, string> record = new producerrecord<string, string>("topic2", 0,"key" + i,"hello world" + i);
            
            // 未指定分区，使用 hash(key) % partition 来选择分区。核心代码 `utils.topositive(utils.murmur2(keybytes)) % numpartitions`
            producerrecord<string, string> record = new producerrecord<string, string>("my-replication-topic","key" + i, "hello world" + i);
            // 使用的并发编程 `future` ，异步发送，拿结果还是同步的。
            // 也就是消息发送不成功，这里一直会阻塞
            future<recordmetadata> send = kafkaproducer.send(record);
            
            // 异步回调拿结果，假如下面有其他业务逻辑可以先行处理。比如扣减库存。
        /*kafkaproducer.send(record, new callback() {
            public void oncompletion(recordmetadata metadata, exception exception) {
            
            }
        });*/
            // todo 扣减库存的逻辑
            
            // 拿到发送者的相关信息
            recordmetadata recordmetadata = send.get();
            string sendtargettopic = recordmetadata.topic();
            long sendtargetoffset = recordmetadata.offset();
            int sendtargetpartition = recordmetadata.partition();
            
            system.out.println(
                    "发送的主题：" + sendtargettopic + ", 偏移量：" + sendtargetoffset + ", 分区：" + sendtargetpartition);
        }
        
        kafkaproducer.flush();
        // 优雅关闭，关闭之后kafka会有个心跳机制，默认10秒，
        // 如果还没有连接上就认为当前生产者宕机了，然后重新进行分区负载均衡
        kafkaproducer.close();
    }
}

public class consumer {
    
    public static void main(string[] args) {
        properties properties = new properties();
        properties.setproperty(consumerconfig.bootstrap_servers_config, "ip:9092,ip:9093,ip");
        // 反序列化器
        properties.setproperty(consumerconfig.key_deserializer_class_config, stringdeserializer.class.getname());
        // 反序列化器
        properties.setproperty(consumerconfig.value_deserializer_class_config, stringdeserializer.class.getname());
        // 关闭自动提交，默认值为true。true表示自动提交，也就是消费过的消息不再重复消费。false会一直重复消费。
        // 配置true的话下一次消费会从最后消费的下一个偏移量去进行消费。
        properties.setproperty(consumerconfig.enable_auto_commit_config, "false");
        // 自动提交的间隔时间
        properties.setproperty(consumerconfig.auto_commit_interval_ms_config, "1000");
        
        // 定义消费者群组
        properties.setproperty(consumerconfig.group_id_config, "1111");
        
        // 最小拉取消息的大小，默认1字节，一般配合下面的时间参数一起使用
        //        properties.setproperty(consumerconfig.fetch_min_bytes_config,"1");
        // 最久等待数据时间
        //        properties.setproperty(consumerconfig.fetch_max_wait_ms_config,"500");
        
        //最大拉取消息的大小，一般也是配合时间使用,默认50mb（50 * 1024 * 1024）
        //        properties.setproperty(consumerconfig.fetch_max_bytes_config, "50");
        // 从每个分区里面所能读取到的最大字节数，默认为1mb（1 * 1024 * 1024）
        //        properties.setproperty(consumerconfig.max_partition_fetch_bytes_config,"1");
        // 消费者定时发送心跳/多久没有发送心跳kafka判断消费者死亡，默认值为10秒（10000ms，单位毫秒），一般配合下面参数使用
        //        properties.setproperty(consumerconfig.session_timeout_ms_config, "10000");
        // 发送心跳的间隔时间，默认3秒（3000ms，单位也是毫秒）
        //        properties.setproperty(consumerconfig.heartbeat_interval_ms_config, "1000");
        
        kafkaconsumer<string, string> consumer = new kafkaconsumer<string, string>(properties);
        
        string topicname = "my-replication-topic";
        // 绑定主题。注意：subscribe和assign方法只能是用一个
//        kafkaconsumer.subscribe(collections.singletonlist(topicname));
        // 指定分区消费
//        kafkaconsumer.assign(collections.singletonlist(new topicpartition(topicname, 0)));
        
        // 从头开始消费，对应from-beginning命令
//        kafkaconsumer.assign(collections.singletonlist(new topicpartition(topicname, 0)));
//        kafkaconsumer.seektobeginning(collections.singletonlist(new topicpartition(topicname, 0)));
        
        // 指定offset消费
//        kafkaconsumer.assign(collections.singletonlist(new topicpartition(topicname, 0)));
//        kafkaconsumer.seek(new topicpartition(topicname, 0), 10);
        
        // 消费1小时之前的数据
        list<partitioninfo> partitions = consumer.partitionsfor(topicname);
        long fetchdatetime = new date().gettime() - 60 * 60 * 1000;
        map<topicpartition, long> partitionlongmap = new hashmap<>(10);
        for (partitioninfo partitioninfo : partitions) {
            partitionlongmap.put(new topicpartition(topicname, partitioninfo.partition()), fetchdatetime);
        }
    
        // 根据时间戳找分区的偏移量，并从该偏移量往后面去进行消费
        map<topicpartition, offsetandtimestamp> map = consumer.offsetsfortimes(partitionlongmap);
        for (map.entry<topicpartition, offsetandtimestamp> entry : map.entryset()) {
            topicpartition topicpartition = entry.getkey();
            offsetandtimestamp offsetandtimestamp = entry.getvalue();
            if (topicpartition == null || offsetandtimestamp == null) {
                continue;
            }
        
            // 得到分区
            int partition = topicpartition.partition();
            // 得到偏移量
            long offset = offsetandtimestamp.offset();
            system.out.println("分区：" + partition + "，偏移量：" + offset);
    
            // 最终还是根据分区和偏移量消费消息
            consumer.assign(collections.singletonlist(topicpartition));
            consumer.seek(topicpartition, offset);
        }
        
        
        try {
            //拉取消息
            while (true) {
                // 参数为每间隔多久拉取一次
                consumerrecords<string, string> consumerrecords = consumer.poll(duration.ofmillis(1000));
                for (consumerrecord<string, string> consumerrecord : consumerrecords) {
                    //偏移量是根据同一个分区里面的偏移量递增的
                    system.out.println(
                            "消息所在分区：" + consumerrecord.partition() + "，消息的偏移量：" + consumerrecord.offset() + "，key:"
                                    + consumerrecord.key() + ",value：" + consumerrecord.value());
                }
                
                // 同步提交偏移量，当前线程阻塞，直至消息提交完成之后处理后面的业务逻辑。
                consumer.commitsync();
    
                // 异步提交偏移量（可能会有消息丢失的情况）。也就是不管当前消费的消息是否提交了offset，都不会阻塞，可以继续处理后面的业务逻辑。
                /*kafkaconsumer.commitasync(new offsetcommitcallback() {
                    @override
                    public void oncomplete(map<topicpartition, offsetandmetadata> offsets, exception exception) {
                        if (null != exception) {
                            system.out.println("error offset:" + offsets);
                            system.out.println("error stacktrace:" + exception.getstacktrace());
                        }
                    }
                });*/
            }
        } catch (exception e) {
            e.printstacktrace();
        } finally {
            consumer.close();
        }
        
    }
}

<?xml version="1.0" encoding="utf-8"?>
<project xmlns="http://maven.apache.org/pom/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/xmlschema-instance"
         xsi:schemalocation="http://maven.apache.org/pom/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelversion>4.0.0</modelversion>
    <groupid>com.tqz</groupid>
    <artifactid>kafka</artifactid>
    <version>1.0-snapshot</version>
    <parent>
        <groupid>org.springframework.boot</groupid>
        <artifactid>spring-boot-starter-parent</artifactid>
        <version>2.4.1</version>
        <relativepath/>
    </parent>
    <dependencies>
        <dependency>
            <groupid>org.springframework.boot</groupid>
            <artifactid>spring-boot-starter-web</artifactid>
        </dependency>
        <dependency>
            <groupid>org.apache.kafka</groupid>
            <artifactid>kafka-clients</artifactid>
            <version>2.4.1</version>
        </dependency>
        <dependency>
            <groupid>org.springframework.kafka</groupid>
            <artifactid>spring-kafka</artifactid>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupid>org.apache.maven.plugins</groupid>
                <artifactid>maven-compiler-plugin</artifactid>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

spring.application.name=kafka-service
### 服务端地址
spring.kafka.bootstrap-servers=ip:9092,ip:9093,ip:9094
### 生产者配置
### 等leader写成功，不要等follower写成功就可以发送下一条消息
spring.kafka.producer.acks=1
#重试次数
spring.kafka.producer.retries=3
### kafka会从本地缓冲区取数据，批量发送到broker，设置批量发送的大小，默认值是16384，即16kb。也就是说一个batch满了16kb就发送出去。
spring.kafka.producer.batch-size=16384
### 生产者可用于缓冲等待发送到服务器的记录的总内存字节数，设置了该缓冲区，消息会先缓存到本地，可以提高消息发送的性能。默认值是32mb，即 33554432
spring.kafka.producer.buffer-memory=33554432
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.stringserializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.stringserializer
### 消费者配置
### 消费者组名
spring.kafka.consumer.group-id=mygroup
### 是否自动提交，true表示自动提交。false关闭自动提交，会一直重复消费。
spring.kafka.consumer.enable-auto-commit=false
### 当kafka中没有初始偏移量或者服务器上不再存在当前偏移量时该怎么办，默认值为latest，表示自动将偏移重置为最新的偏移量。可选的值为latest, earliest, none
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.stringdeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.stringdeserializer
spring.kafka.listener.ack-mode=manual_immediate
# record 当每一条记录被消费者监听器（listenerconsumer）处理之后提交
# batch 当每一批poll()的数据被消费者监听器（listenerconsumer）处理之后提交
# time 当每一批poll()的数据被消费者监听器（listenerconsumer）处理之后，距离上次提交时间大于time时提交
# count 当每一批poll()的数据被消费者监听器（listenerconsumer）处理之后，被处理record数量大于等于count时提交
# time 当每一批poll()的数据被消费者监听器（listenerconsumer）处理之后，被处理record数量大于等于count时提交
# count_time time或count有一个条件满足时提交
# manual 当每一批poll()的数据被消费者监听器（listenerconsumer）处理之后, 手动调用acknowledgment.acknowledge()后提交
# manual_immediate 手动调用acknowledgment.acknowledge()后立即提交

/**
 * <p>发送消息的控制器
 *
 * @author tianqingzhao
 * @since 2021/10/28 13:50
 */
@restcontroller
public class producercontroller {
    
    @autowired
    private kafkatemplate<string, string> kafkatemplate;
    
    private static final string topic_name = "my-replication-topic";
    
    @requestmapping("send")
    public long send(string key, string data) throws executionexception, interruptedexception {
        listenablefuture<sendresult<string, string>> result = kafkatemplate.send(topic_name, 0, key, data);
        
        sendresult<string, string> sendresult = result.get();
        recordmetadata metadata = sendresult.getrecordmetadata();
        
        string msg = "主题：" + metadata.topic() + "，分区：" + metadata.partition() + "，偏移量：" + metadata.offset();
        system.out.println(msg);
        
        return system.currenttimemillis();
    }
    
}

/**
 * <p>kafka消费者
 *
 * @author tianqingzhao
 * @since 2021/10/28 13:59
 */
@component
public class kafkaconsumer {
    
    /**
     * 一个消费者可以接收多个topic
     *
     * @param record 数据
     * @param ack    确认机制
     */
    @kafkalistener(groupid = "mygroup", topicpartitions = {@topicpartition(topic = "topic1", partitions = "0"),
            @topicpartition(topic = "my-replication-topic", partitions = {"0", "1"})})
    public void kafkaconsumer(consumerrecord<string, string> record, acknowledgment ack) {
        string key = record.key();
        string value = record.value();
        system.out.println(key + "===" + value);
        
        // 手动提交offset
        ack.acknowledge();
    }
}

hash(consumergroupid) % __consumer_offsets主题的分区数

# 更改zookeeper的连接地址
kafka-manager.zkhosts="localhost:2181"

./bin/kafka-manager &

 properties props = new properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("transactional.id", "my-transactional-id");
 producer<string, string> producer = new kafkaproducer<>(props, new stringserializer(), new stringserializer());

 // 初始化
 producer.inittransactions();

 try {
     // 开启
     producer.begintransaction();
     for (int i = 0; i < 100; i++)
         producer.send(new producerrecord<>("my-topic", integer.tostring(i), integer.tostring(i)));
     // 提交
     producer.committransaction();
 } catch (producerfencedexception | outofordersequenceexception | authorizationexception e) {
     // we can't recover from these exceptions, so our only option is to close the producer and exit.
     producer.close();
 } catch (kafkaexception e) {
     // for all other exceptions, just abort the transaction and try again.
     // 回滚
     producer.aborttransaction();
 }
 producer.close();