- 相关版本：（非常重要，版本不对后面可能出现各种坑）
<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)
再次说明，版本不对，努力白费。
网友评论