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() {@Overridepublic 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() {@Overridepublic 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));// 创建AlterConfigsOptionsAlterConfigsOptions 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);}}
}