开源软件之消息系统-springboot整合kafka

Springboot整合kafka使用Idea创建生产者与消费者工程Idea创建Springboot使用Idea创建生产者与消费者工程选择依赖生产者工程配置生产者工程配置bootstrap-servers：kafka服务IP及端口号key-serializer&value-serializer：发送消息键值序列化类kafka.topic.test：为自定义主题配置（需要在程序中动态加载）Kafka监听配置Kafka默认只监听本机消息Kafka监听其他IP消息在config/perties添加：listeners=PLAINTEXT://192.168.XXX.XXX:9092

红色IP为Kafka所在服务器IP地址，需根据实际填写以上配置后使用命令本机获得消息，需将localhost改为上面IP地址生产者消息发送类 @Component

publicclassKafkaMessageProducer{

privatestaticfinalLoggerlogger=LoggerFactory.getLogger(KafkaMessageProducer.class);

@Resource

privateKafkaTemplate<String,String>kafkaTemplate;//装配Kafka模板Bean

@Value("${kafka.topic.test}")//读取配置文件中Topic的设置

publicStringtopic;

publicvoidsend(){

Stringmessage="HelloWorld---"+System.currentTimeMillis();

//向kafka发送消息

ListenableFuture<SendResult<String,String>>future=kafkaTemplate.send(this.topic,message);

//设置成功与失败的回调方法

future.addCallback(success->logger.info("KafkaMessageProducer发送消息成功！"),

fail->logger.error("KafkaMessageProducer发送消息失败！"));

}

}

生产者消息发送测试类 @RunWith(SpringRunner.class)

@SpringBootTest

publicclassKafkaMessageProducerTest{

@Autowired

KafkaMessageProducerkafkaMessageProducer;

@Test

publicvoidsend(){

for(intindex=0;index<10;index++){

kafkaMessageProducer.send();

}

}

}

从消费者窗口看到程序发出的消息

消费者工程配置消费者工程配置消费者消息接收类 @Component

publicclassKafkaMessageConsumer{

privatestaticfinalLoggerlogger=LoggerFactory.getLogger(KafkaMessageConsumer.class);

@KafkaListener(topics={"${kafka.topic.test}"})

publicvoidreceive(@PayloadStringmessage,@HeadersMessageHeadersheaders){

logger.info("KafkaMessageConsumer接收到消息："+messa