Kafka：Springboot整合Kafka消息隊列

本文主要分享下Spring Boot和Spring Kafka如何配置整合，實現發送和接收來自Spring 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.java.zy</groupId>
    <artifactId>base-kafka</artifactId>
    <version>1.0-SNAPSHOT</version>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.2.RELEASE</version>
        <relativePath/>
    </parent>

    <dependencies>

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.47</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <!--引入kafak和spring整合的jar-->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.2.8.RELEASE</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka-test -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <version>2.2.8.RELEASE</version>
            <scope>test</scope>
        </dependency>

    </dependencies>

</project>

Springboot+kafka對應表

 JAVA代碼

定義數據傳輸對象

 

package com.java.common.entity;

import org.springframework.stereotype.Component;

@Component
public class UserLog {

    private String username;
    private String userid;
    private String state;

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getUserid() {
        return userid;
    }

    public void setUserid(String userid) {
        this.userid = userid;
    }

    public String getState() {
        return state;
    }

    public void setState(String state) {
        this.state = state;
    }
}

 

定義發送信息實體 ( 消息的發送直接使用KafkaTemplate模板即可,都封裝好了,直接使用 )

 

 

package com.java.common.controller;

import com.alibaba.fastjson.JSON;
import com.java.common.entity.UserLog;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class UserLogProducer {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 發送數據
     * @param userid
     */
    public void sendLog(String userid){
        UserLog userLog = new UserLog();
        userLog.setUsername("jhp");
        userLog.setUserid(userid);
        userLog.setState("0");
        System.err.println("發送用戶日志數據:"+userLog);
        kafkaTemplate.send("user-log", JSON.toJSONString(userLog));
    }

}

 

獲取消息的消費者實體

注意：消費機制是通過監聽器實現的,直接使用@KafkaListener(topics = {"user-log"})注解，根據指定的條件進行消息的監聽

 

 

package com.java.common.controller;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Component
public class UserLogConsumer {

    @KafkaListener(topics = {"user-log"})
    public void consumer(ConsumerRecord<?,?> consumerRecord){
        //判斷是否為null
        Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
        System.out.println(">>>>>>>>>> record = " + kafkaMessage);
        if(kafkaMessage.isPresent()){
            //得到Optional實例中的值
            Object message = kafkaMessage.get();
            System.err.println("消費消息:"+message);
        }
    }

}

 

Springboot應用啟動類（方便測試,在容器初始化的就開始模擬消息的發送）

 

 

package com.java.common;

import com.java.common.controller.UserLogProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import javax.annotation.PostConstruct;

@SpringBootApplication
public class KafkaApplication {

    @Autowired
    private UserLogProducer kafkaSender;
    @PostConstruct
    public void init(){
        for (int i = 0; i < 10; i++) {
            //調用消息發送類中的消息發送方法
            kafkaSender.sendLog(String.valueOf(i));
        }
    }

    public static void main(String[] args) {
        SpringApplication.run(KafkaApplication.class,args);
    }

}

 

對應的application.properties配置文件

 

 

spring.application.name=base-kafka
server.port=8080
#============== kafka ===================
# 指定kafka 代理地址，可以多個
spring.kafka.bootstrap-servers=localhost:9092,192.168.100.1:9093
 
#=============== provider  =======================
spring.kafka.producer.retries=0
# 每次批量發送消息的數量
spring.kafka.producer.batch-size=16384
spring.kafka.producer.buffer-memory=33554432
# 指定消息key和消息體的編解碼方式
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
 
#=============== consumer  =======================
# 指定默認消費者group id
spring.kafka.consumer.group-id=user-log-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.auto-commit-interval=100
# 指定消息key和消息體的編解碼方式
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

 

對應的application.yml配置文件

 

 

server:
  port: 8080

spring:
  application:
    name: base-kafka
  #============== kafka ===================
  # 指定kafka 代理地址，可以多個
  kafka:
    bootstrap-servers: 192.168.200.27:19092

    #=============== provider  =======================
    producer:
      retries: 0
      # 每次批量發送消息的數量
      batch-size: 16384
      buffer-memory: 33554432
      # 指定消息key和消息體的編解碼方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

    #=============== consumer  =======================
    consumer:
      # 指定默認消費者group id(消息組)
      group-id: user-log
      auto-offset-reset: earliest
      enable-auto-commit: true
      auto-commit-interval: 100
      # 指定消息key和消息體的編解碼方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

 

測試結果

 

啟動項目，觀察控制台打印的消息

 

 

和

 

 

 

springboot和kafka成功整合！

 

文章轉載至：https://blog.csdn.net/qq_18603599/article/details/81169488