Springboot 2.x集成kafka 2.2.0的示例代码

<? xml version = "1.0" encoding = "UTF-8" ?>

< project xmlns = "http://maven.apache.org/POM/4.0.0" xmlns:xsi = "http://HdhCmsTestw3.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 >

     < parent >

         < groupId >org.springframework.boot</ groupId >

         < artifactId >spring-boot-starter-parent</ artifactId >

         < version >2.1.4.RELEASE</ version >

         < relativePath /> <!-- lookup parent from repository -->

     </ parent >

     < groupId >com.example</ groupId >

     < artifactId >demo</ artifactId >

     < version >0.0.1-SNAPSHOT</ version >

     < name >kafkademo</ name >

     < description >Demo project for Spring Boot</ description >

     < properties >

         < java.version >1.8</ java.version >

     </ properties >

     < dependencies >

         < dependency >

             < groupId >org.springframework.boot</ groupId >

             < artifactId >spring-boot-starter-web</ artifactId >

         </ dependency >

         < dependency >

             < groupId >mysql</ groupId >

             < artifactId >mysql-connector-java</ artifactId >

             < scope >runtime</ scope >

         </ 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 >2.2.0.RELEASE</ version >

         </ dependency >

         < dependency >

             < groupId >com.google.code.gson</ groupId >

             < artifactId >gson</ artifactId >

             < version >2.7</ version >

         </ dependency >

     </ dependencies >

     < build >

         < plugins >

             < plugin >

                 < groupId >org.springframework.boot</ groupId >

                 < artifactId >spring-boot-maven-plugin</ artifactId >

             </ plugin >

         </ plugins >

     </ build >

</ project >

spring.kafka.bootstrap-servers= 2.1 . 1.1 : 9092

spring.kafka.consumer.group-id=test-consumer-group

spring.kafka.consumer.key-deserializer=org.apache.kafka测试数据mon.serialization.StringDeserializer

spring.kafka.consumer.value-deserializer=org.apache.kafka测试数据mon.serialization.StringDeserializer

spring.kafka.producer.key-serializer=org.apache.kafka测试数据mon.serialization.StringSerializer

spring.kafka.producer.value-serializer=org.apache.kafka测试数据mon.serialization.StringSerializer

#logging.level.root=debug

package com.example.demo.model;

import java.util.Date;

public class Messages {

     private Long id;

     private String msg;

     private Date sendTime;

     public Long getId() {

         return id;

     }

     public void setId(Long id) {

         this .id = id;

     }

     public String getMsg() {

         return msg;

     }

     public void setMsg(String msg) {

         this .msg = msg;

     }

     public Date getSendTime() {

         return sendTime;

     }

     public void setSendTime(Date sendTime) {

         this .sendTime = sendTime;

     }

}

package com.example.demo.service;

import com.example.demo.model.Messages;

import com.google.gson.Gson;

import com.google.gson.GsonBuilder;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.kafka.core.KafkaTemplate;

import org.springframework.kafka.support.SendResult;

import org.springframework.stereotype.Service;

import org.springframework.util.concurrent.ListenableFuture;

import java.util.Date;

import java.util.UUID;

@Service

public class KafkaSender {

     @Autowired

     private KafkaTemplate<String, String> kafkaTemplate;

     private Gson gson = new GsonBuilder().create();

     public void send() {

         Messages message = new Messages();

         message.setId(System.currentTimeMillis());

         message.setMsg( "123" );

         message.setSendTime( new Date());

         ListenableFuture<SendResult<String, String>> test0 = kafkaTemplate.send( "newtopic" , gson.toJson(message));

     }

}

package com.example.demo.service;

import org.apache.kafka.clients.consumer.ConsumerRecord;

import org.springframework.kafka.annotation.KafkaListener;

import org.springframework.stereotype.Service;

import java.util.Optional;

@Service

public class KafkaReceiver {

     @KafkaListener (topics = { "newtopic" })

     public void listen(ConsumerRecord<?, ?> record) {

         Optional<?> kafkaMessage = Optional.ofNullable(record.value());

         if (kafkaMessage.isPresent()) {

             Object message = kafkaMessage.get();

             System.out.println( "record =" + record);

             System.out.println( "message =" + message);

         }

     }

}

package com.example.demo;

import com.example.demo.service.KafkaSender;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.context.ConfigurableApplicationContext;

@SpringBootApplication

public class KafkademoApplication {

     public static void main(String[] args) {

         ConfigurableApplicationContext context = SpringApplication.run(KafkademoApplication. class , args);

         KafkaSender sender = context.getBean(KafkaSender. class );

         for ( int i = 0 ; i < 1000 ; i++) {

             sender.send();

             try {

                 Thread.sleep( 300 );

             } catch (InterruptedException e) {

                 e.printStackTrace();

             }

         }

     }

}