kafka 学习（四）spring-kafka生产者消费者收发消息

痛定思痛。 2021-11-10 23:50 382阅读 0赞

# 1、maven 依赖 #

<dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>

2.yml配置

spring:
      kafka:
        bootstrap-servers: 192.168.1.3:9092
        producer:
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
        consumer:
          group-id: test
          enable-auto-commit: true
          auto-commit-interval: 1000
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    
    server:
      port: 8081

# 3.代码 #

## 生产者 ##

package com.example.bootkafka;
    
    import org.junit.After;
    import org.junit.Before;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.test.context.junit4.SpringRunner;
    
    /**
     * kafka生产者测试
     *
     * @author chenye
     */
    @RunWith(SpringRunner.class)
    @SpringBootTest(classes = {BootKafkaProducerApplication.class})
    public class ProducerTest {
    
        @Autowired(required = false)
        private KafkaTemplate defaultKafkaTemplate;
    
        @Test
        public void testDefaultKafkaTemplate() {
            defaultKafkaTemplate.send("test_topic", "I`m send msg to default topic");
        }
    
    
        @Before
        public void testBefore() {
            System.out.println("before");
        }
    
        @After
        public void testAfter() {
            System.out.println("after");
        }
    }

## 消费者 ##

package com.example.bootkafka;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    /**
     * kafka消费者测试
     *
     * @author chenye
     */
    @Component
    public class ConsumerTest {
    
        @KafkaListener(topics = "test_topic")
        public void listen(ConsumerRecord<?, ?> record) {
            System.out.printf("topic = %s, offset = %d, value = %s \n", record.topic(), record.offset(), record.value());
        }
    }