Kafka官方文档有 https://docs.spring.io/spring-kafka/reference/htmlsingle/
这里是配置文件实现的方式
先引入依赖
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>2.1.0.RELEASE</version> </dependency> 创建 spring-context-kafka-consumer.xml 当然要配置spring扫描该配置文件 配置文件里边内容如下
<!-- 基本的配置参数 可以写成配置文件或者这种${bootstrap.servers} 配置文件获取的 可以区分开发测试环境 --> <bean id="consumerProperties" class="java.util.HashMap"> <constructor-arg> <map> <entry key="bootstrap.servers" value="${bootstrap.servers}" /> <entry key="group.id" value="0" /> <entry key="enable.auto.commit" value="true" /> <entry key="auto.commit.interval.ms" value="1000" /> <entry key="session.timeout.ms" value="15000" /> <entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer" /> <entry key="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer" /> </map> </constructor-arg> </bean>
<!-- 创建工厂 然后把配置信息注入--> <bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory"> <constructor-arg> <ref bean="consumerProperties" /> </constructor-arg> </bean>
<!-- 把实际消费的类关联进来 --> <bean id="messageListernerConsumerService" class="com.test.kafkaConsumer.KafkaConsumer" />
<!-- 然后把这个类和消费的topic注入这个container topic也配置成灵活的 --> <bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties"> <constructor-arg name="topics" value="${topic}"/> <property name="messageListener" ref="messageListernerConsumerService" /> </bean>
<!-- 把这个container和factory 注入 --> <bean id="messageListenerContainer" class="org.springframework.kafka.listener.KafkaMessageListenerContainer" init-method="doStart"> <constructor-arg ref="consumerFactory" /> <constructor-arg ref="containerProperties" /> </bean>
<!-- 这个可以配置一个类消费多个topic 如果需要不同的类消费不同的topic 就配置多个container关联不通的类 --> 消费的类
public class KafkaConsumer implements MessageListener<Integer, String> { @Override public void onMessage(ConsumerRecord<Integer, String> record) { String value = record.value(); }
//因为配置文件已经关联这个类 所以只要启动spring项目 就可以监听消费配置的topic value就是推送过来的消息 }
本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。
我来说两句