Kafka 生产者发送消息方式 java

心已赠人 2021-12-15 23:29 826阅读 0赞

Kafka是一种分布式的基于发布/订阅的消息系统，它的高吞吐量、灵活的offset是其它消息系统所没有的。

Kafka发送消息主要有三种方式：

1.发送并忘记 2.同步发送 3.异步发送+回调函数

\*\*

## 1.发送并忘记 ##

\*\*  
只发送，不会获取消息发送的返回结果。（吞吐量是最高的，但是无法保证消息的可靠性）  
举例：

@Autowired
        private KafkaTemplate kafkaTemplate;
        @Test
        public void kafkaFirst(){
            for (int i=0;i<101;i++){
                Long startTime = System.currentTimeMillis();
                kafkaTemplate.send("yearns",0,null,"yearns"+i);
                Long endTime = System.currentTimeMillis();
                System.out.println("耗时："+(endTime-startTime)+"毫秒");
            }
        }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70]  
\*\*

## 2.同步发送 ##

\*\*  
一条一条的发送，对每条消息返回的结果判断， 可以明确地知道每条消息的发送情况，但是由于同步的方式会阻塞，只有当消息通过get返回future对象时，才会继续下一条消息的发送。  
举例：

@Autowired
        private KafkaTemplate kafkaTemplate;
        @Test
        public void kafkaSecond(){
            for (int i=0;i<100;i++){
                Long startTime = System.currentTimeMillis();
                ListenableFuture<SendResult> sendResultListenableFuture =kafkaTemplate.send("yearns",0,null,"yearns"+i);
                try {
                   SendResult sendResult = sendResultListenableFuture.get();
                   String topic= sendResult.getRecordMetadata().topic();
                   int partition= sendResult.getRecordMetadata().partition();
                   Long endTime = System.currentTimeMillis();
                   System.out.println("topic:"+topic+",partition:"+partition+",耗时："+(endTime-startTime)+"毫秒");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
    
            }
        }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70 1]  
因为消费的监听在同一个项目，所以测试类测试时，因为同步发送的原因，大量数据发送时，就会有一部分数据提前在测试类被消费。

\*\*

## 3.异步发送+回调函数 ##

\*\*  
在调用send方法发送消息的同时，指定一个回调函数，服务器在返回响应时会调用该回调函数，通过回调函数能够对异常情况进行处理，当调用了回调函数时，只有回调函数执行完毕生产者才会结束，否则一直会阻塞  
举例：

@Autowired
        private KafkaTemplate kafkaTemplate;
        @Autowired
        private KafkaSendResultHandler kafkaSendResultHandler;
        @Test
        public void kafkaThird(){
            for (int i=0;i<100;i++){
                Long startTime = System.currentTimeMillis();
                kafkaTemplate.setProducerListener(kafkaSendResultHandler);
                kafkaTemplate.send("yearns",0,null,"yearns"+i);
                Long endTime = System.currentTimeMillis();
                System.out.println("耗时："+(endTime-startTime)+"毫秒");
    
            }
        }

package com.yearns.kafka.util;
    
    
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.clients.producer.RecordMetadata;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.kafka.support.ProducerListener;
    import org.springframework.stereotype.Component;
    
    /**
     * @author yearns
     * @date 2019/7/3 14:27
     */
    @Component
    public class KafkaSendResultHandler implements ProducerListener {
        private static final Logger log = LoggerFactory.getLogger(KafkaSendResultHandler.class);
    
        @Override
        public void onSuccess(ProducerRecord producerRecord, RecordMetadata recordMetadata) {
            log.info("Message send success : " + producerRecord.toString());
        }
    
        @Override
        public void onError(ProducerRecord producerRecord, Exception exception) {
            log.info("Message send error : " + producerRecord.toString());
        }
    
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70 3]  
异步回调，可以实现发送失败重新，或者失败后进行其它操作。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70]: /images/20211215/74a0f84892144e9abf4a1b81235efd0b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70 1]: /images/20211215/74065a6d6ee94e8e8eb9aebe955b887e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70 2]: /images/20211215/f8003a5fb5cc45409ccee37bf5765e3e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70 3]: /images/20211215/b1a1a1d56a3847efbee484f4d3fef31e.png