Spring boot集成Kafka消息中间件代码实例

Spring boot集成Kafka消息中间件代码实例

一．创建Spring boot项目，添加如下依赖

org.springframework.boot

spring-boot-starter-web

org.projectlombok

lombok

true

org.springframework.kafka

spring-kafka

org.apache.kafka

kafka-clients

com.alibaba

fastjson

1.2.41

二．配置文件

server.port=4400

#kafka配置

#============== kafka ===================

# 指定kafka 代理地址，可以多个

spring.kafka.bootstrap-servers=192.168.102.88:9092

# 指定默认消费者group id

spring.kafka.consumer.group-id=jkafka.demo

#earliest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费

#latest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据

#none topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常

spring.kafka.consumer.auto-offset-reset=latest

spring.kafka.consumer.enable-auto-commit=false

spring.kafka.consumer.auto-commit-interval=100

# 指定消费者消息key和消息体的编解码方式

spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer

spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

三．编辑消息实体

@Data

public class Message implements Serializable{

/**

*

*/

private static final long serialVersionUID = 2522280475099635810L;

//消息ID

private String id;

//消息内容

private String msg;

// 消息发送时间

private Date sendTime;

}

四．消息发送类

@Component

public class KfkaProducer {

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

@Autowired

private KafkaTemplate kafkaTemplate;

public void send(String topic,Message message) {

try {

logger.info("正在发送消息...");

kafkaTemplate.send(topic,JSON.toJSONString(message));

logger.info("发送消息成功 ----->>>>> message = {}", JSON.toJSONString(message));

} catch (Exception e) {

e.getMessage();

}

}

}

五．发现监听接收类

@Component

public class KfkaListener {

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

@KafkaListener(topics = {"hello"})

public void listen(ConsumerRecord, ?> reOEKolWcord) {

Optional> kafkaMessage = Optional.ofNullable(record.value());

if (kafkaMessage.isPresent()) {

Object message = kafkaMessage.get();

logger.info("接收消息------------ record =" + record);

logger.info("接收消息----------- message =" + message);

}

}

}

六．定时发送信息测试类

@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("发送这条消息给你，你好啊！！！！！！");

msg.setSendTime(new Date());

kfkaProduOEKolWcer.send("hello", msg);;

log.info("Publisher sendes Topic... ");

}

}

七．测试结果

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。