用java（idea）来测试kafka中的生产者和消费者

怼烎@ 2023-06-16 07:52 73阅读 0赞

**一、检查环境**  
先把linux中的zookeeper和kafka都安装，并且启动完成。  
![在这里插入图片描述][20191125104030928.png]  
二、idea环境  
打开idea并且创建maven项目。（不会的自己去百度）  
**pow.xml**：

<dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.12</artifactId>
            <version>1.0.0</version>
            <scope>provided</scope>
        </dependency>
    
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>1.0.0</version>
        </dependency>
    
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
            <version>1.0.0</version>
        </dependency>
    </dependencies>
        <build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.2</version>
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                    </configuration>
                </plugin>
            </plugins>
        </build>

**生产者：**

import org.apache.kafka.clients.producer.*;
    import org.apache.kafka.common.serialization.StringSerializer;
    import java.util.Properties;
    /** * @ClassName: ProducerDemo * @author: zhl * @date: 2019/11/20 9:33 */
    public class ProducerDemo  {   public static void main(String[] args) { 
                String topicName = "HelloWorld";//这个是创建好的topic
    
                // create instance for properties to access producer configs
                Properties props = new Properties();
    
                //这里是填你linux的ip地址:kafka的端口号
                props.put("bootstrap.servers", "192.168.137.128:9092");
    
                //Set acknowledgements for producer requests.
                props.put("acks", "all");
    
                //If the request fails, the producer can automatically retry,
                props.put("retries", 0);
    
                //Specify buffer size in config
                props.put("batch.size", 16384);
    
                //Reduce the no of requests less than 0
                props.put("linger.ms", 1);
    
                //The buffer.memory controls the total amount of memory available to the producer for buffering.
                props.put("buffer.memory", 33554432);
    
                props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
                props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
                Producer<String, String> producer = new KafkaProducer<String, String>(props);
    
                for (int i = 0; i < 10; i++)
                    producer.send(new ProducerRecord<String, String>(topicName, Integer.toString(i), Integer.toString(i)));
                System.out.println("Message sent successfully");
                producer.close();
            }
    }

**消费者：**

import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.common.serialization.StringDeserializer;
    import java.util.Arrays;
    import java.util.Properties;
    
    /** * @ClassName: ConsumerDemo * @author: zhl * @date: 2019/11/20 9:33 */
    public class ConsumerDemo { 
    public static void main(String[] args) throws Exception
    { 
    
        //Kafka consumer configuration settings
        String topicName = "HelloWorld";//创建好的topic
        Properties props = new Properties();
    
        props.put("bootstrap.servers", "192.168.137.128: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", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
    
        //Kafka Consumer subscribes list of topics here.
        consumer.subscribe(Arrays.asList(topicName));
    
        //print the topic name
        System.out.println("Subscribed to topic " + topicName);
        int i = 0;
    
        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());
        }
    }
    }

**备注：**  
1，如果，不会创建topic的话，看这篇文章，这里面包括topic的查看，销毁，创建，打开生产者和消费者的命令都有[kafka命令大全][kafka]  
2，先执行ConsumerDemo（消费者），再执行ProducerDemo（生产者）。因为执行好了消费者之后，你才能去生产，执行了之后，消费者就相当于再等待生产。执行了之后的界面如下：  
生产者页面：  
![在这里插入图片描述][20191125104922488.png]  
消费者页面：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjM2OTQxOA_size_16_color_FFFFFF_t_70]  
此时打开linux里面的消费者，然后再执行生产者的代码，也能看到消费：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjM2OTQxOA_size_16_color_FFFFFF_t_70 1]  
**注：如果执行了消费者，再执行生产者之后，发现消费者没数据：接着你就来看这篇文章：[测试生产者和消费者代码，发现消费者并没有输出数据，怎么办？（Kafka服务器允许客户端远程连接）][Kafka]**

[20191125104030928.png]: https://img-blog.csdnimg.cn/20191125104030928.png
[kafka]: https://blog.csdn.net/weixin_42369418/article/details/103125952
[20191125104922488.png]: https://img-blog.csdnimg.cn/20191125104922488.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjM2OTQxOA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191125104938205.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjM2OTQxOA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjM2OTQxOA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/201911251050180.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjM2OTQxOA==,size_16,color_FFFFFF,t_70
[Kafka]: https://blog.csdn.net/weixin_42369418/article/details/103234294