kafka的安装和简单使用

「爱情、让人受尽委屈。」 2023-07-04 12:40 7阅读 0赞

### 一.安装 ###

#### 1.zookeeper ####

kafka依赖于Zookeeper，所以我们下载解压 `zookeeper-3.4.14`  
进入conf目录，复制`zoo_sample.cfg`并重命名为`zoo.cfg`,打开文件将dataDir修改为你创建的某个目录,同时我们可以知道zookeeper的默认端口是`2181`  
![在这里插入图片描述][20200210231839303.png]  
进入bin目录，启动zookeeper.

./zkServer.sh start /opt/zookeeper/zookeeper-3.4.14/conf/zoo.cfg

输入`jps`,看到`QuorumPeerMain`代表启动成功  
![在这里插入图片描述][20200210232130411.png]

#### 2.kafka ####

下载解压`kafka_2.12-2.1.1`,进入config目录，打开`server.properties`  
将log.dirs修改为自己创建的某个目录(这里要注意虽然它名字叫做log.dir，实际上这个设置是指kafka消息存储的位置),kafka启动的默认端口是`9092`  
![在这里插入图片描述][20200210232530858.png]  
进入bin目录，启动kafka.

./kafka-server-start.sh -daemon --config /opt/kafka/kafka_2.12-2.1.1/config/server.properties

输入`jps`,看到`Kafka`代表启动成功.  
![在这里插入图片描述][20200210232958117.png]

### 二.kafka的命令行操作 ###

**1.创建topic**

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

**2.查看topic列表**

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

**3.查看详细topic**

./kafka-topics.sh --describe --zookeeper localhost:2181 --topic first

**4.删除topic**

./kafka-topics.sh --delete --zookeeper localhost:2181 --topic first

**5.创建生产者**

./kafka-console-producer.sh --topic first --broker-list 101.132.172.64:9092

![在这里插入图片描述][20200211003847974.png]  
**6.创建消费者**

./kafka-console-consumer.sh --topic first --bootstrap-server 101.132.172.64:9092

![在这里插入图片描述][20200211003910128.png]  
**老版本创建消费者连接的是zookeeper**

./kafka-console-consumer.sh --topic first --zookeeper localhost:2181

### 三.使用kafka API ###

#### 1.添加依赖`pom.xml` ####

<dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
                <version>2.1.0</version>
            </dependency>

#### 2.创建生产者 ####

`MyProducer,java`

package producer;
    
    import org.apache.kafka.clients.producer.*;
    
    import java.util.Properties;
    
    public class MyProducer { 
        public static void main(String[] args) { 
            //创建kafka生产者配置信息
            Properties properties = new Properties();
            //连接kafka
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "101.132.172.64:9092");
            //ack应答级别
            properties.put(ProducerConfig.ACKS_CONFIG, "all");
            //重试次数
            properties.put(ProducerConfig.RETRIES_CONFIG, 1);
            //批次大小
            properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
            //等待时间
            properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
            //RecordAccumulator缓冲区的大小
            properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 33554432);
            //Key,Value的序列化类
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    
            KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
            for (int i = 0; i < 10; i++) { 
                producer.send(new ProducerRecord<String, String>("first", "helloworld---" + i), new Callback() { 
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) { 
                        //成功返回recordMetadata，失败返回Exception
                        if (e == null) { 
                            System.out.println("topic: " + recordMetadata.topic());
                        }
                    }
                });
            }
            producer.close();
        }
    }

可通过命令行运行消费者，查看结果.  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2_size_16_color_FFFFFF_t_70]  
**我是在阿里云的服务器运行的kafka，然后在我本地windows跑的程序，程序不报错，但是消费者没有任何反应**  
解决方法:  
1.登录阿里云服务器，配置服务器实例安全访问规则，把`9092`,`2181`端口都给开放  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2_size_16_color_FFFFFF_t_70 1]  
2.修改之前`kafka/bin/server.properties`配置文件

host.name=172.19.206.17  #内网ip
    advertised.host.name=101.132.172.64  #公网ip
    advertised.port=9092

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2_size_16_color_FFFFFF_t_70 2]

#### 3.创建消费者 ####

`MyConsumer.java`

package consumer;
    
    import org.apache.kafka.clients.consumer.ConsumerConfig;
    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.Collections;
    import java.util.Properties;
    
    public class MyConsumer { 
        public static void main(String[] args) { 
            //创建kafka消费者配置信息
            Properties properties = new Properties();
            //连接Kafka
            properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "101.132.172.64:9092");
            //开启自动提交
            properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
            //自动提交延迟
            properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
            //消费者组
            properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
            //key,value反序列化
            properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
            properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
            //订阅主题，可同时订阅多个
            consumer.subscribe(Collections.singletonList("first"));
            while (true) { 
                ConsumerRecords<String, String> consumerRecords = consumer.poll(100);
                for (ConsumerRecord record : consumerRecords) { 
                    System.out.println("key: " + record.key() + " " + "value: " + record.value());
                }
            }
        }
    }

开启消费者，再启动上面的生产者查看结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2_size_16_color_FFFFFF_t_70 3]

[20200210231839303.png]: https://img-blog.csdnimg.cn/20200210231839303.png
[20200210232130411.png]: https://img-blog.csdnimg.cn/20200210232130411.png
[20200210232530858.png]: https://img-blog.csdnimg.cn/20200210232530858.png
[20200210232958117.png]: https://img-blog.csdnimg.cn/20200210232958117.png
[20200211003847974.png]: https://img-blog.csdnimg.cn/20200211003847974.png
[20200211003910128.png]: https://img-blog.csdnimg.cn/20200211003910128.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2020021100441995.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/2020021100494176.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200211005200270.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200211005409842.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2,size_16,color_FFFFFF,t_70