springboot之配置雙kafka全過程
springboot配置雙kafka
使用spring boot 2.0.8.RELEASE 版本
引入Maven kafka jar、準備兩個kafka；
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>
配置yml配置文件
spring: kafka: bootstrap-servers: 180.167.180.242:9092 #kafka的訪問地址，多個用","隔開 consumer: enable-auto-commit: true group-id: kafka #群組ID outkafka: bootstrap-servers: localhost:9092 #kafka的訪問地址，多個用","隔開 consumer: enable-auto-commit: true group-id: kafka_1 #群組ID
配置KafkaConfig類
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
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.context.annotation.Primary;
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.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
@Configuration
@EnableKafka
public class KafkaConfig {
@Value("${spring.kafka.bootstrap-servers}")
private String innerServers;
@Value("${spring.kafka.consumer.group-id}")
private String innerGroupid;
@Value("${spring.kafka.consumer.enable-auto-commit}")
private String innerEnableAutoCommit;
@Bean
@Primary//理解為默認優先選擇當前容器下的消費者工廠
KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setConcurrency(3);
factory.getContainerProperties().setPollTimeout(3000);
return factory;
}
@Bean//第一個消費者工廠的bean
public ConsumerFactory<Integer, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, innerServers);
props.put(ConsumerConfig.GROUP_ID_CONFIG, innerGroupid);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, innerEnableAutoCommit);
// props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
// props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
return props;
}
@Bean //生產者工廠配置
public ProducerFactory<String, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(senderProps());
}
@Bean //kafka發送消息模板
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<String, String>(producerFactory());
}
/**
* 生產者配置方法
*
* 生產者有三個必選屬性
* <p>
* 1.bootstrap.servers broker地址清單，清單不要包含所有的broker地址，
* 生產者會從給定的broker裡查找到其他broker的信息。不過建議至少提供兩個broker信息，一旦 其中一個宕機，生產者仍能能夠連接到集群上。
* </p>
* <p>
* 2.key.serializer broker希望接收到的消息的鍵和值都是字節數組。 生產者用對應的類把鍵對象序列化成字節數組。
* </p>
* <p>
* 3.value.serializer 值得序列化方式
* </p>
*
*
* @return
*/
private Map<String, Object> senderProps() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, innerServers);
/**
* 當從broker接收到的是臨時可恢復的異常時，生產者會向broker重發消息，但是不能無限
* 制重發，如果重發次數達到限制值，生產者將不會重試並返回錯誤。
* 通過retries屬性設置。默認情況下生產者會在重試後等待100ms，可以通過 retries.backoff.ms屬性進行修改
*/
props.put(ProducerConfig.RETRIES_CONFIG, 0);
/**
* 在考慮完成請求之前，生產者要求leader收到的確認數量。這可以控制發送記錄的持久性。允許以下設置：
* <ul>
* <li>
* <code> acks = 0 </ code>如果設置為零，則生產者將不會等待來自服務器的任何確認。該記錄將立即添加到套接字緩沖區並視為已發送。在這種情況下，無法保證服務器已收到記錄，並且
* <code>retries </ code>配置將不會生效（因為客戶端通常不會知道任何故障）。為每條記錄返回的偏移量始終設置為-1。
* <li> <code> acks = 1 </code>
* 這意味著leader會將記錄寫入其本地日志，但無需等待所有follower的完全確認即可做出回應。在這種情況下，
* 如果leader在確認記錄後立即失敗但在關註者復制之前，則記錄將丟失。
* <li><code> acks = all </code>
* 這意味著leader將等待完整的同步副本集以確認記錄。這保證瞭隻要至少一個同步副本仍然存活，記錄就不會丟失。這是最強有力的保證。
* 這相當於acks = -1設置
*/
props.put(ProducerConfig.ACKS_CONFIG, "1");
/**
* 當有多條消息要被發送到統一分區是，生產者會把他們放到統一批裡。kafka通過批次的概念來 提高吞吐量，但是也會在增加延遲。
*/
// 以下配置當緩存數量達到16kb，就會觸發網絡請求，發送消息
// props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
// 每條消息在緩存中的最長時間，如果超過這個時間就會忽略batch.size的限制，由客戶端立即將消息發送出去
// props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
// props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}
@Value("${spring.outkafka.bootstrap-servers}")
private String outServers;
@Value("${spring.outkafka.consumer.group-id}")
private String outGroupid;
@Value("${spring.outkafka.consumer.enable-auto-commit}")
private String outEnableAutoCommit;
static {
}
/**
* 連接第二個kafka集群的配置
*/
@Bean
KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactoryOutSchedule() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactoryOutSchedule());
factory.setConcurrency(3);
factory.getContainerProperties().setPollTimeout(3000);
return factory;
}
@Bean
public ConsumerFactory<Integer, String> consumerFactoryOutSchedule() {
return new DefaultKafkaConsumerFactory<>(consumerConfigsOutSchedule());
}
/**
* 連接第二個集群的消費者配置
*/
@Bean
public Map<String, Object> consumerConfigsOutSchedule() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, outServers);
props.put(ConsumerConfig.GROUP_ID_CONFIG, outGroupid);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, outEnableAutoCommit);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
return props;
}
@Bean //生產者工廠配置
public ProducerFactory<String, String> producerOutFactory() {
return new DefaultKafkaProducerFactory<>(senderOutProps());
}
@Bean //kafka發送消息模板
public KafkaTemplate<String, String> kafkaOutTemplate() {
return new KafkaTemplate<String, String>(producerOutFactory());
}
/**
* 生產者配置方法
*
* 生產者有三個必選屬性
* <p>
* 1.bootstrap.servers broker地址清單，清單不要包含所有的broker地址，
* 生產者會從給定的broker裡查找到其他broker的信息。不過建議至少提供兩個broker信息，一旦 其中一個宕機，生產者仍能能夠連接到集群上。
* </p>
* <p>
* 2.key.serializer broker希望接收到的消息的鍵和值都是字節數組。 生產者用對應的類把鍵對象序列化成字節數組。
* </p>
* <p>
* 3.value.serializer 值得序列化方式
* </p>
*
*
* @return
*/
private Map<String, Object> senderOutProps() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, outServers);
/**
* 當從broker接收到的是臨時可恢復的異常時，生產者會向broker重發消息，但是不能無限
* 制重發，如果重發次數達到限制值，生產者將不會重試並返回錯誤。
* 通過retries屬性設置。默認情況下生產者會在重試後等待100ms，可以通過 retries.backoff.ms屬性進行修改
*/
props.put(ProducerConfig.RETRIES_CONFIG, 0);
/**
* 在考慮完成請求之前，生產者要求leader收到的確認數量。這可以控制發送記錄的持久性。允許以下設置：
* <ul>
* <li>
* <code> acks = 0 </ code>如果設置為零，則生產者將不會等待來自服務器的任何確認。該記錄將立即添加到套接字緩沖區並視為已發送。在這種情況下，無法保證服務器已收到記錄，並且
* <code>retries </ code>配置將不會生效（因為客戶端通常不會知道任何故障）。為每條記錄返回的偏移量始終設置為-1。
* <li> <code> acks = 1 </code>
* 這意味著leader會將記錄寫入其本地日志，但無需等待所有follower的完全確認即可做出回應。在這種情況下，
* 如果leader在確認記錄後立即失敗但在關註者復制之前，則記錄將丟失。
* <li><code> acks = all </code>
* 這意味著leader將等待完整的同步副本集以確認記錄。這保證瞭隻要至少一個同步副本仍然存活，記錄就不會丟失。這是最強有力的保證。
* 這相當於acks = -1設置
*/
props.put(ProducerConfig.ACKS_CONFIG, "1");
/**
* 當有多條消息要被發送到統一分區是，生產者會把他們放到統一批裡。kafka通過批次的概念來 提高吞吐量，但是也會在增加延遲。
*/
// 以下配置當緩存數量達到16kb，就會觸發網絡請求，發送消息
// props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
// 每條消息在緩存中的最長時間，如果超過這個時間就會忽略batch.size的限制，由客戶端立即將消息發送出去
// props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
// props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}
}
發送工具類MyKafkaProducer
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j;
/**
* <p>
* <b>KafkaProducer Description:</b> kafka生產者
* </p>
*
* @author douzaixing<b>DATE</b> 2019年7月8日 下午4:09:29
*/
@Component // 這個必須加入容器不然，不會執行
@EnableScheduling // 這裡是為瞭測試加入定時調度
@Slf4j
public class MyKafkaProducer {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
@Autowired
private KafkaTemplate<String, String> kafkaOutTemplate;
public ListenableFuture<SendResult<String, String>> send(String topic, String key, String json) {
ListenableFuture<SendResult<String, String>> result = kafkaTemplate.send(topic, key, json);
log.info("inner kafka send #topic=" + topic + "#key=" + key + "#json=" + json + "#推送成功===========");
return result;
}
public ListenableFuture<SendResult<String, String>> sendOut(String topic, String key, String json) {
ListenableFuture<SendResult<String, String>> result = kafkaOutTemplate.send(topic, key, json);
log.info("out kafka send #topic=" + topic + "#key=" + key + "#json=" + json + "#推送成功===========");
return result;
}
}
測試類
@Slf4j
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes={OesBcServiceApplication.class})
public class MoreKafkaTest {
@Autowired
private MyKafkaProducer kafkaProducer;
@Test
public void sendInner() {
for (int i = 0; i < 1; i++) {
kafkaProducer.send("inner_test", "douzi" + i, "liyuehua" + i);
kafkaProducer.sendOut("out_test", "douziout" + i, "fanbingbing" + i);
}
}
}
接收類
@Component
@Slf4j
public class KafkaConsumer {
@KafkaListener(topics={"inner_test"}, containerFactory="kafkaListenerContainerFactory")
public void innerlistener(ConsumerRecord<String, String> record) {
log.info("inner kafka receive #key=" + record.key() + "#value=" + record.value());
}
@KafkaListener(topics={"out_test"}, containerFactory="kafkaListenerContainerFactoryOutSchedule")
public void outListener(ConsumerRecord<String, String> record) {
log.info("out kafka receive #key=" + record.key() + "#value=" + record.value());
}
}
測試結果
07-11 12:41:27.811 INFO [com.wondertek.oes.bc.service.send.MyKafkaProducer] – inner kafka send #topic=inner_test#key=douzi0#json=liyuehua0#推送成功===========
07-11 12:41:27.995 INFO [com.wondertek.oes.bc.service.send.KafkaConsumer] – inner kafka receive #key=douzi0#value=liyuehua0
07-11 12:41:28.005 INFO [com.wondertek.oes.bc.service.send.MyKafkaProducer] – out kafka send #topic=out_test#key=douziout0#json=fanbingbing0#推送成功===========
07-11 12:41:28.013 INFO [com.wondertek.oes.bc.service.send.KafkaConsumer] – out kafka receive #key=douziout0#value=fanbingbing0
總結
以上為個人經驗，希望能給大傢一個參考，也希望大傢多多支持WalkonNet。
推薦閱讀：
- Springboot集成Kafka進行批量消費及踩坑點
- 解決kafka消息堆積及分區不均勻的問題
- Java Kafka實現延遲隊列的示例代碼
- springboot項目配置多個kafka的示例代碼
- Java分佈式學習之Kafka消息隊列