Kafka接入

Kafka接入

1、引入依赖

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

2、配置kafka

# Spring
spring: 
  kafka:
    bootstrap-servers: 192.168.133.69:9092
    producer:
      retries: 0
      batch-size: 16384
      buffer-memory: 33554432
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: mf-group
      auto-offset-reset: earliest
      enable-auto-commit: true
      auto-commit-interval: 100
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

3、kafka生产者

package com.unionman.iot.platform.system.kafka;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @description 消息生产者
 * @author junlin.ru
 * @date 2022/4/29 9:49
 **/
@Component
@Slf4j
public class KafkaProducer<T> {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * kafka 发送消息
     * @param obj 消息对象
     */
    public void send(T obj, String topic) {

        String jsonObj = JSON.toJSONString(obj);
        log.info("------------ message = {}", jsonObj);

        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, jsonObj);

        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                log.info("Produce: The message failed to be sent:" + throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                //TODO 业务处理
                log.info("Produce: The message was sent successfully:");
                log.info("Produce: _+_+_+_+_+_+_+ result: " + stringObjectSendResult.toString());
            }
        });
    }
}

4、kafka消费者

package com.unionman.iot.platform.system.kafka;

import com.alibaba.fastjson.JSONObject;
import com.unionman.iot.platform.entity.DeviceMessageMonitor;
import com.unionman.iot.platform.es.DeviceMessageMonitorRepository;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @description kafka生产者
 * @author junlin.ru
 * @date 2022/4/29 10:11
 **/
@Component
@Slf4j
public class KafkaConsumer {

    @Autowired
    private DeviceMessageMonitorRepository deviceMessageMonitorRepository;

    @KafkaListener(topics = {"monitor_msg"})
    public void listen(ConsumerRecord<?, String> record){
        log.debug(record.toString());
        Optional<String> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            String message = kafkaMessage.get();
            DeviceMessageMonitor deviceMessageMonitor = JSONObject.parseObject(message, DeviceMessageMonitor.class);
            deviceMessageMonitorRepository.save(deviceMessageMonitor);
        }
    }

}