這篇文章主要介紹了Springboot集成Kafka實現producer和consumer的示例代碼,詳細的介紹了什麼是Kafka和安裝Kafka以及在springboot項目中集成kafka收發message,感興趣的小夥伴們可以參考一下
本文介紹如何在springboot項目中集成kafka收發message。
Kafka是一種高吞吐量的分佈式發佈訂閱消息系統,有如下特性: 通過O(1)的磁盤數據結構提供消息的持久化,這種結構對於即使數以TB的消息存儲也能夠保持長時間的穩定性能。高吞吐量:即使是非常普通的硬件Kafka也可以支持每秒數百萬的消息。支持通過Kafka服務器和消費機集群來分區消息。支持Hadoop並行數據加載。
安裝Kafka
因為安裝kafka需要zookeeper的支持,所以Windows安裝時需要將zookeeper先安裝上,然後將kafka安裝好就可以了。 下面我給出Mac安裝的步驟以及需要注意的點吧,windows的配置除了所在位置不太一樣其他幾乎沒什麼不同。
brew install kafka
對,就是這麼簡單,mac上一個命令就可以搞定了,這個安裝過程可能需要等一會兒,應該是和網絡狀況有關係。安裝提示信息可能有錯誤消息,如"Error: Could not link: /usr/local/share/doc/homebrew" 這個沒關係,自動忽略掉了。 最終我們看到下面的樣子就成功咯。
==> Summary ðŸº/usr/local/Cellar/kafka/1.1.0: 157 files, 47.8MB
安裝的配置文件位置如下,根據自己的需要修改端口號什麼的就可以了。
安裝的zoopeeper和kafka的位置 /usr/local/Cellar/
配置文件 /usr/local/etc/kafka/server.properties /usr/local/etc/kafka/zookeeper.properties
啟動zookeeper
啟動kafka
./bin/kafka-server-start /usr/local/etc/kafka/server.properties &
為kafka創建Topic,topic 名為test,可以配置成自己想要的名字,回頭再代碼中配置正確就可以了。
1、先解決依賴
springboot相關的依賴我們就不提了,和kafka相關的只依賴一個spring-kafka集成包
這裡我們先把配置文件展示一下
#============== kafka ===================
kafka.consumer.zookeeper.connect=10.93.21.21:2181
kafka.consumer.servers=10.93.21.21:9092
kafka.consumer.enable.auto.commit=true
kafka.consumer.session.timeout=6000
kafka.consumer.auto.commit.interval=100
kafka.consumer.auto.offset.reset=latest
kafka.consumer.topic=test
kafka.consumer.group.id=test
kafka.consumer.concurrency=10
kafka.producer.servers=10.93.21.21:9092
kafka.producer.retries=0
kafka.producer.batch.size=4096
kafka.producer.linger=1
kafka.producer.buffer.memory=40960
2、Configuration:Kafka producer
1)通過@Configuration、@EnableKafka,聲明Config並且打開KafkaTemplate能力。
2)通過@Value注入application.properties配置文件中的kafka配置。
3)生成bean,@Bean
package com.kangaroo.sentinel.collect.configuration;
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.beans.factory.annotation.Value;
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 {
@Value("${kafka.producer.servers}")
private String servers;
@Value("${kafka.producer.retries}")
private int retries;
@Value("${kafka.producer.batch.size}")
private int batchSize;
@Value("${kafka.producer.linger}")
private int linger;
@Value("${kafka.producer.buffer.memory}")
private int bufferMemory;
public Map
Map
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
props.put(ProducerConfig.RETRIES_CONFIG, retries);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
props.put(ProducerConfig.LINGER_MS_CONFIG, linger);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}
public ProducerFactory
return new DefaultKafkaProducerFactory<>(producerConfigs());
}
@Bean
public KafkaTemplate
return new KafkaTemplate
}
}
實驗我們的producer,寫一個Controller。想topic=test,key=key,發送消息message
package com.kangaroo.sentinel.collect.controller;
import com.kangaroo.sentinel.common.response.Response;
import com.kangaroo.sentinel.common.response.ResultCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
@RestController
@RequestMapping("/kafka")
public class CollectController {
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private KafkaTemplate kafkaTemplate;
@RequestMapping(value = "/send", method = RequestMethod.GET)
public Response sendKafka(HttpServletRequest request, HttpServletResponse response) {
try {
String message = request.getParameter("message");
logger.info("kafka的消息={}", message);
kafkaTemplate.send("test", "key", message);
logger.info("發送kafka成功.");
return new Response(ResultCode.SUCCESS, "發送kafka成功", null);
} catch (Exception e) {
logger.error("發送kafka失敗", e);
return new Response(ResultCode.EXCEPTION, "發送kafka失敗", null);
}
}
}
3、configuration:kafka consumer
1)通過@Configuration、@EnableKafka,聲明Config並且打開KafkaTemplate能力。
2)通過@Value注入application.properties配置文件中的kafka配置。
3)生成bean,@Bean
package com.kangaroo.sentinel.collect.configuration;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
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.listener.ConcurrentMessageListenerContainer;
import java.util.HashMap;
import java.util.Map;
@Configuration
@EnableKafka
public class KafkaConsumerConfig {
@Value("${kafka.consumer.servers}")
private String servers;
@Value("${kafka.consumer.enable.auto.commit}")
private boolean enableAutoCommit;
@Value("${kafka.consumer.session.timeout}")
private String sessionTimeout;
@Value("${kafka.consumer.auto.commit.interval}")
private String autoCommitInterval;
@Value("${kafka.consumer.group.id}")
private String groupId;
@Value("${kafka.consumer.auto.offset.reset}")
private String autoOffsetReset;
@Value("${kafka.consumer.concurrency}")
private int concurrency;
@Bean
public KafkaListenerContainerFactory
ConcurrentKafkaListenerContainerFactory
factory.setConsumerFactory(consumerFactory());
factory.setConcurrency(concurrency);
factory.getContainerProperties().setPollTimeout(1500);
return factory;
}
public ConsumerFactory
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
public Map
Map
propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
return propsMap;
}
@Bean
public Listener listener() {
return new Listener();
}
}
new Listener()生成一個bean用來處理從kafka讀取的數據。Listener簡單的實現demo如下:只是簡單的讀取並打印key和message值
@KafkaListener中topics屬性用於指定kafka topic名稱,topic名稱由消息生產者指定,也就是由kafkaTemplate在發送消息時指定。
package com.kangaroo.sentinel.collect.configuration;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
public class Listener {
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
@KafkaListener(topics = {"test"})
public void listen(ConsumerRecord, ?> record) {
logger.info("kafka的key: " + record.key());
logger.info("kafka的value: " + record.value().toString());
}
}
tips:
1)我沒有介紹如何安裝配置kafka,配置kafka時最好用完全bind網絡ip的方式,而不是localhost或者127.0.0.1
2)最好不要使用kafka自帶的zookeeper部署kafka,可能導致訪問不通。
3)理論上consumer讀取kafka應該是通過zookeeper,但是這裡我們用的是kafkaserver的地址,為什麼沒有深究。
4)定義監聽消息配置時,GROUP_ID_CONFIG配置項的值用於指定消費者組的名稱,如果同組中存在多個監聽器對象則只有一個監聽器對象能收到消息。
以上就是本文的全部內容,希望對大家的學習有所幫助,喜歡小編文章的可以關注並轉發哦~~~
閱讀更多 架構師公社 的文章