kafka基本的消息生产与消费

旧城等待， 2022-05-14 07:21 215阅读 0赞

分布式的一个消息系统，生产者-broker-消费者

### 分布式 ###

每个分区都由一个服务器作为“leader”，零或若干服务器作为“followers”,leader负责处理消息的读和写，followers则去复制leader.如果leader down了，followers中的一台则会自动成为leader。集群中的每个服务都会同时扮演两个角色：作为它所持有的一部分分区的leader，同时作为其他分区的followers，这样集群就会据有较好的负载均衡。

### producer ###

spring-kafka提供了KafkaTemplate包装了一个生产者，并提供了方便的方法将数据发送到kafka的主题，Producer也能决定将此消息归属于哪个partition

### Consumers ###

本质上kafka只支持Topic.每个consumer属于一个consumer group;反过来说,每个group中可以有多个consumer.发送到Topic的消息,只会被订阅此Topic的每个group中的一个consumer消费.

如果所有的consumer都具有相同的group,这种情况和queue模式很像;消息将会在consumers之间负载均衡.  
如果所有的consumer都具有不同的group,那这就是”发布-订阅”;消息将会广播给所有的消费者.  
在kafka中,一个partition中的消息只会被group中的一个consumer消费;每个group中consumer消息消费互相独立;我们可以认为一个group是一个”订阅”者,一个Topic中的每个partions,只会被一个”订阅者”中的一个consumer消费,不过一个consumer可以消费多个partitions中的消息.kafka只能保证一个partition中的消息被某个consumer消费时,消息是顺序的.事实上,从Topic角度来说,消息仍不是有序的.

一个Topic可以认为是一类消息，每个topic将被分成多个partition(区),每个partition在存储层面是append log文件。任何发布到此partition的消息都会被直接追加到log文件的尾部，每条消息在文件中的位置称为offset（偏移量），offset为一个long型数字，它是唯一标记一条消息。它唯一的标记一条消息。kafka并没有提供其他额外的索引机制来存储offset，因为在kafka中几乎不允许对消息进行“随机读写”。

kafka和JMS（Java Message Service）实现(activeMQ)不同的是:即使消息被消费,消息仍然不会被立即删除.日志文件将会根据broker中的配置要求,保留一定的时间之后删除;比如log文件保留2天,那么两天后,文件会被清除,无论其中的消息是否被消费.kafka通过这种简单的手段,来释放磁盘空间,以及减少消息消费之后对文件内容改动的磁盘IO开支.

对于consumer而言,它需要保存消费消息的offset,对于offset的保存和使用,有consumer来控制;当consumer正常消费消息时,offset将会”线性”的向前驱动,即消息将依次顺序被消费.事实上consumer可以使用任意顺序消费消息,它只需要将offset重置为任意值..(offset将会保存在zookeeper中,参见下文)

kafka集群几乎不需要维护任何consumer和producer状态信息,这些信息有zookeeper保存;因此producer和consumer的客户端实现非常轻量级,它们可以随意离开,而不会对集群造成额外的影响.

partitions的设计目的有多个.最根本原因是kafka基于文件存储.通过分区,可以将日志内容分散到多个server上,来避免文件尺寸达到单机磁盘的上限,每个partiton都会被当前server(kafka实例)保存;可以将一个topic切分多任意多个partitions,来消息保存/消费的效率.此外越多的partitions意味着可以容纳更多的consumer,有效提升并发消费的能力.(具体原理参见下文).

### yml配置 ###

kafka:
        bootstrap-servers: 127.0.0.1:9092
        producer:
          acks: 1
          batch-size: 100
          client-id: canal
          retries: 3
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
          bootstrap-servers: 127.0.0.1:9092
        consumer:
          enable-auto-commit: true
          group-id: kafkatest
          auto-offset-reset: latest
          bootstrap-servers: 127.0.0.1:9092
        template:
          default-topic: canal_test

### Producer ###

package com.kexin.kafka.conf;
    
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.kafka.annotation.EnableKafka;
    import org.springframework.kafka.core.DefaultKafkaProducerFactory;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.kafka.core.ProducerFactory;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * @Author KeXin
     * @Date 2018/7/19 下午1:30
     **/
    @Configuration
    @EnableKafka
    public class MyProducerConfig {
        
    
    
        public Map<String,Object> producerConfigs(){
            Map<String,Object> props = new HashMap<>();
            System.out.println("---------------------producerConfigs------------------");
            props.put("serializer.class", "kafka.serializer.StringEncoder");
    //        props.put("metadata.broker.list", "127.0.0.1:9092");
            props.put("bootstrap.servers","127.0.0.1:9092");
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            return props;
        }
        /**
         * 生产者创建工厂
         * @return
         */
        public ProducerFactory<String, String> producerFactory() {
            return new DefaultKafkaProducerFactory<>(producerConfigs());
        }
    
        /**
         * kafkaTemplate 覆盖默认配置类中的kafkaTemplate
         * @return
         */
        @Bean
        public KafkaTemplate<String, String> kafkaTemplate() {
            return new KafkaTemplate<>(producerFactory());
        }
    
    }

生产消息：

package com.kexin.kafka;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    /**
     * @Author KeXin
     * @Date 2018/7/19 下午1:39
     **/
    @RestController
    public class SendMessageController {
        
        @Autowired
        KafkaTemplate kafkaTemplate;
    
        @GetMapping(value = "/sendMessage")
        public String sendMessage(String message){
    
            kafkaTemplate.send("test",message);
            return "success";
        }
    
    
    }

### Consumer ###

package com.kexin.canal.test;
    
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    
    /**
     * @Author KeXin
     * @Date 2018/7/25 下午3:27
     **/
    @Component
    public class Consumer {
        
        @KafkaListener(topics = {
       "kafkatest"})
        public void receive(String message){
            System.out.println("consume->[" + message+"]");
        }
    }