Kafka学习（5）- kafka使用Java开发-1-入门

柔情只为你懂 2022-03-27 01:44 268阅读 0赞

**目录：**

1.  官方文档
2.  测试
    
    1.  生产者代码
    2.  消费者代码
3.  问题：参数配置能不能换一种简便的方法来替代？

### 一.官方文档 ###

地址：[http://kafka.apachecn.org/][http_kafka.apachecn.org]，找到安装的版本的对应文档，本人是用的版本是0.10.0.1

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 1][]

### 二.测试 ###

**1.创建项目**  
![2019011118214038.png][]

**2.添加依赖包  
pom.xml**

<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.yiguang.kafkaTest</groupId>
      <artifactId>KafkaTest</artifactId>
      <version>0.0.1-SNAPSHOT</version>
      
      <dependencies>
        <dependency>
          <groupId>junit</groupId>
          <artifactId>junit</artifactId>
          <version>3.8.1</version>
          <scope>test</scope>
        </dependency>
        
    	<dependency>
    	    <groupId>org.apache.kafka</groupId>
    	    <artifactId>kafka-clients</artifactId>
    	    <version>0.10.0.0</version>
    	</dependency>
      </dependencies>
    </project>

**3.生产者代码**

package kafkaTest;
    
    import java.util.Properties;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    public class ProducerSend {
    	public static void main(String args[]) {
    		
    		//1.参数配置：端口、缓冲内存、最大连接数、key序列化、value序列化等等（不是每一个非要配置）
    		Properties props=new Properties();
    		props.put("bootstrap.servers", "localhost:9092");
    		props.put("acks", "all");
    		props.put("retries", 0);
    		props.put("batch.size", 16384);
    		props.put("linger.ms", 1);
    		props.put("buffer.memory", 33554432);
    		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    		
    		//2.创建生产者对象，并建立连接
    		Producer<String, String> producer = new KafkaProducer<>(props);
    		
    		//3.在my-topic主题下，发送消息
    		for(int i = 0; i < 100; i++) {
    			producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));
    		}
    		
    		//4.关闭
    		producer.close();
    		
    	}
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 2][]

**4.消费者代码**

package kafkaTest;
    
    import java.util.Arrays;
    import java.util.Properties;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    
    public class ConsumerReceive {
    	public static void main(String args[]) {
    		
    	    //1.参数配置:不是每一非得配置
    	    Properties props = new Properties();
    	    props.put("bootstrap.servers", "localhost:9092");
    	    props.put("auto.commit.interval.ms", "1000");
    	    //因为每一个消费者必须属于某一个消费者组，所以必须还设置group.id
    	    props.put("group.id", "test");
    	    props.put("enable.auto.commit", "true");
    	    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");
    	    
    	    //2.创建消费者对象，并建立连接
    	    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    	    
    	    //3.设置从"my-topic"主题下拿取数据
    	    consumer.subscribe(Arrays.asList("my-topic"));
    	    
    	    //4.消费数据
    	    while (true) {
    	    	//阻塞时间，从kafka中取出100毫秒的数据，有可能一次性去除0-n条
    	        ConsumerRecords<String, String> records = consumer.poll(100);
    	        //遍历
    	        for (ConsumerRecord<String, String> record : records)
    	        	//打印结果
                            //System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
    	        	System.out.println("消费者消费的数据为："+record.value());
    	    }
    	}
    }

**5.开启zookeeper和kafka**

开启zookeepr  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 3][]

开启kafka：  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 4][]

**6.运行**

1.  运行生产者代码
2.  运行list（列表）命令来查看这个topic：>bin/windows/kafka-topics.bat --list --zookeeper localhost:2181  
    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 5][]  
    可以看到已经有这个主题了
3.  运行消费者代码，可以看到会消费生产者生产的消息  
    因为只有my-topic主题下只有默认的一个分区，所以消息都放在此分区下面，所以消费的数据都是有序的，如果有该主题下有多个分区，那么分区与分区之间就不是有序的了  
    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 6][]![20190111181756248.png][]

### 三.参数配置能不能换一种简便的方法？ ###

看下一节

[http_kafka.apachecn.org]: http://kafka.apachecn.org/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70]: /images/20220327/1affe317be56493da7706d4b8aac89b0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 1]: /images/20220327/8f66bc20097340aeb7857b8777ab78db.png
[2019011118214038.png]: /images/20220327/a6fe83a091de48aba067d95b7f2ba3e6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 2]: /images/20220327/2fbcab9420b1480fbad00f5af6096336.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 3]: /images/20220327/0a3553c9cb9b486e8be08b4626027e9a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 4]: /images/20220327/b6f4bfd542b44b94bb5b648168fdb92a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 5]: /images/20220327/ed7a584272cf47ed8d6a97c13221b228.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20190111181737140.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw,size_16,color_FFFFFF,t_70
[20190111181756248.png]: /images/20220327/8fd1351de6f64cc694b204c31aac1bfe.png