How to configure dual kafka in springboot

springboot configures dual kafka

Use spring boot 2.0.8.RELEASE version

Introduce Maven kafka jar and prepare two kafka;

 org.springframework.kafka spring-kafka 

Configure yml configuration file

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

Configuring KafkaConfig class

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> kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); factory.setConcurrency(3); factory.getContainerProperties().setPollTimeout(3000); return factory; } @Bean//第一个消费者工厂的bean public ConsumerFactory consumerFactory() { return new DefaultKafkaConsumerFactory<>(consumerConfigs()); } @Bean public Map consumerConfigs() { Map 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 producerFactory() { return new DefaultKafkaProducerFactory<>(senderProps()); } @Bean //kafka发送消息模板 public KafkaTemplate kafkaTemplate() { return new KafkaTemplate(producerFactory()); } /** * 生产者配置方法 * * 生产者有三个必选属性 * 
 * 1.bootstrap.servers broker地址清单，清单不要包含所有的broker地址， * 生产者会从给定的broker里查找到其他broker的信息。不过建议至少提供两个broker信息，一旦 其中一个宕机，生产者仍能能够连接到集群上。 * 
 * 
 * 2.key.serializer broker希望接收到的消息的键和值都是字节数组。 生产者用对应的类把键对象序列化成字节数组。 * 
 * 
 * 3.value.serializer 值得序列化方式 * 
 * * * @return */ private Map senderProps() { Map props = new HashMap<>(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, innerServers); /** * 当从broker接收到的是临时可恢复的异常时，生产者会向broker重发消息，但是不能无限 * 制重发，如果重发次数达到限制值，生产者将不会重试并返回错误。 * 通过retries属性设置。默认情况下生产者会在重试后等待100ms，可以通过 retries.backoff.ms属性进行修改 */ props.put(ProducerConfig.RETRIES_CONFIG, 0); /** * 在考虑完成请求之前，生产者要求leader收到的确认数量。这可以控制发送记录的持久性。允许以下设置： * 
 * 
 *  acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认。该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且 * retries 配置将不会生效（因为客户端通常不会知道任何故障）。为每条记录返回的偏移量始终设置为-1。 * 
  acks = 1  * 这意味着leader会将记录写入其本地日志，但无需等待所有follower的完全确认即可做出回应。在这种情况下， * 如果leader在确认记录后立即失败但在关注者复制之前，则记录将丢失。 * 
 acks = all  * 这意味着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> kafkaListenerContainerFactoryOutSchedule() { ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactoryOutSchedule()); factory.setConcurrency(3); factory.getContainerProperties().setPollTimeout(3000); return factory; } @Bean public ConsumerFactory consumerFactoryOutSchedule() { return new DefaultKafkaConsumerFactory<>(consumerConfigsOutSchedule()); } /** * 连接第二个集群的消费者配置 */ @Bean public Map consumerConfigsOutSchedule() { Map 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 producerOutFactory() { return new DefaultKafkaProducerFactory<>(senderOutProps()); } @Bean //kafka发送消息模板 public KafkaTemplate kafkaOutTemplate() { return new KafkaTemplate(producerOutFactory()); } /** * 生产者配置方法 * * 生产者有三个必选属性 * 
 * 1.bootstrap.servers broker地址清单，清单不要包含所有的broker地址， * 生产者会从给定的broker里查找到其他broker的信息。不过建议至少提供两个broker信息，一旦 其中一个宕机，生产者仍能能够连接到集群上。 * 
 * 
 * 2.key.serializer broker希望接收到的消息的键和值都是字节数组。 生产者用对应的类把键对象序列化成字节数组。 * 
 * 
 * 3.value.serializer 值得序列化方式 * 
 * * * @return */ private Map senderOutProps() { Map props = new HashMap<>(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, outServers); /** * 当从broker接收到的是临时可恢复的异常时，生产者会向broker重发消息，但是不能无限 * 制重发，如果重发次数达到限制值，生产者将不会重试并返回错误。 * 通过retries属性设置。默认情况下生产者会在重试后等待100ms，可以通过 retries.backoff.ms属性进行修改 */ props.put(ProducerConfig.RETRIES_CONFIG, 0); /** * 在考虑完成请求之前，生产者要求leader收到的确认数量。这可以控制发送记录的持久性。允许以下设置： * 
 * 
 *  acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认。该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且 * retries 配置将不会生效（因为客户端通常不会知道任何故障）。为每条记录返回的偏移量始终设置为-1。 * 
  acks = 1  * 这意味着leader会将记录写入其本地日志，但无需等待所有follower的完全确认即可做出回应。在这种情况下， * 如果leader在确认记录后立即失败但在关注者复制之前，则记录将丢失。 * 
 acks = all  * 这意味着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; } }

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; /** * 
 * KafkaProducer Description: kafka生产者 * 
 * * @author douzaixingDATE 2019年7月8日 下午4:09:29 */ @Component // 这个必须加入容器不然，不会执行 @EnableScheduling // 这里是为了测试加入定时调度 @Slf4j public class MyKafkaProducer { @Autowired private KafkaTemplate kafkaTemplate; @Autowired private KafkaTemplate kafkaOutTemplate; public ListenableFuture> send(String topic, String key, String json) { ListenableFuture> result = kafkaTemplate.send(topic, key, json); log.info("inner kafka send #topic=" + topic + "#key=" + key + "#json=" + json + "#推送成功==========="); return result; } public ListenableFuture> sendOut(String topic, String key, String json) { ListenableFuture> 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 record) { log.info("inner kafka receive #key=" + record.key() + "#value=" + record.value()); } @KafkaListener(topics={"out_test"}, containerFactory="kafkaListenerContainerFactoryOutSchedule") public void outListener(ConsumerRecord record) { log.info("out kafka receive #key=" + record.key() + "#value=" + record.value()); } }