kafka

1.概念

kafka是由Scale语言编写,是一个分布式、分区的、多副本的、多订阅者,基于zookeeper协调的分布式日志系统(也可以当做MQ系统),其主要应用场景是:日志收集系统和消息系统。

2.kafka的安装

登录以下网站下载
http://kafka.apache.org/downloads.html

kafka_第1张图片

下载好之后,通过ftp上传至虚拟机上,使用命令tar -xzvf kafka_2.12-2.4.0.tgz 解压

然后修改配置文件

Zookeeper的配制文件是config/zookeeper.properties(一般不用修改)

kafka的配置文件在config/server.properties文件中,可以修改其中的参数(主要是修改监听端口),具体配置:https://blog.csdn.net/universsky2015/article/details/115234772

kafka_第2张图片

然后到达kafka目录中,启动zk和kafka 

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

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

3.kafka的基础架构

参考:https://blog.csdn.net/cao1315020626/article/details/112590786

Kafka的基础架构

  • Producer:消息生产者,向Kafka中发布消息的角色。
  • Consumer:消息消费者,即从Kafka中拉取消息消费的客户端。
  • Consumer Group:消费者组,消费者组则是一组中存在多个消费者,消费者消费Broker中当前Topic的不同分区中的消息,消费者组之间互不影响,所有的消费者都属于某个消费者组,即消费者组是逻辑上的一个订阅者。某一个分区中的消息只能够一个消费者组中的一个消费者所消费
  • Broker:经纪人,一台Kafka服务器就是一个Broker,一个集群由多个Broker组成,一个Broker可以容纳多个Topic。
  • Topic:主题,可以理解为一个队列,生产者和消费者都是面向一个Topic
  • Partition:分区,为了实现扩展性,一个非常大的Topic可以分布到多个Broker上,一个Topic可以分为多个Partition,每个Partition是一个有序的队列(分区有序,不能保证全局有序)
  • Replica:副本Replication,为保证集群中某个节点发生故障,节点上的Partition数据不丢失,Kafka可以正常的工作,Kafka提供了副本机制,一个Topic的每个分区有若干个副本,一个Leader和多个Follower
  • Leader:每个分区多个副本的主角色,生产者发送数据的对象,以及消费者消费数据的对象都是Leader。
  • Follower:每个分区多个副本的从角色,实时的从Leader中同步数据,保持和Leader数据的同步,Leader发生故障的时候,某个Follower会成为新的Leader。

上述一个Topic会产生多个分区Partition,分区中分为Leader和Follower,消息一般发送到Leader,Follower通过数据的同步与Leader保持同步,消费的话也是在Leader中发生消费,如果多个消费者,则分别消费Leader和各个Follower中的消息,当Leader发生故障的时候,某个Follower会成为主节点,此时会对齐消息的偏移量。

 注意:多个Kafka消费者要想同时消费相同Topic下的相同Partition的数据,则需要将这些Kafka消费者放到不同的消费者组中。

4.java使用kafka的具体实例

本例中未指定分区和Broker,均使用默认的。

参照:https://www.cnblogs.com/xuwujing/p/8371127.html

Maven的依赖

    
        
            org.apache.kafka
            kafka_2.11
            2.2.0
        
        
            org.apache.kafka
            kafka-clients
            2.2.0
        
        
            org.slf4j
            slf4j-nop
            1.7.2
        
    

 生产者:

package kafka;


import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProducerTest {
    public static void main(String args[]) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.174.103:9092");//kafka的地址
        /**
         * acks:消息的确认机制,默认值是0。
         * acks=0:如果设置为0,生产者不会等待kafka的响应。
         * acks=1:这个配置意味着kafka会把这条消息写到本地日志文件中,但是不会等待集群中其他机器的成功响应。
         * acks=all:这个配置意味着leader会等待所有的follower同步完成。这个确保消息不会丢失,除非kafka集群中所有机器挂掉。这是最强的可用性保证。
         */
        props.put("acks", "0");
        //配置为大于0的值的话,客户端会在消息发送失败时重新发送。
        props.put("retries", 0);
        //当多条消息需要发送到同一个分区时,生产者会尝试合并网络请求。这会提高client和生产者的效率
//        props.put("batch.size", 16384);
        //键序列化,默认org.apache.kafka.common.serialization.StringDeserializer。
        props.put("key.serializer", StringSerializer.class.getName());
        //值序列化,默认org.apache.kafka.common.serialization.StringDeserializer。
        props.put("value.serializer", StringSerializer.class.getName());
        KafkaProducer producer = new KafkaProducer(props);
        String topic = "KAFKA_TOPIC1";
        /**
         * 本例中未创建topic也可以使用
         * 创建topic的两中方法
         * 1.使用java创建如本例子中(详情:https://www.cnblogs.com/huxi2b/p/6592862.html)
         * 2.在服务器上使用命令创建
         * bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic test
         * 这条命令会创建一个名为test的topic,有3个分区,每个分区需分配3个副本。
         */
//        ZkUtils zkUtils = ZkUtils.apply("192.168.174.103:2181", 30000, 30000, JaasUtils.isZkSecurityEnabled());
//        // 创建一个单分区单副本名为KAFKA_TOPIC的topic
//        AdminUtils.createTopic(zkUtils, "KAFKA_TOPIC", 1, 1, new Properties(), RackAwareMode.Enforced$.MODULE$);
//        zkUtils.close();
        producer.send(new ProducerRecord(topic, "producer1", "hello kafka"));
        producer.close();

    }

}

消费者:

package kafka;


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;

public class KafkaConsumerTest {
    public static void main(String args[]) {
        String GROUPID = "groupA";
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.174.103:9092");
        props.put("group.id", GROUPID);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("auto.offset.reset", "earliest");
        //键反序列化
        props.put("key.deserializer", StringDeserializer.class.getName());
        //值反序列化
        props.put("value.deserializer", StringDeserializer.class.getName());
        KafkaConsumer consumer = new KafkaConsumer(props);
        String topic = "KAFKA_TOPIC1";
        consumer.subscribe(Arrays.asList(topic));
        ConsumerRecords msgList = consumer.poll(1000);
        for (ConsumerRecord record : msgList) {
            System.out.printf("offset = %d, key = %s, value = %s, topic = %s", record.offset(), record.key(), record.value(), record.topic());
        }
    }

}

你可能感兴趣的