验证中...
开源中国 2018 年度最后一场技术盛会邀你来约~错过就要等明年啦!点此立即预约
语言: Java
分类: 编程语言基础
最后更新于 2018-12-13 09:07
片段 1 片段 2 片段 3
producer.xml
原始数据 复制代码
<!--基本配置 -->
<bean id="producerProperties" class="java.util.HashMap">
<constructor-arg>
<map>
<!-- kafka服务地址,可能是集群-->
<entry key="bootstrap.servers" value="localhost:9092,localhost:9093,localhost:9094" />
<!-- 有可能导致broker接收到重复的消息,默认值为3-->
<entry key="retries" value="10" />
<!-- 每次批量发送消息的数量-->
<entry key="batch.size" value="1638" />
<!-- 默认0ms,在异步IO线程被触发后(任何一个topic,partition满都可以触发)-->
<entry key="linger.ms" value="1" />
<!--producer可以用来缓存数据的内存大小。如果数据产生速度大于向broker发送的速度,producer会阻塞或者抛出异常 -->
<entry key="buffer.memory" value="33554432 " />
<!-- producer需要server接收到数据之后发出的确认接收的信号,此项配置就是指procuder需要多少个这样的确认信号-->
<entry key="acks" value="all" />
<entry key="key.serializer" value="org.apache.kafka.common.serialization.StringSerializer" />
<entry key="value.serializer" value="org.apache.kafka.common.serialization.StringSerializer" />
</map>
</constructor-arg>
</bean>
<!-- 创建kafkatemplate需要使用的producerfactory bean -->
<bean id="producerFactory"
class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
<constructor-arg>
<ref bean="producerProperties" />
</constructor-arg>
</bean>
<!-- 创建kafkatemplate bean,使用的时候,只需要注入这个bean,即可使用template的send消息方法 -->
<bean id="KafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">
<constructor-arg ref="producerFactory" />
<!--设置对应topic-->
<property name="defaultTopic" value="test" />
</bean>
consumer.java
原始数据 复制代码
<bean id="consumerProperties" class="java.util.HashMap">
<constructor-arg>
<map>
<!--Kafka服务地址 -->
<entry key="bootstrap.servers" value="localhost:9092" />
<!--Consumer的组ID,相同goup.id的consumer属于同一个组。 -->
<entry key="group.id" value="order-beta" />
<!--如果此值设置为true,consumer会周期性的把当前消费的offset值保存到zookeeper。当consumer失败重启之后将会使用此值作为新开始消费的值。 -->
<entry key="enable.auto.commit" value="true" />
<!--网络请求的socket超时时间。实际超时时间由max.fetch.wait + socket.timeout.ms 确定 -->
<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 -->
<bean id="messageListernerConsumerService" class="mq.kafka.KafkaConsumerListener" />
<!-- 创建consumerFactory bean -->
<bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
<constructor-arg>
<ref bean="consumerProperties"/>
</constructor-arg>
</bean>
<bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties">
<constructor-arg value="test"/>
<property name="messageListener" ref="messageListernerConsumerService"/>
</bean>
<bean id="messageListenerContainer" class="org.springframework.kafka.listener.KafkaMessageListenerContainer" init-method="doStart">
<constructor-arg ref="consumerFactory"/>
<constructor-arg ref="containerProperties"/>
</bean>
步骤说明
原始数据 复制代码
producer:
1.创建属性配置的bean
2.创建producerFactory,通过第一步的属性 bean进行注入。
3.创建kafkaTemplate,通过第二步的factory进行注入,同时设置producer进行传递的topic,通过参数defaultTopic进行设置。
consumer:
1.创建属性配置的bean
2.容器注入监听的bean
3.创建consumerFactory,通过第一步的属性进行注入配置
4.监听容器属性的配置,包括监听的topic以及监听的bean配置
5.监听容器的配置,通过加入容器属性以及consumerFactory

评论列表( 0 )

你可以在登录后,发表评论

搜索帮助