springboot 1.5.2 集成kafka的简单例子

本文介绍了springboot 1.5.2 集成kafka的简单例子 ，分享给大家，具体如下：

随着spring boot 1.5版本的发布，在spring项目中与kafka集成更为简便。

添加依赖

compile("org.springframework.kafka:spring-kafka:1.1.2.release")

添加application.properties

#kafka
# 指定kafka 代理地址，可以多个
spring.kafka.bootstrap-servers=192.168.59.130:9092,192.168.59.131:9092,192.168.59.132:9092
# 指定默认消费者group id
spring.kafka.consumer.group-id=mygroup
# 指定默认topic id
spring.kafka.template.default-topic= my-replicated-topic
# 指定listener 容器中的线程数，用于提高并发量
spring.kafka.listener.concurrency= 3
# 每次批量发送消息的数量
spring.kafka.producer.batch-size= 1000

configuration 启用kafka

package cn.xiaojf.today.data.kafka.configuration;

import org.springframework.context.annotation.configuration;
import org.springframework.kafka.annotation.enablekafka;

/**
 * kafka 配置
 * @author xiaojf 2017/3/24 14:09
 */
@configuration
@enablekafka
public class kafkaconfiguration {


}

消息生产者

package cn.xiaojf.today.data.kafka.producer;

import org.apache.kafka.clients.producer.producer;
import org.apache.kafka.clients.producer.recordmetadata;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.kafka.core.kafkaoperations;
import org.springframework.kafka.core.kafkatemplate;
import org.springframework.kafka.support.producerlistener;
import org.springframework.stereotype.component;

/**
 * 消息生产者
 * @author xiaojf 2017/3/24 14:36
 */
@component
public class msgproducer {
 @autowired
 private kafkatemplate<string,string> kafkatemplate;

 public void send() {
  kafkatemplate.send("my-replicated-topic","xiaojf");
  kafkatemplate.send("my-replicated-topic","xiaojf");

  kafkatemplate.metrics();

  kafkatemplate.execute(new kafkaoperations.producercallback<string, string, object>() {
   @override
   public object doinkafka(producer<string, string> producer) {
    //这里可以编写kafka原生的api操作
    return null;
   }
  });

  //消息发送的监听器，用于回调返回信息
  kafkatemplate.setproducerlistener(new producerlistener<string, string>() {
   @override
   public void onsuccess(string topic, integer partition, string key, string value, recordmetadata recordmetadata) {

   }

   @override
   public void onerror(string topic, integer partition, string key, string value, exception exception) {

   }

   @override
   public boolean isinterestedinsuccess() {
    return false;
   }
  });
 }
}

消息消费者

package cn.xiaojf.today.data.kafka.consumer;

import org.springframework.kafka.annotation.kafkalistener;
import org.springframework.stereotype.component;

/**
 * 消息消费者
 * @author xiaojf 2017/3/24 14:36
 */
@component
public class msgconsumer {
 @kafkalistener(topics = {"my-replicated-topic","my-replicated-topic2"})
 public void processmessage(string content) {
  system.out.println(content);
 }


}

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。