[大数据]Springboot 与kafka简单整合demo

• maven依赖

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

• properties配置

server.port=8881

spring.kafka.bootstrap-servers=localhost:9092

指定默认消费者group id --> 由于在kafka中，同一组中的consumer不会读取到同一个消息，依靠groud.id设置组名
spring.kafka.consumer.group-id=testGroup3

• produce

import com.jifenn.test.kafka.KafkaMsgSend;
import com.jifenn.test.kafka.enums.KafkaMsgTopicEnum;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class MsgSendController {

    @Autowired
    private KafkaMsgSend msgSend;

    @GetMapping(path = "/msg/send")
    public boolean sendMsg(@RequestParam String message) {
        msgSend.sendMessage(KafkaMsgTopicEnum.TEST_TOPIC.getValue(), message);
        return true;
    }

}

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.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;


@Slf4j
@Component
public class KafkaMsgSend {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    public <K, T> void sendMessage(final String topic, final String message) {
        log.info("send kafka message,topic:{},message:{}", topic, message);
        ListenableFuture<SendResult<K, T>> listenableFuture = null;
        listenableFuture = kafkaTemplate.send(topic, message);
        SuccessCallback<SendResult<K, T>> successCallback = new SuccessCallback<SendResult<K, T>>() {
            @Override
            public void onSuccess(SendResult<K, T> result) {
                log.info("发送成功,topic:{},message:{}", topic, message);
            }
        };

        FailureCallback failureCallback = new FailureCallback() {
            @Override
            public void onFailure(Throwable e) {
                log.error("发送失败,topic:{},message:{}", topic, message);
                //todo 记录失败日志
                throw new RuntimeException(e);
            }
        };
        listenableFuture.addCallback(successCallback, failureCallback);
    }

}

• consumer

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;


@Slf4j
@Component
public class KafkaMsgConsumer {

    @KafkaListener(topics = {"kafkaTopicDev"})
    public void receive(List<String> message) {
        log.info("接收到消息：{}", message);
    }
}

• 测试

?最基础的配置就OK了