springboot 1.5.3集成kafka_2.10-0.10.2.0

pom.xml

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</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.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <version>1.5.3.RELEASE</version>
        </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
        <version>1.2.1.RELEASE</version>
    </dependency>
    <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-log4j2</artifactId>
            <version>1.5.3.RELEASE</version>
        </dependency>

application.properties

spring.application.name=springboot-kafka-test
#log4j2
logging.config=classpath:log4j2.xml

KafkaProducerConfig.java

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

@Configuration
@EnableKafka
public class KafkaProducerConfig {

    public Map<String, Object> producerConfig(){
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.143.47.32:9092");
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 1000);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 40960);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }
    
    public ProducerFactory<String, String> producerFactory(){
        return new DefaultKafkaProducerFactory<>(producerConfig());
    }
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<String, String>(producerFactory());
    }
}

KafkaConsumerConfig.java

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

import com.yue.practise.consumer.MsgConsumer;

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    public Map<String, Object> consumerConfig(String consumerGroupId){
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.143.47.32:9092");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }
    public ConsumerFactory<String, String> consumerFactory(String consumerGroupId) {
        return new DefaultKafkaConsumerFactory<>(consumerConfig(consumerGroupId));
    }
    
     
     
     @Bean(name="kafkaListenerContainerFactory")
     public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
          factory.setConsumerFactory(consumerFactory("groupA"));
          factory.setConcurrency(3);
          factory.getContainerProperties().setPollTimeout(3000);
          return factory;
     }
     
     @Bean(name="kafkaListenerContainerFactory1")
     public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory1() {
          ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
          factory.setConsumerFactory(consumerFactory("groupB"));
          factory.setConcurrency(3);
          factory.getContainerProperties().setPollTimeout(3000);
          return factory;
     }
}

MsgProducer.java

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class MsgProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    public void send(String value){
        System.out.println("send start-----------");
        kafkaTemplate.send("test", value+"1");
        kafkaTemplate.send("test", "key1", value+"2");
        System.out.println("send end-----------");
    }
}

MsgConsumer.java

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;


@Component
public class MsgConsumer {
    static Logger subscribelogger = LoggerFactory.getLogger("subscribelogger");
    
    @KafkaListener(topics="test",containerFactory="kafkaListenerContainerFactory")
    public void processMsg(ConsumerRecord<?, ?> record){
        subscribelogger.info("{}|{}|{}|{}",record.topic(),record.partition(),record.offset(),record.value());    
    }
    
    @KafkaListener(topics="test",containerFactory="kafkaListenerContainerFactory1")
    public void processMsg1(ConsumerRecord<?, ?> record){
        subscribelogger.info("{}|{}|{}|{}",record.topic(),record.partition(),record.offset(),record.value());    
    }
    
}

參考

spring boot與kafka集成

最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
  • 序言:七十年代末碌上,一起剝皮案震驚了整個濱河市茅坛,隨后出現(xiàn)的幾起案子屑咳,更是在濱河造成了極大的恐慌罢屈,老刑警劉巖后豫,帶你破解...
    沈念sama閱讀 218,451評論 6 506
  • 序言:濱河連續(xù)發(fā)生了三起死亡事件掂之,死亡現(xiàn)場離奇詭異掠哥,居然都是意外死亡霞赫,警方通過查閱死者的電腦和手機(jī)梁棠,發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 93,172評論 3 394
  • 文/潘曉璐 我一進(jìn)店門置森,熙熙樓的掌柜王于貴愁眉苦臉地迎上來,“玉大人符糊,你說我怎么就攤上這事凫海。” “怎么了男娄?”我有些...
    開封第一講書人閱讀 164,782評論 0 354
  • 文/不壞的土叔 我叫張陵行贪,是天一觀的道長。 經(jīng)常有香客問我沪伙,道長瓮顽,這世上最難降的妖魔是什么? 我笑而不...
    開封第一講書人閱讀 58,709評論 1 294
  • 正文 為了忘掉前任围橡,我火速辦了婚禮暖混,結(jié)果婚禮上,老公的妹妹穿的比我還像新娘翁授。我一直安慰自己拣播,他們只是感情好晾咪,可當(dāng)我...
    茶點故事閱讀 67,733評論 6 392
  • 文/花漫 我一把揭開白布。 她就那樣靜靜地躺著贮配,像睡著了一般谍倦。 火紅的嫁衣襯著肌膚如雪。 梳的紋絲不亂的頭發(fā)上泪勒,一...
    開封第一講書人閱讀 51,578評論 1 305
  • 那天昼蛀,我揣著相機(jī)與錄音,去河邊找鬼圆存。 笑死叼旋,一個胖子當(dāng)著我的面吹牛,可吹牛的內(nèi)容都是我干的沦辙。 我是一名探鬼主播夫植,決...
    沈念sama閱讀 40,320評論 3 418
  • 文/蒼蘭香墨 我猛地睜開眼,長吁一口氣:“原來是場噩夢啊……” “哼油讯!你這毒婦竟也來了详民?” 一聲冷哼從身側(cè)響起,我...
    開封第一講書人閱讀 39,241評論 0 276
  • 序言:老撾萬榮一對情侶失蹤陌兑,失蹤者是張志新(化名)和其女友劉穎沈跨,沒想到半個月后,有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體诀紊,經(jīng)...
    沈念sama閱讀 45,686評論 1 314
  • 正文 獨居荒郊野嶺守林人離奇死亡谒出,尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點故事閱讀 37,878評論 3 336
  • 正文 我和宋清朗相戀三年,在試婚紗的時候發(fā)現(xiàn)自己被綠了邻奠。 大學(xué)時的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片笤喳。...
    茶點故事閱讀 39,992評論 1 348
  • 序言:一個原本活蹦亂跳的男人離奇死亡,死狀恐怖碌宴,靈堂內(nèi)的尸體忽然破棺而出杀狡,到底是詐尸還是另有隱情,我是刑警寧澤贰镣,帶...
    沈念sama閱讀 35,715評論 5 346
  • 正文 年R本政府宣布呜象,位于F島的核電站,受9級特大地震影響碑隆,放射性物質(zhì)發(fā)生泄漏恭陡。R本人自食惡果不足惜,卻給世界環(huán)境...
    茶點故事閱讀 41,336評論 3 330
  • 文/蒙蒙 一上煤、第九天 我趴在偏房一處隱蔽的房頂上張望休玩。 院中可真熱鬧,春花似錦、人聲如沸拴疤。這莊子的主人今日做“春日...
    開封第一講書人閱讀 31,912評論 0 22
  • 文/蒼蘭香墨 我抬頭看了看天上的太陽呐矾。三九已至苔埋,卻和暖如春,著一層夾襖步出監(jiān)牢的瞬間蜒犯,已是汗流浹背组橄。 一陣腳步聲響...
    開封第一講書人閱讀 33,040評論 1 270
  • 我被黑心中介騙來泰國打工, 沒想到剛下飛機(jī)就差點兒被人妖公主榨干…… 1. 我叫王不留愧薛,地道東北人晨炕。 一個月前我還...
    沈念sama閱讀 48,173評論 3 370
  • 正文 我出身青樓衫画,卻偏偏與公主長得像毫炉,于是被迫代替她去往敵國和親。 傳聞我的和親對象是個殘疾皇子削罩,可洞房花燭夜當(dāng)晚...
    茶點故事閱讀 44,947評論 2 355

推薦閱讀更多精彩內(nèi)容