spring boot 配置双kafka方法

1.使用spring boot 2.0.8.RELEASE 版本

2.引入Maven kafka jar、准备两个kafka；

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

3.配置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

4.配置KafkaConfig类

5.发送工具类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;
    }

}

6.测试类：

@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);
        }
    }
}

7.接收类：

@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());
    }
}

8.测试结果

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