Kafka

消息中间件

区别于rabbitmq,kafka更适用于量级较大的数据(100w级),主要在大数据领域使用

Kafka介绍

一个分布式流媒体平台,类似于消息队列或企业消息传递系统

Kafak的结构如下

producer:发布消息的对象

topic:Kafak将消息分门别类,每类的消息称为一个主题(Topic)

consumer:订阅主题并处理发布的消息的对象称为主题消费者

broker:已发布的消息保存在一组服务器中,称为kafka集群,集群中的每一个服务器都是一个代理(broker)

消费者可以订阅一个或者多个的主题,并从broker中拉取数据,从而消费这些已发布的消息.

Kafka对zookeeper强依赖,需要使用zk进行分区的负载均衡以及节点的注册

docker安装zk和kafka

docker pull zookeeper:3.4.14
docker run -d --name zookeeper --restart=always -p 2181:2181 zookeeper:3.4.14
docker pull wurstmeister/kafka:2.12-2.3.1
docker run -d --name kafka \
--env KAFKA_ADVERTISED_HOST_NAME=192.168.200.130 \
--env KAFKA_ZOOKEEPER_CONNECT=192.168.200.130:2181 \
--env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.200.130:9092 \
--env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
--env KAFKA_HEAP_OPTS="-Xmx256M -Xms256M" \
--restart=always \
--net=host wurstmeister/kafka:2.12-2.3.1

Kafka入门

kafka的生产者消费者模型分为单播和多播

单播:同组内存在多个消费者,一个主题只能发送消息到同一个组内的一个消费者

多播:不同组的多个消费者,一个主题可以发送消息到不同组的多个不同消费者

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

Kafka的高可用设计(集群模式)

多个broker组成一个cluster集群

集群中的某一条机器宕机,其他机器上的broker依然能对外提供服务

Kafka的备份机制(republication)

消息的备份称为副本(replica)

分为两种副本

领导者副本(leader replica)

追随者副本(follower replica) 分为ISR 和 普通

同步方式:

领导者副本直接接收发布者的消息

随后领导者副本会同步将消息复制保存到ISR follower副本,异步将消息复制保存到普通follower副本

如果leader失效,需要选举出新的leader

依据以下原则选举:

选举优先从ISR中选举,因为ISR副本是同步的

如果ISR中没有生效的副本,就从普通中选举一个

如果所有副本都失效

可以等待ISR副本活过来保证数据完整性,也可以选举第一个活过来的保证数据可用性

Kafka的生产者详解

消息发送:

分为同步发送和异步发送

同步发送:

ProducerRecord<String,String> record = new ProducerRecord<String, String>("itheima-topic","100001","hello kafka");
RecordMetadata recordMetadata = producer.send(ProducerRecord).get();//获取发送的结果

异步发送:

传入一个callback对象处理回调结果

//异步消息发送
producer.send(ProducerRecord, new Callback() {
    @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        if(e != null){
            System.out.println("记录异常信息到日志表中");
        }
        System.out.println(recordMetadata.offset());
    }
});

生产者配置

消息确认配置acks

//ack配置  消息确认机制
prop.put(ProducerConfig.ACKS_CONFIG,"all");
acks = 0 //生产者不会等待任何来自服务器的响应
acks = 1 //集群leader收到消息后,生产者会接受一个来自服务器的响应
acks = -1/all(默认) //参与复制的所有生产者全部收到消息后,生产者才会收到响应

重试次数retries

//重试次数
prop.put(ProducerConfig.RETRIES_CONFIG,10);

消息压缩方式

消息默认不会被压缩

//数据压缩
prop.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"lz4");
| snappy | 占用较少的  CPU，  却能提供较好的性能和相当可观的压缩比， 如果看重性能和网络带宽，建议采用 |
| lz4    | 占用较少的 CPU， 压缩和解压缩速度较快，压缩比也很客观        |
| gzip   | 占用较多的  CPU，但会提供更高的压缩比，网络带宽有限，可以使用这种算法 |

Kafka消费者详解

消费者组:一个或者多个消费者组成的群体

topic会分发消息给每个消费者组中的一个消费者

消息有序性

跨分区的消息无法决定先后顺序

如果需要保证消息的有序性需要让多个消费者去监听同一个分区

提交和偏移量

消费者会在消费前或者消费后向__consumer_offset的特殊topic中发送偏移量,记录每个消息的处理进度

如果消费者崩溃或者新的消费者加入就会触发负载均衡

如果使用默认方式自动提交偏移量,就会在消费前直接自动提交偏移量,可能会出现重复消费或者漏消费的情况

所以我们需要使用手动提交的方式提交偏移量

手动提交(同步,会自动重试):

consumer.commitSync()

异步提交:

consumer.commitAsync(new OffsetCommitCallback() {
        @Override
        public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
            if(e!=null){
                System.out.println("记录错误的提交偏移量："+ map+",异常信息"+e);
            }
        }
    });

Spring继承Kafka

引入依赖

<!-- kafkfa -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
        <exclusions>
            <exclusion>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
            </exclusion>
        </exclusions>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
    </dependency>

Topic的创建和删改(可省略)

public static void createTopic(String topicName, int partitions, short replicas) throws Exception {
        NewTopic newTopic = new NewTopic(topicName, partitions, replicas);
        CreateTopicsResult topics = adminClient.createTopics(Collections.singleton(newTopic));
        topics.all().get();
        log.info("【{}】topic创建成功", topicName);
    }

    /**
     * @Title deleteTopic
     * @Description 删除topic
     * @param topicName  topic名称
     * @return void
     */
    public static void deleteTopic(String topicName) throws Exception {
        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Collections.singleton(topicName));
        deleteTopicsResult.all().get();
        log.info("【{}】topic删除成功", topicName);

    }

    /**
     * @Title updateTopicRetention
     * @Description 修改topic的过期时间
     * @param topicName  topic名称
     * @param ms  过期时间（毫秒值）
     * @return void
     */
    public static void updateTopicRetention(String topicName, String ms) throws Exception {
        ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
        ConfigEntry configEntry = new ConfigEntry(TopicConfig.RETENTION_MS_CONFIG, ms);
        Config config = new Config(Collections.singleton(configEntry));
        // 创建AlterConfigsOptions
        AlterConfigsOptions alterConfigsOptions = new AlterConfigsOptions().timeoutMs(10000);
        // 执行修改操作
        adminClient.alterConfigs(Collections.singletonMap(resource, config), alterConfigsOptions).all().get();
        log.info("【{}】topic过期时间设置完成，过期时间为：{}毫秒", topicName, ms);
    }

生产者发送消息

@Resource
private KafkaTemplate<String,String> kafkaTemplate;

kafkaTemplate.send("topic","test");

消费者消费消息

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

@Component
public class Listener {

    @KafkaListener(topics = "topic")
    public void onMessage(String message){
        if(!StringUtils.isEmpty(message)){
            System.out.println(message);
        }

    }
}