kafka整合springboot2.0（六）

╰+攻爆jí腚メ 2021-10-14 01:44 253阅读 0赞

## 1、生产者 消费者pom ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70]

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.4.RELEASE</version>
        <relativePath/> 
    </parent>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>
    </dependencies>

## 2、生产者 ##

application.yml:

spring:
      kafka:
        #生产者
        producer:
          #发送失败重试次数
          retries: 0
          #生产者批次投递大小  16384(16k)默认
          batch-size: 16384
          #生产缓冲区大小 默认值 (32M)
          buffer-memory: 33554432
          #配置序列化
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
        #kafka代理地址
        bootstrap-servers: 192.168.92.39:9092

KafkaProviderService

@Service
    public class KafkaProviderService {
        @Autowired
        private KafkaTemplate kafkaTemplate;
        public void sendMsg() {
            for (int i = 1; i <= 50; i++) {
                kafkaTemplate.send("kafka-spring","spring-key", "hello spring boot");
            }
        }
    }

测试类：

@RunWith(SpringRunner.class)
    //主application方法
    @SpringBootTest(classes=KafkaProviderApplication.class)
    public class KafkaTest {
        @Autowired
        private KafkaProviderService kafkaProviderService;
        @Test
        public void mainTest(){
            kafkaProviderService.sendMsg();
        }
    }

## 3、消费者 ##

application.yml

spring:
      kafka:
        bootstrap-servers: 192.168.92.39:9092
        consumer:
          #组iD
          group-id: kafka-spring
          #反序列化
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          #配置手动提交
          enable-auto-commit: false
        listener:
          ack-mode: manual

@Service
    @Slf4j
    public class KafkaCustomer {
        @KafkaListener(topics = {"kafka-spring"})
        public void listen(ConsumerRecord<?, ?> record, Acknowledgment ack) {
            try {
                log.info("自行确认方式收到消息的key: " + record.key());
                log.info("自行确认方式收到消息的value: " + record.value().toString());
              ack.acknowledge();
            } finally {
                log.info("消息确认！");
               
            }
        }
    }

## 4、启动消费者生产者测试类 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70]: /images/20211013/307688eb849a4bbfb0cfc330eb5cd728.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 1]: /images/20211013/25e49cef337f48398530f7db507b9fc3.png