欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

Kafka的安装与使用(单节点Kafka的安装)

程序员文章站 2024-01-08 13:27:34
...

在这里总结下Kafka的安装与使用

Kafka消息队列

官网下载安装包

http://archive.apache.org/dist/kafka/

官方文档教程

https://kafka.apachecn.org/quickstart.html

启动Kafka

在启动kafka之前,我们需要启动zookeeper,kafka内置了一个zookeeper节点

bin/zookeeper-server-start.sh config/zookeeper.properties

如果没有错误日志输出,表示启动正常
接下来去启动kafka

bin/kafka-server-start.sh config/server.properties

终端日志没有报错即是启动正常,接下来我们使用kafka模拟一次生产者与消费者通信的过程,我们创建一个订阅主题,他携带一个分区和副本

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

当终端出现创建成功时,我们就可以去创建一个生产者和消费者来发送消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

启动之后会出现一个交互的命令行,我们在里面输入一些消息
Kafka的安装与使用(单节点Kafka的安装)
同时启动消费者脚本,这里的--from-beginning表示获取从生产者队列启动开始时到现在队列中的数据

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

可以看到,消费者是能获取到数据的
Kafka的安装与使用(单节点Kafka的安装)

JavaApi操作Kafka

这里示范一下使用maven的Javaapi操作Kafka服务器,这里要保证Kafka和zookeeper是已经启动的
pom.xml

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

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka_2.11</artifactId>
  <version>0.9.0.0</version>
</dependency>

MyProducer.java

package cn.itcast.kafkaapi;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

public class MyProducer {
    private static KafkaProducer<String, String> producer;
    private final static String TOPIC = "demo";
    public MyProducer(){
        Properties props = new Properties();
        //这里填写自己的Kafka服务器ip地址与端口号
        props.put("bootstrap.servers", "node1: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");
        //设置分区类,根据key进行数据分区
        producer = new KafkaProducer<String, String>(props);
    }
    public void produce(){
        for (int i = 40;i<50;i++){
            String key = String.valueOf(i);
            String data = "hello kafka message:"+key;
            producer.send(new ProducerRecord<String, String>(TOPIC,key,data));
            System.out.println(data);
        }
        producer.close();
    }
    public static void main(String[] args) {
        new MyProducer().produce();
    }
}

MyConsumer.java

package cn.itcast.kafkaapi;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.log4j.Logger;

import java.util.Arrays;
import java.util.Properties;

public class MyConsumer {
    private static KafkaConsumer<String, String> consumer;
    private final static String TOPIC = "demo";
    public MyConsumer(){
        Properties props = new Properties();
        props.put("bootstrap.servers", "node1:9092");
        //每个消费者分配独立的组号
        props.put("group.id", "test2");
        //如果value合法,则自动提交偏移量
        props.put("enable.auto.commit", "true");
        //设置多久一次更新被消费消息的偏移量
        props.put("auto.commit.interval.ms", "1000");
        //设置会话响应的时间,超过这个时间kafka可以选择放弃消费或者消费下一条消息
        props.put("session.timeout.ms", "30000");
        //自动重置offset
        props.put("auto.offset.reset","earliest");
        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<String, String>(props);
    }

    public void consume(){
        consumer.subscribe(Arrays.asList(TOPIC));
        while (true) {
            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();
            }
        }
    }

    public static void main(String[] args) {
        new MyConsumer().consume();
    }
}

分别启动生产者和消费者,可以看到生产者写入的数据,已经被消费者拿到了
Kafka的安装与使用(单节点Kafka的安装)
Kafka的安装与使用(单节点Kafka的安装)

相关标签: 微服务

上一篇:

下一篇: