快捷導(dǎo)航

Spring boot集成Kafka消息中間件代碼實(shí)例

更新時(shí)間：2020年05月16日 08:43:45 作者：墨營(yíng)

這篇文章主要介紹了Spring boot集成Kafka消息中間件代碼實(shí)例,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

一．創(chuàng)建Spring boot項(xiàng)目，添加如下依賴

<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>
    <!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
    <dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
    </dependency>
    <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-clients</artifactId>
    </dependency>
    <dependency>
      <groupId>com.alibaba</groupId>
      <artifactId>fastjson</artifactId>
      <version>1.2.41</version>
    </dependency>

二．配置文件

server.port=4400

#kafka配置
#============== kafka ===================
# 指定kafka 代理地址，可以多個(gè)
spring.kafka.bootstrap-servers=192.168.102.88:9092
# 指定默認(rèn)消費(fèi)者group id
spring.kafka.consumer.group-id=jkafka.demo
#earliest 當(dāng)各分區(qū)下有已提交的offset時(shí)，從提交的offset開始消費(fèi)；無提交的offset時(shí)，從頭開始消費(fèi)
#latest 當(dāng)各分區(qū)下有已提交的offset時(shí)，從提交的offset開始消費(fèi)；無提交的offset時(shí)，消費(fèi)新產(chǎn)生的該分區(qū)下的數(shù)據(jù)
#none topic各分區(qū)都存在已提交的offset時(shí)，從offset后開始消費(fèi)；只要有一個(gè)分區(qū)不存在已提交的offset，則拋出異常
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.auto-commit-interval=100
# 指定消費(fèi)者消息key和消息體的編解碼方式
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

三．編輯消息實(shí)體

@Data
public class Message implements Serializable{

  /**
   * 
   */
  private static final long serialVersionUID = 2522280475099635810L;

  //消息ID
  private String id;

  //消息內(nèi)容
  private String msg;

  // 消息發(fā)送時(shí)間
  private Date sendTime;

}

四．消息發(fā)送類

@Component
public class KfkaProducer {

  private static Logger logger = LoggerFactory.getLogger(KfkaProducer.class);

  @Autowired
  private KafkaTemplate<String, String> kafkaTemplate;

  public void send(String topic,Message message) {
    try {
      logger.info("正在發(fā)送消息...");
      kafkaTemplate.send(topic,JSON.toJSONString(message));
      logger.info("發(fā)送消息成功 ----->>>>> message = {}", JSON.toJSONString(message));
    } catch (Exception e) {
      e.getMessage();
    }

  }
}

五．發(fā)現(xiàn)監(jiān)聽接收類

@Component
public class KfkaListener {

  private static Logger logger = LoggerFactory.getLogger(KfkaListener.class);

  @KafkaListener(topics = {"hello"})
  public void listen(ConsumerRecord<?, ?> record) {
    Optional<?> kafkaMessage = Optional.ofNullable(record.value());
    if (kafkaMessage.isPresent()) {
      Object message = kafkaMessage.get();
      logger.info("接收消息------------ record =" + record);
      logger.info("接收消息----------- message =" + message);
    }
  }
}

六．定時(shí)發(fā)送信息測(cè)試類

@EnableScheduling
@Component
public class PublisherController {

  private static final Logger log = LoggerFactory.getLogger(PublisherController.class);

  @Autowired
  private KfkaProducer kfkaProducer;

  @Scheduled(fixedRate = 5000)
  public void pubMsg() {
    Message msg=new Message();
    msg.setId(UUID.randomUUID().toString());
    msg.setMsg("發(fā)送這條消息給你，你好?。。。。。?！");
    msg.setSendTime(new Date());
    kfkaProducer.send("hello", msg);;
    log.info("Publisher sendes Topic... ");
  }
}

七．測(cè)試結(jié)果