Spring Kafka 教程 – spring读取和发送kakfa消息

秒速五厘米 2022-05-25 10:25 461阅读 0赞

Apache Kafka, 分布式消息系统， 非常流行。Spring是非常流行的Java快速开发框架。将两者无缝平滑结合起来可以快速实现很多功能。本文件简要介绍Spring Kafka，如何使用 KafkaTemplate发送消息到kafka的broker上， 如何使用“listener container“接收Kafka消息。

**1，Spring Kafka的组成**  
这一节我们首先介绍Spring Kafka的各个组成部分。  
1.1 发送消息  
与 JmsTemplate 或者JdbcTemplate类似，Spring Kafka提供了 KafkaTemplate. 该模板封装了Kafka消息生产者并提供各种消息发送方法。  
消息发送的各种方法。

ListenableFuture<SendResult<K, V>> send(String topic, V data);
    
    ListenableFuture<SendResult<K, V>> send(String topic, K key, V data);
    
    ListenableFuture<SendResult<K, V>> send(String topic, int partition, V data);
    
    ListenableFuture<SendResult<K, V>> send(String topic, int partition, K key, V data);
    
    ListenableFuture<SendResult<K, V>> send(Message<?> message);

1.2 接收消息  
要接收消息，我们需要配置MessageListenerContainer并提供一个Message Listener，或者使用 @KafkaListener注解。

MessageListenserContainer  
MessageListenserContainer 有以下两个实现类：

KafkaMessageListenerContainer
    ConcurrentMessageListenerContainer

KafkaMessageListenerContainer可以让我们使用单线程消费Kafka topic的消息，而ConcurrentMessageListenerContainer 可以让我们多线程消费消息。

@KafkaListener 注解  
Spring Kafka提供的 @KafkaListener注解，可以让我们监听某个topic或者topicPattern的消息。

监听符合topicPattern = “topic.\*”的所有topic的消息

@Component
    @Slf4j
    public class CmdReceiver {
        @KafkaListener(topicPattern = "topic.*")
        public void listen(ConsumerRecord<?, ?> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                Object message = kafkaMessage.get();
                log.info("----------------- record =topic："  + topic+ ", " + record);
                log.info("------------------ message =topic：" + topic+ ", " + message);
            }
        }
    }

监听某个topic的消息    
    public class Listener {
    
        @KafkaListener(id = "id01", topics = "Topic1")
        public void listen(String data) {
    
        }
    }

**2， Spring Kafka 例子**

下面我们介绍一个具体的例子， 这个例会发送和接收指定topic的消息。  
**准备工作**：  
kafka\_2.11-1.1.0.tgz和zookeeper-3.4.10.tar.gz  
JDK jdk-8u171-linux-x64.tar.gz  
IDE (Eclipse or IntelliJ)  
Build tool (Maven or Gradle)  
本文不涉及安装Kafka的介绍，请自行搜索，或者看官方文档。

**pom文件**：  
也就是我们的依赖包. 这是笔者使用的依赖版本，仅供参考。

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.yq</groupId>
        <artifactId>kafkademo</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>1.5.12.RELEASE</version>
            <relativePath/> 
        </parent>
    
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
            <java.version>1.8</java.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <optional>true</optional>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka</artifactId>
                <version>1.1.8.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>com.google.code.gson</groupId>
                <artifactId>gson</artifactId>
                <version>2.8.2</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
                <version>0.10.1.1</version>
            </dependency>
    
            <dependency>
                <groupId>io.springfox</groupId>
                <artifactId>springfox-swagger2</artifactId>
                <version>2.7.0</version>
            </dependency>
            <dependency>
                <groupId>io.springfox</groupId>
                <artifactId>springfox-swagger-ui</artifactId>
                <version>2.7.0</version>
            </dependency>
            <dependency>
                <groupId>io.springfox</groupId>
                <artifactId>springfox-spring-web</artifactId>
                <version>2.7.0</version>
            </dependency>
    
            
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.1.33</version>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    
    </project>

***\*KafkaDemoApplication\****  
我们使用springboot的框架，这是我们程序的入口点。

@SpringBootApplication
    public class KafkaDemoApplication { 
        private static final Logger logger = LoggerFactory.getLogger(KafkaDemoApplication.class);
        public static void main(String[] args) {
            ConfigurableApplicationContext context = SpringApplication.run(KafkaDemoApplication.class, args);
            logger.info("Done start Spring boot");
        }
    }

**ProducerConfig**  
其实我们可以可以不用编写KafkaProducerConfig，直接使用KafkaTemplate（当然前提是我们要设置好producer需要的配置项，例如spring.kafka.bootstrap-servers, spring.kafka.producer.key-serializer, spring.kafka.producer.retries等等）

@Configuration
    @EnableKafka
    public class KafkaProducerConfig { 
        @Bean
        public ProducerFactory<String, String> producerFactory() {
            return new DefaultKafkaProducerFactory<>(producerConfigs());
        }
    
        @Bean
        public Map<String, Object> producerConfigs() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092(根据实际情况修改)");
            props.put(ProducerConfig.RETRIES_CONFIG, 0);
            props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
            props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
            props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            return props;
        }
    
        @Bean
        public KafkaTemplate<String, String> kafkaTemplate() {
            return new KafkaTemplate<String, String>(producerFactory());
        }
    }

**KafkaConsumerConfig**  
同理，其实我们可以可以不用编写KafkaConsumerConfig，直接使用 @KafkaListener（当然前提是我们要设置好consumer需要的配置项，例如spring.kafka.bootstrap-servers, spring.kafka.consumer.key-deserializer, spring.kafka.consumer.group-id、spring.kafka.consumer.auto-offset-reset等等）

@Configuration
    @EnableKafka
    public class KafkaConsumerConfig { 
        @Bean
        KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory());
            factory.setConcurrency(3);
            factory.getContainerProperties().setPollTimeout(3000);
            return factory;
        }
    
        @Bean
        public ConsumerFactory<String, String> consumerFactory() {
            return new DefaultKafkaConsumerFactory<>(consumerConfigs());
        }
    
        @Bean
        public Map<String, Object> consumerConfigs() {
            Map<String, Object> propsMap = new HashMap<>();
            propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092(根据实际情况修改)");
            propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
            propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
            propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
            propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
            propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
            return propsMap;
        }
    
        @Bean
        public MyListener listener() {
            return new MyListener();
        }
    }

定义了ProducerConfig和ConsumerConfig后我们需要实现具体的生产者和消费者。

本文的KafkaListenerContainerFactory 中使用了ConcurrentKafkaListenerContainer， 我们将使用多线程消费消息。

注意消息代理的地址是localhost:9092， 需要根据实际情况修改。需要特别注意的是，我在windows运行程序，kafka在我的linux虚拟机， 我需要配置windows的hosts文件，配置虚拟机hostname和ip的映射，例如192.168.119.131 ubuntu01

**开发Listener**

我们来开发自己的Listener监听具体的topic， 这里例子中我们监听以topic开头的主题，不做其他业务，只是打印出来。

@Component
    @Slf4j
    public class MyListener{
        @KafkaListener(topicPattern = "topic.*")
        public void listen(ConsumerRecord<?, ?> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                Object message = kafkaMessage.get();
                log.info("------------------ message =topic：" + topic+ ", " + message);
            }
        }
    }

**开发producer**  
我在程序中增加了controller，这样我们可以通过controller给topic发消息。consumer一直在监听，只要有消息发送过去，就会打印出来。controller中调用了ProducerServiceImpl ， 具体代码比较简单就不再罗列。

我们producerServiceImpl主要是有这句， 通过KafkaTemplate 发送消息。  
@Autowired  
private KafkaTemplate template;

@Service
    public class ProducerServiceImpl implements ProducerService { 
        private static final Logger logger = LoggerFactory.getLogger(ProducerServiceImpl.class);
    
        private Gson gson = new GsonBuilder().create();
    
        @Autowired
        private KafkaTemplate template;
    
        //发送消息方法
        public void sendJson(String topic, String json) {
            JSONObject jsonObj = JSON.parseObject(json);
    
            jsonObj.put("topic", topic);
            jsonObj.put("ts", System.currentTimeMillis() + "");
    
            logger.info("json+++++++++++++++++++++ message = {}", jsonObj.toJSONString());
    
            ListenableFuture<SendResult<String, String>> future = template.send(topic, jsonObj.toJSONString());
            future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onSuccess(SendResult<String, String> result) {
                    System.out.println("msg OK." + result.toString());
                }
    
                @Override
                public void onFailure(Throwable ex) {
                    System.out.println("msg send failed: ");
                }
            });
        }

**运行程序**

运行第一步，确保Kafka broker配置正确，笔者的程序在Windows10机器上，Kafka在虚拟机上，因为我的地址是192.168.119.129:9092, 而不是localhost:9092.

运行第二步骤，在IDEA中选中KafkaDemoApplication , 单击鼠标右键，选择 Run KafkaDemoApplication

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

kafka段命令行接收到的消息

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

**3，总结**  
Spring Kafka提供了很好的集成，我们只需配置properties文件，就可以直接使用KafkaTemplate发送消息，使用@KafkaListener监听消息。

参考文档：  
[https://docs.spring.io/spring-kafka/docs/2.1.6.RELEASE/reference/html/\_reference.html\#kafka-template][https_docs.spring.io_spring-kafka_docs_2.1.6.RELEASE_reference_html_reference.html_kafka-template]

[70]: /images/20220525/3ab9440f12a64f169604efa62199419c.png
[70 1]: /images/20220525/e3ba4885f6ed4168a5c2306d382aaa96.png
[https_docs.spring.io_spring-kafka_docs_2.1.6.RELEASE_reference_html_reference.html_kafka-template]: https://docs.spring.io/spring-kafka/docs/2.1.6.RELEASE/reference/html/_reference.html#kafka-template