Kafka+SpringBoot 整合使用

相关版本：（非常重要，版本不对后面可能出现各种坑）

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <version>1.5.3.RELEASE</version>
            <exclusions>
                <exclusion>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-starter-logging</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </dependency>

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

        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
            <version>2.8.2</version>
        </dependency>

如果觉得SpringBoot版本不合胃口，可以自行更换版本，但是Spring-Kafka的版本一定要在你SpringBoot的支持范围内，不然是会跑出Exception的，一般情况下这种找不到Broker建议可以配一下你的host文件追加 127.0.0.1 localhost

添加配置文件

# kafka 核心配置
spring.kafka.bootstrap-servers=10.69.216.50:9092,10.69.216.51:9092,10.69.216.52:9092,10.69.216.46:9092
spring.kafka.consumer.group-id=top-api
spring.kafka.consumer.auto-offset-reset=earliest

发送消息

package com.cmrh.touchlog.kafka;

import java.util.Date;
import java.util.UUID;

import com.google.gson.Gson;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
/**
　* @author GuoGuiRong
  * @Description: 生产者，只要用户将数据转发到kafka中去
　* @date 2018/9/11 10:57
　*/
@Component
public class Producer {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Autowired
    Gson gson;

    /**
     * 发送测试消息
     */
    public void sendMessage(){
        Message m = new Message();
        m.setId(System.currentTimeMillis());
        m.setMsg(UUID.randomUUID().toString());
        m.setSendTime(new Date());
        kafkaTemplate.send("touch", gson.toJson(m));
    }

    /**
     * 发送指定的消息
     * @param msgContent 
     */
    public void sendMessage(String msgContent){
        Message m = new Message();
        m.setId(System.currentTimeMillis());
        m.setMsg(msgContent);
        m.setSendTime(new Date());
        kafkaTemplate.send("top-api", gson.toJson(m));
    }
}

接收消息

package com.cmrh.touchlog.kafka;

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 com.google.gson.Gson;

import java.util.Optional;

/**
　* @author GuoGuiRong
  * @Description:
　* @date 2018/9/11 10:59
　*/
@Component
@Slf4j
public class Receiver {

    @Autowired
    Gson gson;

    @KafkaListener(topics = "top-api")
    public void handler(ConsumerRecord<?, ?> record) {

        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            Message message1 = gson.fromJson(message.toString(), Message.class);
            log.info("收到消息message={}",message1);
        }

    }
}

运行结果：

2018-09-11 15:30:27,585  INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 org.springframework.kafka.listener.KafkaMessageListenerContainer  [//]  partitions assigned:[touch-0]
2018-09-11 15:30:27,684  INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 com.cmrh.touchlog.kafka.Receiver  [//]  收到消息message=Message(id=1536650724028, msg=47e05a03-2d23-4c51-9c7b-1fdde25d4d4e, sendTime=Tue Sep 11 15:25:24 CST 2018)
2018-09-11 15:30:27,684  INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 com.cmrh.touchlog.kafka.Receiver  [//]  收到消息message=Message(id=1536650724079, msg=c2076c3e-a0cb-42ca-a743-cf3cef6cf5de, sendTime=Tue Sep 11 15:25:24 CST 2018)
2018-09-11 15:30:27,685  INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 com.cmrh.touchlog.kafka.Receiver  [//]  收到消息message=Message(id=1536650724080, msg=6bbdb3e2-d60d-4875-9216-a1b7b9cbd434, sendTime=Tue Sep 11 15:25:24 CST 2018)
2018-09-11 15:30:27,685  INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 com.cmrh.touchlog.kafka.Receiver  [//]  收到消息message=Message(id=1536650724080, msg=e6dea21b-cae0-40da-a6c3-47fc2c801fbb, sendTime=Tue Sep 11 15:25:24 CST 2018)
2018-09-11 15:30:27,685  INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 com.cmrh.touchlog.kafka.Receiver  [//]  收到消息message=Message(id=1536650724080, msg=82972e6d-692f-4721-963e-69802ef55719, sendTime=Tue Sep 11 15:25:24 CST 2018)
2018-09-11 15:30:27,686  INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 com.cmrh.touchlog.kafka.Receiver  [//]  收到消息message=Message(id=1536650724080, msg=772af06e-b1ce-47b0-930d-d51c752a181a, sendTime=Tue Sep 11 15:25:24 CST 2018)
2018-09-11 15:30:27,686  INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 com.cmrh.touchlog.kafka.Receiver  [//]  收到消息message=Message(id=1536650724080, msg=d2922487-ef5b-4a06-bbb0-7e1a9be76e93, sendTime=Tue Sep 11 15:25:24 CST 2018)
2018-09-11 15:30:27,687  INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 com.cmrh.touchlog.kafka.Receiver  [//]  收到消息message=Message(id=1536650724081, msg=e5de1fa2-0c5e-41cc-806e-27868ceb34d8, sendTime=Tue Sep 11 15:25:24 CST 2018)
2018-09-11 15:30:27,688  INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 com.cmrh.touchlog.kafka.Receiver  [//]  收到消息message=Message(id=1536650724081, msg=72087eb6-62b4-40ef-9a83-f7d06e8da556, sendTime=Tue Sep 11 15:25:24 CST 2018)
2018-09-11 15:30:27,688  INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 com.cmrh.touchlog.kafka.Receiver  [//]  收到消息message=Message(id=1536650724081, msg=26c8a1a1-d153-4eda-a474-b12a5d63aeae, sendTime=Tue Sep 11 15:25:24 CST 2018)
2018-09-11 15:30:27,688  INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 com.cmrh.touchlog.kafka.Receiver  [//]  收到消息message=Message(id=1536650724081, msg=93bafc22-632c-4cf0-b501-bd8aa39e5db2, sendTime=Tue Sep 11 15:25:24 CST 2018)
2018-09-11 15:30:27,689  INFO  org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 com.cmrh.touchlog.kafka.Receiver  [//]  收到消息message=Message(id=1536650724081, msg=9efec4bb-ec32-4507-a518-358728e2e6fe, sendTime=Tue Sep 11 15:25:24 CST 2018)

再次说明，版本不对，努力白费。