kafka

kafka 1.概念

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

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

下载好之后，通过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目录中，启动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

• 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的地址
        
        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";
        
//        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());
        }
    }

}