kafka学习(四)---- Kafka整合SpringMVC实例(二)
2016-11-28 15:43
471 查看
摘要: Kafka整合SpringMVC实例
Kafka整合SpringMVC实例,但同时也发现官方文档也不全,所以又用spring简单的实现了一下,感觉这个比使用spring-integration-kafka框架更简单一点,但是需要对kafka作深入的了解,废话不多说直接切入正题。
Zookeeper下载基本使用
2. 安装Kafka
kafka基本概念以及环境搭建
3. 创建spring项目(建议使用maven方式创建)
项目截图(小红叉不影响项目的启动)
pom.xml配置
kafka-producer.xml配置
kafka-consumer.xml配置
其他代码请参看实例源码:源码下载
github下载地址
效果如图:
b、查看kafka控制台信息输出,如下图:
kafka学习(四)---- Kafka整合SpringMVC实例(二)
1、概述
目前没有很好的整合Kafka的案例,自己参考着使用spring-integration-kafka框架写了一个:Kafka整合SpringMVC实例,但同时也发现官方文档也不全,所以又用spring简单的实现了一下,感觉这个比使用spring-integration-kafka框架更简单一点,但是需要对kafka作深入的了解,废话不多说直接切入正题。
2、实例
1. 安装ZookeeperZookeeper下载基本使用
2. 安装Kafka
kafka基本概念以及环境搭建
3. 创建spring项目(建议使用maven方式创建)
项目截图(小红叉不影响项目的启动)
pom.xml配置
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.kafka.demo</groupId> <artifactId>SpringWithKafka</artifactId> <packaging>war</packaging> <version>0.0.1-SNAPSHOT</version> <name>SpringWithKafka Maven Webapp</name> <url>http://maven.apache.org</url> <properties> <spring.version>4.2.5.RELEASE</spring.version> </properties> <dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.11</version> <scope>test</scope> </dependency> <dependency> <groupId>javax.servlet</groupId> <artifactId>javax.servlet-api</artifactId> <version>3.1.0</version> <scope>provided</scope> </dependency> <!-- spring --> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-core</artifactId> <version>${spring.version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-web</artifactId> <version>${spring.version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-webmvc</artifactId> <version>${spring.version}</version> </dependency> <!-- spring-kafka --> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>1.0.3.RELEASE</version> </dependency> <!-- 日志 --> <dependency> <groupId>commons-logging</groupId> <artifactId>commons-logging</artifactId> <version>1.1.1</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> <version>1.7.5</version> </dependency> <dependency> <groupId>javax</groupId> <artifactId>javaee-api</artifactId> <version>7.0</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-core</artifactId> <version>2.7.6</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.7.6</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-annotations</artifactId> <version>2.7.6</version> </dependency> </dependencies> <build> <finalName>SpringWithKafka</finalName> </build> </project>
kafka-producer.xml配置
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> <context:property-placeholder location="classpath:init.properties" /> <!-- 定义producer的参数 --> <bean id="producerProperties" class="java.util.HashMap"> <constructor-arg> <map> <entry key="bootstrap.servers" value="${bootstrap.servers}"/> <entry key="group.id" value="0"/> <entry key="retries" value="10"/> <entry key="batch.size" value="16384"/> <entry key="linger.ms" value="1"/> <entry key="buffer.memory" value="33554432"/> <entry key="key.serializer" value="org.apache.kafka.common.serialization.IntegerSerializer"/> <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,即可使用template的send消息方法 --> <bean id="KafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate"> <constructor-arg ref="producerFactory"/> <constructor-arg name="autoFlush" value="true"/> <property name="defaultTopic" value="myTopic"/> </bean> </beans>
kafka-consumer.xml配置
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> <context:property-placeholder location="classpath:init.properties" /> <!-- 定义consumer的参数 --> <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.IntegerDeserializer"/> <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="messageListernerConsumerService" class="com.kafka.demo.service.KafkaConsumerService"/> <!-- 消费者容器配置信息 --> <bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties"> <constructor-arg value="myTopic"/> <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> </beans>
其他代码请参看实例源码:源码下载
github下载地址
3、实例演示
a、运行项目访问http://localhost:8080/SpringWithKafka/kafka/test //测试地址效果如图:
b、查看kafka控制台信息输出,如下图:
相关文章推荐
- kafka学习(三)----- Kafka整合SpringMVC实例
- zTree树与struts2的整合实例学习
- Flume-Kafka-Storm 整合实例
- dwr学习5-spring与dwr整合(springmvc模式,没有加入annotation)
- SpringMVC与iReport(JasperReports) 5.6整合开发实例
- SpringMVC与iReport(JasperReports) 5.6整合开发实例
- SpringMVC入门学习(三)+Spring应用(整合)
- ibatis学习笔记(五)>>>>>>>ibatis和Spring整合实例
- springMVC学习(S2SH整合)
- springmvc学习笔记(31)——spring整合springmvc
- 【SpringMVC整合MyBatis】提供学习参考的项目源码
- springmvc整合redis架构搭建实例
- springmvc hibernate easyui 整合 完整理解项目创建 内附demo实例
- springMVC整合DWR3.0 与实例
- dwr学习4-spring与dwr整合(非springmvc)
- java struts2入门学习实例--用户注册和用户登录整合
- springMVC+Hibernate3+spring3整合实例,附带基本权限功能及CURD例子
- SpringMVC入门学习(一) 环境搭建+实例演示
- 学习 Activiti5.13与springMVC整合,自学001期
- springMVC学习+登陆小实例讲解