SpringBoot-Kafka使用（一）

短命女 2022-03-07 06:20 232阅读 0赞

## 一、简介 ##

## Kafka认识一下 ##

Kafka异军突起，是近来非常火热的一款消息中间件。消息中间件的作用非常多，常用作系统业务的解耦。例如最常听到的秒杀业务，我们也能使用消息中间件对业务进行解耦，用户发起秒杀请求后，系统首先会将该请求转发到中间件中，然后返回一个等待的结果(用户界面显示正在抢购，请耐心等待)，而我们系统会有监听器去接收这些秒杀请求进行对应的业务处理，最重要的是，整个系统的扩展显得非常简单，我们只需要部署Kafka的集群，以及后台的负载均衡就能快速提高系统的访问并发量。

Kafka的最大的特点就是高吞吐量以及可水平扩展，正因这两点Kafka非常适合处理数据量庞大的业务，例如使用Kafka做日志分析、数据计算。新版本Kafka也推出了Stream API，可以更好的支持数据流处理。基于这些特性我们可以实现非常多的系统功能。

## 点对点、发布订阅 ##

### 点对点消息系统 ###

以下模型为点对点消息系统的模型，简单来说就是生产者（Producer）发送消息到队列，消费者（Consumer）从队列中取出消息。这种模型的特点就是一条消息只会被一个消费者接收，一但有消费者消费了这条消息，其他消费者就没办法重复消费了。

![Image 1][]

### 发布-订阅消息系统 ###

发布订阅的模型也比较好理解，首先消费者需要订阅这个队列，生产者只要发送一条消息到队列中，所有已订阅该队列的的消费者都能接收到该消息，未订阅的用户则无法接收。就像我们的微信关注微信公众号一样，只有关注了的用户才会收到公众号推送的消息。

![Image 1][]

### Kafka几个主要的概念 ###

![Image 1][]

### Kafka结构 ###

### Broker ###

Broker为节点的意思，我们启动的单个Kafka实例则为一个Broker，多个Broker可以组成Kafka集群

### Topic ###

Topic为主题的意思，也就是相当于消息系统中的队列(queue)，一个Topic中存在多个Partition

### Partition ###

Partition为分区的意思，是构成Kafka存储结构的最小单位

### Partition offset ###

offset为消息偏移量，以Partition为单位，即使在同一个Topic中，不同Partition的offset也是重新开始计算(也就是会重复)

### Group ###

Group为消费者组的意思，一个Group里面包含多个消费者

### Message ###

Message为消息的意思，是队列中消息的承载体，也就是通信的基本单位，Producer可以向Topic中发送Message

这里就需要说说为什么这样设计了：  
首先Topic中有分区的概念，每个分区保存各自的数据，而我们的Group这对应着Topic，也就是这个Topic中的数据都是由该Group去消费，也就是允许多个消费者同时消费，这样能大大提高Kafka的吞吐量。不过这样的设计也会带来不少的不便，比如特定场景下你需要去维护多个Partition之间的关系。

## 二、SpringBoot整合Kafka ##

### 创建工程 ###

![Image 1][]

这里需要注意一下，我们导入的Spring-Kafka为2.1.8版本，SpringBoot为2.0.4的正式版。

![Image 1][]

### 编写第一个Demo ###

实现顺序

1.  创建消费者和生产者的Map配置
2.  根据Map配置创建对应的消费者工厂(consumerFactory)和生产者工厂(producerFactory)
3.  根据consumerFactory创建监听器的监听器工厂
4.  根据producerFactory创建KafkaTemplate(Kafka操作类)
5.  创建监听容器

### 项目结构，记得把Kafka和Zookeeper启动... ###

![Image 1][]

### 创建KafkaConfiguration配置类 ###

都是一些配置参数，具体的作用也在代码中写明了，值得注意的是，KafkaTemplate的类型为<Integer,String>，我们可以找kafkaTemplate的send方法，有多个重载方法，其中有个方法如下，key和data参数都为泛型，这其实就是对应着KafkaTemplate<Integer,String>。那具体有什么用呢，还记得我们的Topic中可以包含多个Partition(分区)吗，那我们如果不想手动指定发送到哪个分区，我们则可以利用key去实现。这里我们的key是Integer类型，template会根据 key 路由到对应的partition中，如果key存在对应的partitionID则发送到该partition中，否则由算法选择发送到哪个partition。

public ListenableFuture<SendResult<K, V>> send(String topic, K key, V data) {
            ProducerRecord<K, V> producerRecord = new ProducerRecord(topic, key, data);
            return this.doSend(producerRecord);
        }

@Configuration
    @EnableKafka
    public class KafkaConfiguration {
    
        //ConcurrentKafkaListenerContainerFactory为创建Kafka监听器的工程类，这里只配置了消费者
        @Bean
        public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory());
            return factory;
        }
    
        //根据consumerProps填写的参数创建消费者工厂
        @Bean
        public ConsumerFactory<Integer, String> consumerFactory() {
            return new DefaultKafkaConsumerFactory<>(consumerProps());
        }
    
        //根据senderProps填写的参数创建生产者工厂
        @Bean
        public ProducerFactory<Integer, String> producerFactory() {
            return new DefaultKafkaProducerFactory<>(senderProps());
        }
    
        //kafkaTemplate实现了Kafka发送接收等功能
        @Bean
        public KafkaTemplate<Integer, String> kafkaTemplate() {
            KafkaTemplate template = new KafkaTemplate<Integer, String>(producerFactory());
            return template;
        }
    
        //消费者配置参数
        private Map<String, Object> consumerProps() {
            Map<String, Object> props = new HashMap<>();
            //连接地址
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            //GroupID
            props.put(ConsumerConfig.GROUP_ID_CONFIG, "bootKafka");
            //是否自动提交
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
            //自动提交的频率
            props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
            //Session超时设置
            props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
            //键的反序列化方式
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
            //值的反序列化方式
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            return props;
        }
    
        //生产者配置
        private Map<String, Object> senderProps (){
            Map<String, Object> props = new HashMap<>();
            //连接地址
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            //重试，0为不启用重试机制
            props.put(ProducerConfig.RETRIES_CONFIG, 1);
            //控制批处理大小，单位为字节
            props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
            //批量发送，延迟为1毫秒，启用该功能能有效减少生产者发送消息次数，从而提高并发量
            props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
            //生产者可以使用的总内存字节来缓冲等待发送到服务器的记录
            props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 1024000);
            //键的序列化方式
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
            //值的序列化方式
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            return props;
        }
    
    }

### 创建DemoListener消费者 ###

这里的消费者其实就是一个监听类，指定监听名为topic.quick.demo的Topic，consumerID为demo。

@Component
    public class DemoListener {
    
        private static final Logger log= LoggerFactory.getLogger(DemoListener.class);
    
        //声明consumerID为demo，监听topicName为topic.quick.demo的Topic
        @KafkaListener(id = "demo", topics = "topic.quick.demo")
        public void listen(String msgData) {
            log.info("demo receive : "+msgData);
        }
    }

### 创建生产者 ###

这里的send方法第一参数为TopicName，第二个参数则是发送的数据

@SpringBootTest
    @RunWith(SpringRunner.class)
    public class DemoTest {
    
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        @Test
        public void testDemo() throws InterruptedException {
            kafkaTemplate.send("topic.quick.demo", "this is my first demo");
            //休眠5秒，为了使监听器有足够的时间监听到topic的数据
            Thread.sleep(5000);
        }
    }

接下来直接运行这个测试方法，我们可以看到日志中输出了我们发送的消息，这就代表我们成功的消费了测试方法中发送的消息。

2018-09-06 17:26:20.850  INFO 6232 --- [     demo-0-C-1] com.viu.kafka.listen.DemoListener        : demo receive : this is my first demo

### 启动项目，我们来观察一下控制台的输出日志 ###

首先这个是KafkaConsumer的配置信息，每个消费者都会输出该配置信息，配置太多就不做讲解了

2018-09-06 17:40:15.258  INFO 9944 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values: 
        auto.commit.interval.ms = 100
        auto.offset.reset = latest
        bootstrap.servers = [localhost:9092]
        check.crcs = true
        client.id = 
        connections.max.idle.ms = 540000
        enable.auto.commit = true
        exclude.internal.topics = true
        fetch.max.bytes = 52428800
        fetch.max.wait.ms = 500
        fetch.min.bytes = 1
        group.id = demo
        heartbeat.interval.ms = 3000
        interceptor.classes = null
        internal.leave.group.on.close = true
        isolation.level = read_uncommitted
        key.deserializer = class org.apache.kafka.common.serialization.IntegerDeserializer
        max.partition.fetch.bytes = 1048576
        max.poll.interval.ms = 300000
        max.poll.records = 500
        metadata.max.age.ms = 300000
        metric.reporters = []
        metrics.num.samples = 2
        metrics.recording.level = INFO
        metrics.sample.window.ms = 30000
        partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
        receive.buffer.bytes = 65536
        reconnect.backoff.max.ms = 1000
        reconnect.backoff.ms = 50
        request.timeout.ms = 305000
        retry.backoff.ms = 100
        sasl.jaas.config = null
        sasl.kerberos.kinit.cmd = /usr/bin/kinit
        sasl.kerberos.min.time.before.relogin = 60000
        sasl.kerberos.service.name = null
        sasl.kerberos.ticket.renew.jitter = 0.05
        sasl.kerberos.ticket.renew.window.factor = 0.8
        sasl.mechanism = GSSAPI
        security.protocol = PLAINTEXT
        send.buffer.bytes = 131072
        session.timeout.ms = 15000
        ssl.cipher.suites = null
        ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
        ssl.endpoint.identification.algorithm = null
        ssl.key.password = null
        ssl.keymanager.algorithm = SunX509
        ssl.keystore.location = null
        ssl.keystore.password = null
        ssl.keystore.type = JKS
        ssl.protocol = TLS
        ssl.provider = null
        ssl.secure.random.implementation = null
        ssl.trustmanager.algorithm = PKIX
        ssl.truststore.location = null
        ssl.truststore.password = null
        ssl.truststore.type = JKS
        value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
    
    2018-09-06 17:40:15.274  INFO 9944 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 1.0.2
    2018-09-06 17:40:15.274  INFO 9944 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : 2a121f7b1d402825

这些日志就代表我们成功的创建了Consumer，由于没有做并发配置，所以现在为单个消费者模式，系统会做一个分配Partition的操作，也就是将某个Partition指定给某个消费者消费。 这里有个地方需要注意一下，  
看到日志中有输出\[Consumer clientId=consumer-1, groupId=demo\]，我们之前在监听中@KafkaListener注解中配置的id=demo，怎么就变成了groupId=demo，这是因为@KafkaListener注解如果没有指定groupId这个属性的值，则会默认把id作为groupId。

2018-09-06 17:40:15.287  INFO 9944 --- [     demo-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=demo] Discovered group coordinator admin-PC:9092 (id: 2147483647 rack: null)
    2018-09-06 17:40:15.290  INFO 9944 --- [     demo-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-1, groupId=demo] Revoking previously assigned partitions []
    2018-09-06 17:40:15.290  INFO 9944 --- [     demo-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions revoked: []
    2018-09-06 17:40:15.290  INFO 9944 --- [     demo-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=demo] (Re-)joining group
    2018-09-06 17:40:15.301  INFO 9944 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8080 (http) with context path ''
    2018-09-06 17:40:15.302  INFO 9944 --- [     demo-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=demo] Successfully joined group with generation 33
    2018-09-06 17:40:15.303  INFO 9944 --- [     demo-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-1, groupId=demo] Setting newly assigned partitions [topic.quick.demo-0]

SpringBoot2.0已经提供了Kafka的自动配置，可以在application.properties/yml文件中配置。

[Image 1]: