kafka与spring整合

一时失言乱红尘 2022-05-25 23:06 200阅读 0赞

标签（空格分隔）： kafka

--------------------

参考: [https://www.cnblogs.com/hei12138/p/7805475.html][https_www.cnblogs.com_hei12138_p_7805475.html]

## 主要功能 ##

*  1 发布和订阅消息流，这个功能类似于消息队列，这也是kafka归类为消息队列框架的原因
 *  2 以容错的方式记录消息流，kafka以文件的方式来存储消息流
 *  3 可以在消息发布的时候进行处理

## 使用场景 ##

*  **消息队列功能**. 在系统或应用程序之间构建可靠的用于传输实时数据的管道，
 *  **数据处理功能** .构建实时的流数据处理程序来变换或处理数据流，

## 详细介绍 ##

Kafka目前主要作为一个分布式的发布订阅式的消息系统使用，下面简单介绍一下kafka的基本机制

### kafka消息传输流程 ###

![760273-20171108181426763-1692750478.png][]

*  **Producer即生产者**，向Kafka集群发送消息，在发送消息之前，会对消息进行分类，即Topic，上图展示了两个producer发送了分类为topic1的消息，另外一个发送了topic2的消息。
 *  **Topic即主题**，通过对消息指定主题可以将消息分类，消费者可以只关注自己需要的Topic中的消息
 *  **Consumer即消费者**，消费者通过与kafka集群建立长连接的方式，不断地从集群中拉取消息，然后可以对这些消息进行处理。

### kafka服务器消息存储策略 ###

![这里写图片描述][70]

谈到kafka的存储，就不得不提到分区，即partitions，创建一个topic时，同时可以指定分区数目，分区数越多，其吞吐量也越大，但是需要的资源也越多，同时也会导致更高的不可用性，kafka在接收到生产者发送的消息之后，会根据均衡策略将消息存储到不同的分区中。

![760273-20171108181534809-1489003398.png][]

在每个分区中消息以顺序存储,最晚接受到的消息,最后被消费.

### 与生产者的交互 ###

![这里写图片描述][70 1]

生产者在向kafka集群发送消息的时候，可以通过指定分区来发送到指定的分区中

也可以通过指定均衡策略来将消息发送到不同的分区中

如果不指定，就会采用默认的随机均衡策略，将消息随机的存储到不同的分区中

### 与消费者的交互 ###

![这里写图片描述][70 2]

在消费者消费消息时，kafka使用offset来记录当前消费的位置

在kafka的设计中，可以有多个不同的group来同时消费同一个topic下的消息，如图，我们有两个不同的group同时消费，他们的的消费的记录位置offset各不项目，不互相干扰。

对于一个group而言，消费者的数量不应该多余分区的数量，因为在一个group中，每个分区至多只能绑定到一个消费者上，即一个消费者可以消费多个分区，一个分区只能给一个消费者消费

因此，若一个group中的消费者数量大于分区数量的话，多余的消费者将不会收到任何消息

## 使用spring-kafka ##

### 1 添加依赖 ###

<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.11.0.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <version>0.11.0.1</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
        <version>1.3.0.RELEASE</version>
    </dependency>

### 2 使用java配置 ###

*  我们在主目录下新建名为KafkaConfig的类

@Configuration
    @EnableKafka
    public class KafkaConfig { 
    
    }

*  配置Topic,在kafkaConfig类中添加配置

//topic config Topic的配置开始
        @Bean
        public KafkaAdmin admin() {
            Map<String, Object> configs = new HashMap<String, Object>();
            configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.180.128:9092");
            return new KafkaAdmin(configs);
        }
    
        @Bean
        public NewTopic topic1() {
            return new NewTopic("foo", 10, (short) 2);
        }
    //topic的配置结束

*  配置生产者Factort及Template

//producer config start
        @Bean
        public ProducerFactory<Integer, String> producerFactory() {
            return new DefaultKafkaProducerFactory<Integer,String>(producerConfigs());
        }
        @Bean
        public Map<String, Object> producerConfigs() {
            Map<String, Object> props = new HashMap<String,Object>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.180.128:9092");
            props.put("acks", "all");
            props.put("retries", 0);
            props.put("batch.size", 16384);
            props.put("linger.ms", 1);
            props.put("buffer.memory", 33554432);
            props.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            return props;
        }
        @Bean
        public KafkaTemplate<Integer, String> kafkaTemplate() {
            return new KafkaTemplate<Integer, String>(producerFactory());
        }
    //producer config end

*  配置ConsumerFactory

//consumer config start
        @Bean
        public ConcurrentKafkaListenerContainerFactory<Integer,String> kafkaListenerContainerFactory(){
            ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<Integer, String>();
            factory.setConsumerFactory(consumerFactory());
            return factory;
        }
    
        @Bean
        public ConsumerFactory<Integer,String> consumerFactory(){
            return new DefaultKafkaConsumerFactory<Integer, String>(consumerConfigs());
        }
    
    
        @Bean
        public Map<String,Object> consumerConfigs(){
            HashMap<String, Object> props = new HashMap<String, Object>();
            props.put("bootstrap.servers", "192.168.180.128:9092");
            props.put("group.id", "test");
            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");
            props.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            return props;
        }
    //consumer config end

## 创建消息生产者 ##

//使用spring-kafka的template发送一条消息 发送多条消息只需要循环多次即可
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(KafkaConfig.class);
        KafkaTemplate<Integer, String> kafkaTemplate = (KafkaTemplate<Integer, String>) ctx.getBean("kafkaTemplate");
            String data="this is a test message";
            ListenableFuture<SendResult<Integer, String>> send = kafkaTemplate.send("topic-test", 1, data);
            send.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
                public void onFailure(Throwable throwable) {
    
                }
    
                public void onSuccess(SendResult<Integer, String> integerStringSendResult) {
    
                }
            });
    }

## 创建消息消费者 ##

我们首先创建一个一个用于消息监听的类，当名为”topic-test”的topic接收到消息之后，我们的这个listen方法就会调用。

public class SimpleConsumerListener { 
        private final static Logger logger = LoggerFactory.getLogger(SimpleConsumerListener.class);
        private final CountDownLatch latch1 = new CountDownLatch(1);
    
        @KafkaListener(id = "foo", topics = "topic-test")
        public void listen(byte[] records) {
            //do something here
            this.latch1.countDown();
        }
    }
    
             我们同时也需要将这个类作为一个Bean配置到KafkaConfig中
    
    @Bean
    public SimpleConsumerListener simpleConsumerListener(){
        return new SimpleConsumerListener();
    }

默认spring-kafka会为每一个监听方法创建一个线程来向kafka服务器拉取消息.

[https_www.cnblogs.com_hei12138_p_7805475.html]: https://www.cnblogs.com/hei12138/p/7805475.html
[760273-20171108181426763-1692750478.png]: /images/20220526/1b0814f38912419c986daa6273db8994.png
[70]: /images/20220526/c8995e7b28bb4ab4a929a3b4f59014cb.png
[760273-20171108181534809-1489003398.png]: /images/20220526/db87c94c5fd941b3a8c6a412fd21c07b.png
[70 1]: /images/20220526/49953d7a1835459c915c0bc1a3cc6e27.png
[70 2]: /images/20220526/3aaa3a9a086346b3ae1d34a109e1b284.png