spring+Kafka+springmvc Demo

1、下載(官網下載:http://kafka.apache.org/downloads.html)

任意下載一個zip解壓

2、製做demo示例，spring+springmvc+Kafka

開發前請先安裝zookeeper，傳送門 

1. 啓動kafka服務：安裝目錄下.\bin\windows\kafka-server-start.bat .\config\server.properties

啓動zookeeper：進入zk的安裝文件夾bin目錄下雙擊zkServer.bat文件（由於kafka的topic須要在zk註冊中心註冊）

2.建立topic：bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

3.引入maven依賴，其中引入了多餘的依賴，直接從筆者的demo項目中複製過來

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.10</artifactId>
    <version>0.8.2.1</version>
    <exclusions>
        <exclusion>
            <artifactId>jmxri</artifactId>
            <groupId>com.sun.jmx</groupId>
        </exclusion>
        <exclusion>
            <artifactId>jms</artifactId>
            <groupId>javax.jms</groupId>
        </exclusion>
        <exclusion>
            <artifactId>jmxtools</artifactId>
            <groupId>com.sun.jdmk</groupId>
        </exclusion>
    </exclusions>
</dependency>

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>1.1.1.RELEASE</version>
</dependency>


<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.9.0.1</version>
</dependency>複製代碼

複製代碼

<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-web</artifactId>
    <version>4.3.10.RELEASE</version>
</dependency>

<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-webmvc</artifactId>
    <version>4.3.14.RELEASE</version>
</dependency>

<dependency>
    <groupId>org.freemarker</groupId>
    <artifactId>freemarker</artifactId>
    <version>2.3.22</version>
</dependency>

<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-context-support</artifactId>
    <version>4.1.6.RELEASE</version>
</dependency>

<dependency>
    <groupId>com.github.miemiedev</groupId>
    <artifactId>mybatis-paginator</artifactId>
    <version>1.2.15</version>
</dependency>

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-core</artifactId>
    <version>2.8.9</version>
</dependency>
<dependency>
    <groupId>carhouse-test</groupId>
    <artifactId>carhouse-test-api</artifactId>
    <version>1.0-SNAPSHOT</version>
</dependency>

<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>dubbo</artifactId>
    <version>2.5.3</version>
    <exclusions>
        <exclusion>
            <artifactId>spring</artifactId>
            <groupId>org.springframework</groupId>
        </exclusion>
    </exclusions>
</dependency>

<dependency>
    <groupId>com.101tec</groupId>

    <artifactId>zkclient</artifactId>
    <version>0.10</version>
</dependency>

<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <version>1.7.5.RELEASE</version>
</dependency>複製代碼

複製代碼

4.producer配置(kafka-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>複製代碼

5.consumer配置(kafka-consumer.xml)

<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>複製代碼

5.在spring-mvc.xml的配置

<!-- 使用zookeeper註冊中心暴露服務地址 -->
<dubbo:application name="test-provider"/>
<dubbo:registry protocol="zookeeper" address="zookeeper://127.0.0.1:2181"/>複製代碼

<!-- 引入kafka配置文件，根據我的文件位置-->
<import resource="classpath:./kafka/kafka-producer.xml"/>
<import resource="classpath:./kafka/kafka-consumer.xml"/>
<context:annotation-config />
複製代碼

6.實際使用

最簡單的一條消息發送

@Resource
private KafkaTemplate<Integer, String> kafkaTemplate;

@RequestMapping(value = "/hello.do")
public void hello(){

    kafkaTemplate.sendDefault("test it");
    
}複製代碼

消息消費

package mq.kafka;


import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.MessageListener;

public class KafkaConsumerListener implements MessageListener<Integer, String> {



    public void onMessage(ConsumerRecord<Integer, String> consumerRecord) {

        Object o = consumerRecord.value();
        System.out.println(String.valueOf(o));
    }
}複製代碼

最後會在控制檯輸出「test it」

寫的有什麼不對的，歡迎各位指出。