Kafka 生产者和消费者实例

爱被打了一巴掌 2024-04-06 10:11 82阅读 0赞

> **目录**
> 
> 一、基于命令行使用Kafka
> 
> 二、创建一个名为“itcasttopic”的主题
> 
> ①、创建生产者
> 
> ②、创建消费者
> 
> ③、测试发送数据
> 
> 三、基于Java API方式使用Kafka
> 
> ①、创建工程添加依赖
> 
> ②、编写生产者客户端
> 
> ③ 、配置环境
> 
> ④、编写消费者客户端
> 
> ⑤、再运行KafkaConsumerTest程序 编辑编辑
> 
> ⑥、再回到KafkaProducerTest.java运行该程序
> 
> --------------------

### **一、基于命令行使用Kafka** ###

**类似scala，mysql等，命令行是初学者操作Kafka的基本方式，kafka的模式是生产者消费者模式，他们之间通讯是通过，一个公共频道完成**

### **二、创建一个名为“itcasttopic”的主题** ###

**kafka-topics.sh --create --topic itcasttopic --partitions 3 --replication-factor 2 --zookeeper master:2181,slave1:2181,slave2:2181**

**--create --topic itcasttopic: 创建主题名称是 itcasttopic**

**--partitions 3 ： 分区数是3**

**--replication-factor 2：副本数是 2**

**--zookeeper master:2181,slave1:2181,slave2:2181 ： zookeeper:服务的IP地址和端口**

![1b35e7041cd746459f7b9bcc011850f6.png][]

**\#****\#****删除主题\#****\#**

**$** **bin/kafka-topics.sh --delete -zookeeper master:2181,slave1:2181,slave2:2181 --topic itcasttopic**

#### **①、创建生产者** ####

**kafka-console-producer.sh** **--broker-list** **master****:9092,****slave1****:9092,****slave2****:9092****--topic itcasttopic**

![e1138514f25c4465a5111fd5a02611db.png][]

**（上面是等待输入光标在闪烁）**

![83087331134e44b38cd72745454d087f.png][]

**转换到slave****1**

#### **②、创建消费者** ####

**kafka-console-consumer.sh** **--from-beginning --topic itcasttopic****--bootstrap-server** **master****:90**

![ce761842a2d84647976512dd17bd843f.png][]

#### **③、测试发送数据** ####

**生产发送数据**

![ea666802b2d2474cacf525450d3c5205.png][]

**消费接收数据**

![2e9816c7bdc142b3a8bfca89a486984b.png][]

### **三、基于Java API方式使用Kafka** ###

![2ab018b696754842a27c5e791c26f33d.png][]

![560f905f3548454a828ed3c4898d1f1f.png][]

**修改配置：**

![7d791f05e2a849c38c3ee4b778d10131.png][]

![a09d2778cfbb47aead16d8bf74d02f3e.png][]

#### **①、创建工程添加依赖** ####

在工程里面的pom.xml文件添加Kafka依赖

**（Kafka依赖需要与虚拟机安装的****K****afka版本保持一致）**

<properties>
    <scala.version>2.11.8</scala.version>
    <hadoop.version>2.7.4</hadoop.version>
    <spark.version>2.3.2</spark.version>
    </properties>
    
    
    
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
    
    
    
    
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.0.0</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <version>2.0.0</version>
    </dependency>

#### **②、编写生产者客户端** ####

在工程的java目录下创建KafkaProducerTest文件

![5e0dada88c894b37872116cb5afc4f84.png][]

import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import java.util.Properties;
    
    public class KafkaProducerTest {
        public static void main(String[] args){
            Properties props = new Properties();
            //
            props.put("bootstrap.servers","master:9092,slave1:9092,slave2: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");
            //
            KafkaProducer<String,String> producer=new KafkaProducer<String, String>(props);
            for (int i=0; i<50; i++){
                producer.send(new ProducerRecord<String, String>("itcasttopic",Integer.toString(i),"hello world [2] -"+i));
            }
            producer.close();
        }

\}![edcb457a6e69430d975b24ad6ddf82d3.png][]

**Slave1****上出现的结果**

![3920872aaecf49f2b6506b6b07e69ea1.png][]

#### **③ 、配置环境** ####

![c26e97e27cb14e5dad208c2ef1ccab15.png][]

![80160532d6a3427d8b1ec7a4464bd345.png][]

#### **④、编写消费者客户端** ####

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.clients.producer.Callback;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.clients.producer.RecordMetadata;
    
    import java.util.Arrays;
    import java.util.Properties;
    
    public class KafkaConsumerTest {
        public static void main(String[] args) {
            // 1、准备配置文件
            Properties props = new Properties();
            // 2、指定Kafka集群主机名和端口号
            props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
            // 3、指定消费者组ID，在同一时刻同一消费组中只有一个线程可以去消费一个分区数据，不同的消费组可以去消费同一个分区的数据。
            props.put("group.id", "itcasttopic");
            // 4、自动提交偏移量
            props.put("enable.auto.commit", "true");
            // 5、自动提交时间间隔，每秒提交一次
            props.put("auto.commit.interval.ms", "1000");
            props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props);
            // 6、订阅数据，这里的topic可以是多个
            kafkaConsumer.subscribe(Arrays.asList("itcasttopic"));
            // 7、获取数据
            while (true) {
                //每隔100ms就拉去一次
                ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("topic = %s,offset = %d, key = %s, value = %s%n", record.topic(), record.offset(), record.key(), record.value());
                }
            }
        }
    
    }

运行KafkaP roducerTest程序

![83ffa7dd0cb14c72b442a98b7f45f136.png][]

#### **⑤、再运行KafkaConsumerTest程序** ![790caac612934c6eb173c4d117372979.png][] ####

#### **⑥、再回到KafkaProducerTest.java运行该程序** ####

(查看KafkaConsumerTest的运行框)由以下图可以看出生产者生产消息成功被终端消费

![5b69e55e64c34dc69f6fccb1f8059e32.png][]

[1b35e7041cd746459f7b9bcc011850f6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/763e069eb4d14b44be93a5004b0d14b2.png
[e1138514f25c4465a5111fd5a02611db.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/5190e2a312f14f6992f553b3cfef7a25.png
[83087331134e44b38cd72745454d087f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/82bd1fe9d92a4722b831a58c60fa4fca.png
[ce761842a2d84647976512dd17bd843f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/3aabf80725d24b6abc320dda1050de81.png
[ea666802b2d2474cacf525450d3c5205.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/57a53ad78d0e4fb990b834c04246aef9.png
[2e9816c7bdc142b3a8bfca89a486984b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/f3481f4aef214b31a8a6bc5dfe22fa2c.png
[2ab018b696754842a27c5e791c26f33d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/6871c26287e54d0397ef06107c6a758d.png
[560f905f3548454a828ed3c4898d1f1f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/0498a5fa487f49a0ab44bab94e7980ee.png
[7d791f05e2a849c38c3ee4b778d10131.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/9aac968666ce4c00ba6809cbbb02e0e7.png
[a09d2778cfbb47aead16d8bf74d02f3e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/bbf974088f724261a4673d95067e23e7.png
[5e0dada88c894b37872116cb5afc4f84.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/ace4e56cbda64b73a159bd35d9fe129c.png
[edcb457a6e69430d975b24ad6ddf82d3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/45d9b32417e2448281574d4b4d5f140d.png
[3920872aaecf49f2b6506b6b07e69ea1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/fee6d57df6e542a5b66b565b0a661053.png
[c26e97e27cb14e5dad208c2ef1ccab15.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/536e2653543b4952846fef6b55d0e56e.png
[80160532d6a3427d8b1ec7a4464bd345.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/fd7af579fdf54fb69406780eb3590a8e.png
[83ffa7dd0cb14c72b442a98b7f45f136.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/b5deaf28a0b741308072bf8613b3c710.png
[790caac612934c6eb173c4d117372979.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/27a0e67e447647b2a129b527f425d7b5.png
[5b69e55e64c34dc69f6fccb1f8059e32.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/866bb10d1488443094d87330a6f0fafd.png