首先配置端口号还有主题应
其次，在spring-bean.xml里面加载配置的生产者与消费者
如下：
consumer:消费者的配置
<bean id="consumerProperties" class="java.util.HashMap">
<constructor-arg>
<map>
<!--Kafka服务地址 -->
<entry key="bootstrap.servers" value="192.168.27.128:9092" />
<!--Consumer的组ID，相同group.id的consumer属于同一个组。一个组中的不同成员,相同的消息只能有一个人收到 -->
<entry key="group.id" value="test-consumer-group" />
<!--如果此值设置为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>
<!-- 创建consumerFactory bean -->
<bean id="consumerFactory"
class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
<constructor-arg>
<ref bean="consumerProperties" />
</constructor-arg>
</bean>
<bean id="messageListenerContainer"
class="org.springframework.kafka.listener.KafkaMessageListenerContainer"
init-method="doStart">
<constructor-arg ref="consumerFactory" />
<constructor-arg ref="containerProperties" />
</bean>
<!-- 记得修改主题 -->
<bean id="containerProperties" class="org.springframework.kafka.listener.ContainerProperties">
<!-- 构造函数 就是 主题的参数值 -->
<constructor-arg value="1708E" />
<!-- 要在此处声明自己定义的监听器的类 -->
<property name="messageListener" ref="msgListener" />
</bean>
<!--指定具体监听类的bean 这个类需要我们手工写-->
<bean id="msgListener" class="com.lixuecheng.kafka.MentListener" />
<!--参数配置 -->
<bean id="producerProperties" class="java.util.HashMap">
<constructor-arg>
<map>
<!-- kafka服务地址，可能是集群 value="localhost:9092,localhost:9093,localhost:9094"-->
<entry key="bootstrap.servers" value="192.168.27.128:9092" />
<!-- 有可能导致broker接收到重复的消息-->
<entry key="retries" value="0" />
<!-- 每次批量发送消息的数量 -->
<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 " />
<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="1708E" />
</bean>
然后测试类加载spring-beans.xml文件
在代码中运行测试
要创建一个监听kafka的配置
package com.lixuecheng.kafka;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.listener.MessageListener;
import com.alibaba.fastjson.JSON;
import com.lixuecheng.entity.Investment;
import com.lixuecheng.service.InMentService;
public class MentListener implements MessageListener<String, String> {
@Autowired
private InMentService inMentService;
@Override
public void onMessage(ConsumerRecord<String, String> data) {
String value = data.value();
System.err.println("11111" + value);
System.out.println("111111111111111111111");
try {
Thread.sleep(500);
} catch (InterruptedException e) { // TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println("收到了！！！！");
Investment parseObject = JSON.parseObject(value, Investment.class);
System.out.println(parseObject);
inMentService.add(parseObject);
}
}
最后要在虚拟机中启动kafka及其相关应用即可