spring kafka test 使用笔记

约定不等于承诺〃 2021-06-24 13:57 672阅读 0赞

## 添加依赖 ##

testImplementation "org.springframework.kafka:spring-kafka-test"

**配置 KafkaTemplate**

@Configuration
    public class KafkaConfig {
    
    	@Autowired
    	EmbeddedKafkaBroker broker;
    
    	@Bean
    	public KafkaTemplate<String, String> template() {
    		return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(broker)));
    	}
    
    }

## 添加测试 ##

@SpringBootTest
    @EmbeddedKafka
    public class KafkaTest {
    
    	private static final Logger log = LoggerFactory.getLogger(KafkaTest.class);
    
    	@Autowired
    	EmbeddedKafkaBroker broker;
    
    	@Autowired
    	KafkaTemplate<String, String> template;
    
    	@Test
    	void kafka() throws InterruptedException, ExecutionException {
    		log.info("start send");
    
    		ListenableFuture<SendResult<String, String>> future = template
    				.send("topic1", "time: " + System.currentTimeMillis());
    		future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
    
    			@Override
    			public void onFailure(Throwable throwable) {
    				log.error("onFailure", throwable);
    			}
    
    			@Override
    			public void onSuccess(SendResult<String, String> stringStringSendResult) {
    				log.info("onSuccess {}", stringStringSendResult);
    			}
    		});
    
    		log.info("result: {}", future.get());
    
    	}
    
    }

## 使用 @KafkaListener 接收消息 ##

修改注解

// 创建topic
    @EmbeddedKafka(topics = "topic1") 
    // 覆盖配置
    @TestPropertySource(properties = {"spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
    		// 消费更早的消息,因为可能消息已经发送但是bean还没注册
    		"spring.kafka.consumer.auto-offset-reset=earliest"})

添加监听

@KafkaListener(id = "webGroup1", topics = "topic1")
    	public void listen(ConsumerRecord<String, String> input) {
    	log.info("input value: {}", input);
    }