天天看点
返回

kafka_0.10.1.0与Spring4的集成

准备工作

kafka版本：kafka_2.10-0.10.1.0

spring版本：spring4.3

pom文件配置 

org.springframework.kafka
   
            
   
    spring-kafka
     
   
            
   
    1.1.1.RELEASE
   
        
  
        
  
            
   
    org.apache.kafka
   
            
   
    kafka-clients
   
            
   
    0.10.1.0
   
        
  
        
  
            
   
    org.springframework.integration
   
            
   
    spring-integration-kafka
   
            
   
    1.3.0.RELEASE

producer配置

　　 

<?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">
 
    <!-- 定义producer的参数 -->
    <bean id="producerProperties" class="java.util.HashMap">
        <constructor-arg>
            <map>
                <entry key="bootstrap.servers" value="192.168.226.129:9092" />
                <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.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" />
        <constructor-arg name="autoFlush" value="true" />
        <property name="defaultTopic" value="orderTopic" />
        <property name="producerListener" ref="producerListener"/>
    </bean>

    <bean id="producerListener" class="com.willow.kafka.KafkaProducerListener" />
</beans>

生产者监听器 

package com.willow.kafka;

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

/**
 * kafkaProducer监听器，在producer配置文件中开启
 * @author 
 *
 */
@SuppressWarnings("rawtypes")
public class KafkaProducerListener implements ProducerListener{
    protected final Logger LOG = LoggerFactory.getLogger("kafkaProducer");
    /**
     * 发送消息成功后调用
     */
    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发送数据成功（日志结束）~~~~~~~~~~");
    }

    /**
     * 发送消息错误后调用
     */
    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监听器
     */
    public boolean isInterestedInSuccess() {
        LOG.info("///kafkaProducer监听器启动///");
        return true;
    }

}

生产者发送消息 

package com.willow.kafka;

import com.alibaba.fastjson.JSON;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.concurrent.ListenableFuture;

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

/**
 * kafkaProducer模板
 *     使用此模板发送消息
 * @author  
 *
 */
@Service
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath*:spring/spring-kafka-provider.xml")
public class KafkaProducerServer{

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Test
    public void testTemplateSend() {
        kafkaTemplate.send("test-topic", "www.656463.com");
    }


    @Test
    public void orderTopic(){
        KafkaProducerServer kafkaProducerServer=new KafkaProducerServer();
        String topic = "orderTopic";
        String value = "发送的内容你看看能收到吗？亲爱的";
        String ifPartition = "1";
        Integer partitionNum = 1;
        String role = "test";//用来生成key
       /* Map<String, Object> res = kafkaProducerServer.sndMesForTemplate
                (topic, value, ifPartition, partitionNum, role);*/
        String key = role+"-"+value.hashCode();
        ListenableFuture<SendResult<String, String>> result = kafkaTemplate.send(topic, key, value);

    }
    /**
     * 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;
        }
    }
    

}

常量类 

package com.willow.kafka;

/**
 * kafkaMessageConstant
 * @author  
 *
 */
public class KafkaMesConstant {

    public static final String SUCCESS_CODE = "00000";
    public static final String SUCCESS_MES = "成功";
    
    /*kakfa-code*/
    public static final String KAFKA_SEND_ERROR_CODE = "30001";
    public static final String KAFKA_NO_RESULT_CODE = "30002";
    public static final String KAFKA_NO_OFFSET_CODE = "30003";
    
    /*kakfa-mes*/
    public static final String KAFKA_SEND_ERROR_MES = "发送消息超时,联系相关技术人员";
    public static final String KAFKA_NO_RESULT_MES = "未查询到返回结果,联系相关技术人员";
    public static final String KAFKA_NO_OFFSET_MES = "未查到返回数据的offset,联系相关技术人员";
    
    
}

消费者配置 

<?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-3.0.xsd 
      ">


    <!-- 定义consumer的参数 -->
    <bean id="consumerProperties" class="java.util.HashMap">
        <constructor-arg>
            <map>
                <entry key="bootstrap.servers" value="192.168.226.129:9092"/>
                <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.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="messageListernerConsumerService" class="com.willow.kafka.KafkaConsumerServer"/>


    <bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties">
        <constructor-arg value="orderTopic"/><!-- 监听的topic ContainerProperties 中构造函数参数，多个toppic 用，隔开 -->
        <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>

消费者监听器 

package com.willow.kafka;

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

/**
 * kafka监听器启动
 * 自动监听是否有消息需要消费
 * @author
 *
 */
public class KafkaConsumerServer implements MessageListener<String, String> {
    protected final Logger LOG = LoggerFactory.getLogger("KafkaConsumerServer");

   

    /**
     * 监听器自动执行该方法
     *     消费消息
     *     自动提交offset
     *     执行业务代码
     *     （high level api 不提供offset管理，不能指定offset进行消费）
     */
    public void onMessage(ConsumerRecord<String, String> record)  {
        try {
            LOG.info("=============kafkaConsumer开始消费=============");
            String topic = record.topic();
            String key = record.key();
            String value = record.value();
            long offset = record.offset();
            int partition = record.partition();
            LOG.info("-------------topic:" + topic);
            LOG.info("-------------value:" + value);
            LOG.info("-------------key:" + key);
            LOG.info("-------------offset:" + offset);
            LOG.info("-------------partition:" + partition);
            LOG.info("~~~~~~~~~~~~~kafkaConsumer消费结束~~~~~~~~~~~~~");
        }catch (Exception e){
            LOG.info("*************");
        }
    }

}

测试类 

package com.willow.kafka;

import org.springframework.context.support.ClassPathXmlApplicationContext;

public class KafkaProducerTest {
    public static void main(String[] args) {
      ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{
              "classpath:spring/spring-kafka-consumer.xml", "classpath:spring/spring-kafka-provider.xml"
      });
      context.start();
      while (true) {
        try {
          Thread.sleep(Long.MAX_VALUE);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }
}
mq spring-kafka 集成
上一篇: 精华,一张图进阶 RocketMQ整体架构元数据管理消息收发示例RocketMQ 部署消息发送消息接收
下一篇: 中间件和分布式架构 一、中间件二、单体架构与分布式 三、基于消息中间件的分布式架构 四、消息中间件的核心组成部分

继续阅读