kafka與Spring的集成

時間 2019-11-26

標籤 kafka spring 集成欄目 Kafka 简体版

原文原文鏈接

準備工做

kafka版本：kafka_2.10-0.10.1.0java

spring版本：spring4.3git

配置文件

pom文件配置（也能夠直接下載jar包）

Kafka和spring集成的支持類庫，spring和kafka通訊監聽web

1 <dependency>
2 　　<groupId>org.springframework.integration</groupId>
3 　　<artifactId>spring-integration-kafka</artifactId>
4 　　<version>1.3.0.RELEASE</version>
5 </dependency>

kafka發送消息以及接受消息使用的類庫spring

1 <dependency>
2     <groupId>org.apache.kafka</groupId>
3     <artifactId>kafka-clients</artifactId>
4     <version>0.10.1.0</version>
5 </dependency>

使用高版本是由於低版本沒法支持kafka監聽，spring和kafka集成很差apache

1 <dependency>
2     <groupId>org.springframework</groupId>
3     <artifactId>spring-webmvc</artifactId>
4     <version>4.3.0.RELEASE</version>
5 </dependency>

kafka自帶監聽器，依賴於spring，因此須要和pring-integration-kafka結合使用json

1 <dependency>
2     <groupId>org.springframework.kafka</groupId>
3     <artifactId>spring-kafka</artifactId>
4     <version>1.0.0.RC1</version>
5 </dependency>

producer配置

　　1.若是你的topic沒有設置名稱，按照默認的topic的名字生成對應的數據文件夾。bootstrap

　　2.producerListener用來判斷kafka發送數據是否成功以及發送反饋信息。api

 1 <?xml version="1.0" encoding="UTF-8"?>
 2 <beans xmlns="http://www.springframework.org/schema/beans"
 3     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
 4     xsi:schemaLocation="http://www.springframework.org/schema/beans
 5          http://www.springframework.org/schema/beans/spring-beans.xsd
 6          http://www.springframework.org/schema/context
 7          http://www.springframework.org/schema/context/spring-context.xsd">
 8 
 9     <!-- 定義producer的參數 -->
10     <bean id="producerProperties" class="java.util.HashMap">
11         <constructor-arg>
12             <map>
13                 <entry key="bootstrap.servers" value="localhost:7000" />
14                 <entry key="group.id" value="0" />
15                 <entry key="retries" value="1" />
16                 <entry key="batch.size" value="16384" />
17                 <entry key="linger.ms" value="1" />
18                 <entry key="buffer.memory" value="33554432" />
19                 <entry key="key.serializer"
20                 value="org.apache.kafka.common.serialization.StringSerializer" />
21                 <entry key="value.serializer"
22                 value="org.apache.kafka.common.serialization.StringSerializer" />
23             </map>
24         </constructor-arg>
25     </bean>
26 
27     <!-- 建立kafkatemplate須要使用的producerfactory bean -->
28     <bean id="producerFactory"
29         class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
30         <constructor-arg>
31             <ref bean="producerProperties" />
32         </constructor-arg>
33     </bean>
34 
35     <!-- 建立kafkatemplate bean，使用的時候，只須要注入這個bean，便可使用template的send消息方法 -->
36     <bean id="KafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">
37         <constructor-arg ref="producerFactory" />
38         <constructor-arg name="autoFlush" value="true" />
39         <property name="defaultTopic" value="defaultTopic" />
40         <property name="producerListener" ref="producerListener"/>
41     </bean>
42     
43     <bean id="producerListener" class="com.git.kafka.producer.KafkaProducerListener" /> 
44 </beans>

consumer配置

　　1.使用kafka的listener進行消息消費監聽，若是有消費消息進入會自動調用OnMessage方法進行消息消費以及後續業務處理。session

　　2.若是要配置多個topic，須要建立新的消費者容器，而後統一指向listner的消息處理類，統一讓這個類進行後續業務處理。mvc

 1 <?xml version="1.0" encoding="UTF-8"?>
 2 <beans xmlns="http://www.springframework.org/schema/beans"
 3      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 4      xmlns:context="http://www.springframework.org/schema/context"
 5      xsi:schemaLocation="http://www.springframework.org/schema/beans 
 6      http://www.springframework.org/schema/beans/spring-beans-3.0.xsd 
 7      http://www.springframework.org/schema/tx 
 8      http://www.springframework.org/schema/tx/spring-tx-3.0.xsd 
 9      http://www.springframework.org/schema/jee 
10      http://www.springframework.org/schema/jee/spring-jee-3.0.xsd 
11      http://www.springframework.org/schema/context 
12       http://www.springframework.org/schema/context/spring-context-3.0.xsd">
13        
14     
15     <!-- 定義consumer的參數 -->
16      <bean id="consumerProperties" class="java.util.HashMap">
17          <constructor-arg>
18              <map>
19                  <entry key="bootstrap.servers" value="127.0.0.1:7000"/>
20                  <entry key="group.id" value="0"/>
21                  <entry key="enable.auto.commit" value="false"/>
22                  <entry key="auto.commit.interval.ms" value="1000"/>
23                  <entry key="session.timeout.ms" value="15000"/>
24                  <entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
25                  <entry key="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
26              </map>
27          </constructor-arg>
28      </bean>
29      
30      <!-- 建立consumerFactory bean -->
31      <bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
32          <constructor-arg>
33              <ref bean="consumerProperties"/>
34          </constructor-arg>
35      </bean>
36      
37      <!-- 實際執行消息消費的類 -->
38      <bean id="messageListernerConsumerService" class="com.git.kafka.consumer.KafkaConsumerServer"/>
39      
40      <!-- 消費者容器配置信息 -->
41      <bean id="containerProperties_trade" class="org.springframework.kafka.listener.config.ContainerProperties">
42          <constructor-arg value="order_test_topic"/>
43          <property name="messageListener" ref="messageListernerConsumerService"/>
44      </bean>
45      <bean id="containerProperties_other" class="org.springframework.kafka.listener.config.ContainerProperties">
46          <constructor-arg value="other_test_topic"/>
47          <property name="messageListener" ref="messageListernerConsumerService"/>
48      </bean>
49      
50      <!-- 建立messageListenerContainer bean，使用的時候，只須要注入這個bean -->
51      <bean id="messageListenerContainer_trade" class="org.springframework.kafka.listener.KafkaMessageListenerContainer" 
52          init-method="doStart">
53          <constructor-arg ref="consumerFactory"/>
54          <constructor-arg ref="containerProperties_trade"/>
55      </bean>
56      
57      <bean id="messageListenerContainer_other" class="org.springframework.kafka.listener.KafkaMessageListenerContainer" 
58          init-method="doStart">
59          <constructor-arg ref="consumerFactory"/>
60          <constructor-arg ref="containerProperties_other"/>
61      </bean>
62      
63 </beans>

applicationContext配置

1 <import resource="classpath:kafkaConsumer.xml" />
2 <import resource="classpath:kafkaProducer.xml" />

具體實現

constant.java　　//常量類

 1 package com.git.kafka.constant;
 2 
 3 /**
 4  * kafkaMessageConstant
 5  * @author wangb
 6  *
 7  */
 8 public class KafkaMesConstant {
 9 
10     public static final String SUCCESS_CODE = "00000";
11     public static final String SUCCESS_MES = "成功";
12     
13     /*kakfa-code*/
14     public static final String KAFKA_SEND_ERROR_CODE = "30001";
15     public static final String KAFKA_NO_RESULT_CODE = "30002";
16     public static final String KAFKA_NO_OFFSET_CODE = "30003";
17     
18     /*kakfa-mes*/
19     public static final String KAFKA_SEND_ERROR_MES = "發送消息超時,聯繫相關技術人員";
20     public static final String KAFKA_NO_RESULT_MES = "未查詢到返回結果,聯繫相關技術人員";
21     public static final String KAFKA_NO_OFFSET_MES = "未查到返回數據的offset,聯繫相關技術人員";
22     
23     
24 }

KafkaConsumerServer.java　　//消費者監聽

 1 package com.git.kafka.consumer;
 2 
 3 import org.apache.kafka.clients.consumer.ConsumerRecord;
 4 import org.slf4j.Logger;
 5 import org.slf4j.LoggerFactory;
 6 import org.springframework.kafka.listener.MessageListener;
 7 
 8 /**
 9  * kafka監聽器啓動
10  * 自動監聽是否有消息須要消費
11  * @author wangb
12  *
13  */
14 public class KafkaConsumerServer implements MessageListener<String, String> {
15     protected final Logger LOG = LoggerFactory.getLogger("kafkaConsumer");
16     /**
17      * 監聽器自動執行該方法
18      *     消費消息
19      *     自動提交offset
20      *     執行業務代碼
21      *     （high level api 不提供offset管理，不能指定offset進行消費）
22      */
23     @Override
24     public void onMessage(ConsumerRecord<String, String> record) {
25         LOG.info("=============kafkaConsumer開始消費=============");
26         String topic = record.topic();
27         String key = record.key();
28         String value = record.value();
29         long offset = record.offset();
30         int partition = record.partition();
31         LOG.info("-------------topic:"+topic);
32         LOG.info("-------------value:"+value);
33         LOG.info("-------------key:"+key);
34         LOG.info("-------------offset:"+offset);
35         LOG.info("-------------partition:"+partition);
36         LOG.info("~~~~~~~~~~~~~kafkaConsumer消費結束~~~~~~~~~~~~~");
37     }
38 
39 }

kafkaProducerListener.java　　//生產者監聽-打印日誌

package com.git.kafka.producer;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.support.ProducerListener;

/**
 * kafkaProducer監聽器，在producer配置文件中開啓
 * @author wangb
 *
 */
@SuppressWarnings("rawtypes")
public class KafkaProducerListener implements ProducerListener{
    protected final Logger LOG = LoggerFactory.getLogger("kafkaProducer");
    /**
     * 發送消息成功後調用
     */
    @Override
    public void onSuccess(String topic, Integer partition, Object key,
            Object value, RecordMetadata recordMetadata) {
        LOG.info("==========kafka發送數據成功（日誌開始）==========");
        LOG.info("----------topic:"+topic);
        LOG.info("----------partition:"+partition);
        LOG.info("----------key:"+key);
        LOG.info("----------value:"+value);
        LOG.info("----------RecordMetadata:"+recordMetadata);
        LOG.info("~~~~~~~~~~kafka發送數據成功（日誌結束）~~~~~~~~~~");
    }

    /**
     * 發送消息錯誤後調用
     */
    @Override
    public void onError(String topic, Integer partition, Object key,
            Object value, Exception exception) {
        LOG.info("==========kafka發送數據錯誤（日誌開始）==========");
        LOG.info("----------topic:"+topic);
        LOG.info("----------partition:"+partition);
        LOG.info("----------key:"+key);
        LOG.info("----------value:"+value);
        LOG.info("----------Exception:"+exception);
        LOG.info("~~~~~~~~~~kafka發送數據錯誤（日誌結束）~~~~~~~~~~");
        exception.printStackTrace();
    }

    /**
     * 方法返回值表明是否啓動kafkaProducer監聽器
     */
    @Override
    public boolean isInterestedInSuccess() {
        LOG.info("///kafkaProducer監聽器啓動///");
        return true;
    }

}

KafkaProducerServer.java　　//生產者

package com.git.kafka.producer;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutionException;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import com.alibaba.fastjson.JSON;
import com.git.kafka.constant.KafkaMesConstant;

/**
 * kafkaProducer模板
 *     使用此模板發送消息
 * @author wangb
 *
 */
@Component
public class KafkaProducerServer{

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    /**
     * kafka發送消息模板
     * @param topic 主題
     * @param value    messageValue
     * @param ifPartition 是否使用分區 0是\1不是
     * @param partitionNum 分區數 若是是否使用分區爲0,分區數必須大於0
     * @param role 角色:bbc app erp...
     */
    public Map<String,Object> sndMesForTemplate(String topic, Object value, String ifPartition, 
            Integer partitionNum, String role){
        String key = role+"-"+value.hashCode();
        String valueString = JSON.toJSONString(value);
        if(ifPartition.equals("0")){
            //表示使用分區
            int partitionIndex = getPartitionIndex(key, partitionNum);
            ListenableFuture<SendResult<String, String>> result = kafkaTemplate.send(topic, partitionIndex, key, valueString);
            Map<String,Object> res = checkProRecord(result);
            return res;
        }else{
            ListenableFuture<SendResult<String, String>> result = kafkaTemplate.send(topic, key, valueString);
            Map<String,Object> res = checkProRecord(result);
            return res;
        }
    }

    /**
     * 根據key值獲取分區索引
     * @param key
     * @param partitionNum
     * @return
     */
    private int getPartitionIndex(String key, int partitionNum){
        if (key == null) {
            Random random = new Random();
            return random.nextInt(partitionNum);
        }
        else {
            int result = Math.abs(key.hashCode())%partitionNum;
            return result;
        }
    }
    
    /**
     * 檢查發送返回結果record
     * @param res
     * @return
     */
    @SuppressWarnings("rawtypes")
    private Map<String,Object> checkProRecord(ListenableFuture<SendResult<String, String>> res){
        Map<String,Object> m = new HashMap<String,Object>();
        if(res!=null){
            try {
                SendResult r = res.get();//檢查result結果集
                /*檢查recordMetadata的offset數據，不檢查producerRecord*/
                Long offsetIndex = r.getRecordMetadata().offset();
                if(offsetIndex!=null && offsetIndex>=0){
                    m.put("code", KafkaMesConstant.SUCCESS_CODE);
                    m.put("message", KafkaMesConstant.SUCCESS_MES);
                    return m;
                }else{
                    m.put("code", KafkaMesConstant.KAFKA_NO_OFFSET_CODE);
                    m.put("message", KafkaMesConstant.KAFKA_NO_OFFSET_MES);
                    return m;
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                m.put("code", KafkaMesConstant.KAFKA_SEND_ERROR_CODE);
                m.put("message", KafkaMesConstant.KAFKA_SEND_ERROR_MES);
                return m;
            } catch (ExecutionException e) {
                e.printStackTrace();
                m.put("code", KafkaMesConstant.KAFKA_SEND_ERROR_CODE);
                m.put("message", KafkaMesConstant.KAFKA_SEND_ERROR_MES);
                return m;
            }
        }else{
            m.put("code", KafkaMesConstant.KAFKA_NO_RESULT_CODE);
            m.put("message", KafkaMesConstant.KAFKA_NO_RESULT_MES);
            return m;
        }
    }
    

}

KafkaProducerTest.java　　//kafka生產者測試（消費者使用spring啓動監聽，自動執行onMessage方法）

package com.git.test;

import java.util.Map;

import com.git.kafka.producer.KafkaProducerServer;

public class KafkaProducerTest {
    public static void main(String[] args) {
        
        KafkaProducerServer kafkaProducer = new KafkaProducerServer();
        String topic = "orderTopic";
        String value = "test";
        String ifPartition = "0";
        Integer partitionNum = 3;
        String role = "test";//用來生成key
        Map<String,Object> res = kafkaProducer.sndMesForTemplate
                (topic, value, ifPartition, partitionNum, role);
        
        System.out.println("測試結果以下：===============");
        String message = (String)res.get("message");
        String code = (String)res.get("code");
        
        System.out.println("code:"+code);
        System.out.println("message:"+message);
    }
}