Spring Integration Kafka

一、概述 Spring Integration Kafka 是基于 Apache Kafka 和Spring Integration来集成Kafka，对开发配置提供了方便。二、配置 1、spring-kafka-consumer.xml1234567891011121314151

百变码农

3374人浏览 · 2017-07-30 17:57:53

百变码农 · 2017-07-30 17:57:53 发布

转自：http://www.cnblogs.com/yuanermen/p/5453339.html

一、概述

Spring Integration Kafka 是基于 Apache Kafka 和Spring Integration来集成Kafka，对开发配置提供了方便。

二、配置

1、spring-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:int="http://www.springframework.org/schema/integration" 
       
            xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka" 
       
            xmlns:task="http://www.springframework.org/schema/task" 
       
            xsi:schemaLocation="http://www.springframework.org/schema/integration/kafka 
       
            http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd 
       
            http://www.springframework.org/schema/integration 
       
            http://www.springframework.org/schema/integration/spring-integration.xsd 
       
            http://www.springframework.org/schema/beans 
       
            http://www.springframework.org/schema/beans/spring-beans.xsd 
       
            http://www.springframework.org/schema/task 
       
            http://www.springframework.org/schema/task/spring-task.xsd"> 
       
            <!-- topic test conf --> 
       
            <int:channel id="inputFromKafka"> 
       
                <int:dispatcher task-executor="kafkaMessageExecutor"/> 
       
            </int:channel> 
       
            <!-- zookeeper配置 可以配置多个 --> 
       
            <int-kafka:zookeeper-connect id="zookeeperConnect" 
       
                zk-connect="192.168.1.237:2181"zk-connection-timeout="6000" 
       
                zk-session-timeout="6000"zk-sync-time="2000"/> 
       
            <!-- channel配置 auto-startup="true" 否则接收不发数据 --> 
       
            <int-kafka:inbound-channel-adapter id="kafkaInboundChannelAdapter" 
       
                kafka-consumer-context-ref="consumerContext"auto-startup="true"channel="inputFromKafka"> 
       
                <int:poller fixed-delay="1"time-unit="MILLISECONDS"/> 
       
            </int-kafka:inbound-channel-adapter> 
       
            <task:executor id="kafkaMessageExecutor"pool-size="8"keep-alive="120"queue-capacity="500"/> 
       
            <bean id="kafkaDecoder" 
       
                class="org.springframework.integration.kafka.serializer.common.StringDecoder"/> 
       
            <bean id="consumerProperties" 
       
                class="org.springframework.beans.factory.config.PropertiesFactoryBean"> 
       
                <property name="properties"> 
       
                    <props> 
       
                        <prop key="auto.offset.reset">smallest</prop> 
       
                        <prop key="socket.receive.buffer.bytes">10485760</prop> <!-- 10M --> 
       
                        <prop key="fetch.message.max.bytes">5242880</prop> 
       
                        <prop key="auto.commit.interval.ms">1000</prop> 
       
                    </props> 
       
                </property> 
       
            </bean> 
       
            <!-- 消息接收的BEEN --> 
       
            <bean id="kafkaConsumerService"class="com.sunney.service.impl.KafkaConsumerService"/> 
       
            <!-- 指定接收的方法 --> 
       
            <int:outbound-channel-adapter channel="inputFromKafka" 
       
                ref="kafkaConsumerService"method="processMessage"/> 
       
            <int-kafka:consumer-context id="consumerContext" 
       
                consumer-timeout="1000"zookeeper-connect="zookeeperConnect" 
       
                consumer-properties="consumerProperties"> 
       
                <int-kafka:consumer-configurations> 
       
                    <int-kafka:consumer-configuration 
       
                        group-id="default1"value-decoder="kafkaDecoder"key-decoder="kafkaDecoder" 
       
                        max-messages="5000"> 
       
                        <!-- 两个TOPIC配置 --> 
       
                        <int-kafka:topic id="mytopic"streams="4"/> 
       
                        <int-kafka:topic id="sunneytopic"streams="4"/> 
       
                    </int-kafka:consumer-configuration> 
       
                </int-kafka:consumer-configurations> 
       
            </int-kafka:consumer-context> 
       
        </beans>

　　2、spring-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/XMLSchemainstance" 
       
             xmlns:int="http://www.springframework.org/schema/integration" 
       
            xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka" 
       
            xmlns:task="http://www.springframework.org/schema/task" 
       
            xsi:schemaLocation="http://www.springframework.org/schema/integration/kafka        http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd 
       
         http://www.springframework.org/schema/integration  
       
         http://www.springframework.org/schema/integration/spring-integration.xsd 
       
         http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd 
       
         http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd"> 
       
           <!-- commons config --> 
       
            <bean id="stringSerializer"class="org.apache.kafka.common.serialization.StringSerializer"/> 
       
            <bean id="kafkaEncoder" 
       
                 class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder"> 
       
                <constructor-arg value="java.lang.String"/> 
       
            </bean> 
       
            <bean id="producerProperties" 
       
                class="org.springframework.beans.factory.config.PropertiesFactoryBean"> 
       
                <property name="properties"> 
       
                    <props> 
       
                        <prop key="topic.metadata.refresh.interval.ms">3600000</prop> 
       
                        <prop key="message.send.max.retries">5</prop> 
       
                        <prop key="serializer.class">kafka.serializer.StringEncoder</prop> 
       
                        <prop key="request.required.acks">1</prop> 
       
                    </props> 
       
                </property> 
       
            </bean> 
       
            <!-- topic test config  --> 
       
            <int:channel id="kafkaTopicTest"> 
       
                <int:queue /> 
       
            </int:channel> 
       
            <int-kafka:outbound-channel-adapter 
       
                id="kafkaOutboundChannelAdapterTopicTest"kafka-producer-context-ref="producerContextTopicTest" 
       
                auto-startup="true"channel="kafkaTopicTest"order="3"> 
       
                <int:poller fixed-delay="1000"time-unit="MILLISECONDS" 
       
                    receive-timeout="1"task-executor="taskExecutor"/> 
       
            </int-kafka:outbound-channel-adapter> 
       
            <task:executor id="taskExecutor"pool-size="5" 
       
                keep-alive="120"queue-capacity="500"/> 
       
            <!-- <bean id="kafkaEncoder" 
       
                class="org.springframework.integration.kafka.serializer.avro.AvroSpecificDatumBackedKafkaEncoder"> 
       
                <constructor-arg value="com.company.AvroGeneratedSpecificRecord"/> 
       
            </bean> --> 
       
            <int-kafka:producer-context id="producerContextTopicTest" 
       
                producer-properties="producerProperties"> 
       
                <int-kafka:producer-configurations> 
       
                    <!-- 多个topic配置 --> 
       
                    <int-kafka:producer-configuration 
       
                        broker-list="192.168.1.237:9090,192.168.1.237:9091,192.168.1.237:9092" 
       
                        key-serializer="stringSerializer" 
       
                        value-class-type="java.lang.String" 
       
                        value-serializer="stringSerializer" 
       
                        topic="mytopic"/> 
       
                    <int-kafka:producer-configuration 
       
                        broker-list="192.168.1.237:9090,192.168.1.237:9091,192.168.1.237:9092" 
       
                        key-serializer="stringSerializer" 
       
                        value-class-type="java.lang.String" 
       
                        value-serializer="stringSerializer" 
       
                        topic="sunneytopic"/> 
       
                </int-kafka:producer-configurations> 
       
            </int-kafka:producer-context> 
       
        </beans>

　　3、发消息接口 KafkaService

 
        packagecom.sunney.service; 
       
        /** 
       
         * 类KafkaService.java的实现描述：发消息接口类 
       
         * @author Sunney 2016年4月30日 上午11:30:53 
       
         */ 
       
        publicinterfaceKafkaService { 
       
            /** 
       
             * 发消息 
       
             * @param topic 主题 
       
             * @param obj 发送内容 
       
             */ 
       
            publicvoidsendUserInfo(String topic, Object obj); 
       
        }

　　4、发消息实现类 KafkaServiceImpl

 
        packagecom.sunney.service.impl; 
       
        importorg.springframework.beans.factory.annotation.Autowired; 
       
        importorg.springframework.beans.factory.annotation.Qualifier; 
       
        importorg.springframework.integration.kafka.support.KafkaHeaders; 
       
        importorg.springframework.integration.support.MessageBuilder; 
       
        importorg.springframework.messaging.MessageChannel; 
       
        importorg.springframework.stereotype.Service; 
       
        importcom.sunney.service.KafkaService; 
       
        /** 
       
         * 类KafkaServiceImpl.java的实现描述：发消息实现类 
       
         * @author Sunney 2016年4月30日 上午11:31:13 
       
         */ 
       
        @Service("kafkaService") 
       
        publicclassKafkaServiceImpl implementsKafkaService{ 
       
            @Autowired 
       
            @Qualifier("kafkaTopicTest") 
       
            MessageChannel channel; 
       
            publicvoidsendUserInfo(String topic, Object obj) { 
       
                channel.send(MessageBuilder.withPayload(obj) 
       
                                            .setHeader(KafkaHeaders.TOPIC,topic) 
       
                                            .build()); 
       
            } 
       
        }

　　5、消费接收类KafkaConsumerService

 
        packagecom.sunney.service.impl; 
       
        importjava.util.Collection; 
       
        importjava.util.Iterator; 
       
        importjava.util.LinkedHashMap; 
       
        importjava.util.List; 
       
        importjava.util.Map; 
       
        importjava.util.Set; 
       
        importorg.slf4j.Logger; 
       
        importorg.slf4j.LoggerFactory; 
       
        importcom.alibaba.fastjson.JSON; 
       
        importcom.sunney.service.UserDto; 
       
        /** 
       
         * 类KafkaConsumerService.java的实现描述：消费接收类 
       
         * 
       
         * @author Sunney 2016年4月30日 上午11:46:14 
       
         */ 
       
        publicclassKafkaConsumerService { 
       
            staticfinalLogger logger = LoggerFactory.getLogger(KafkaConsumerService.class); 
       
            publicvoidprocessMessage(Map<String, Map<Integer, String>> msgs) { 
       
                logger.info("===============processMessage==============="); 
       
                for(Map.Entry<String, Map<Integer, String>> entry : msgs.entrySet()) { 
       
                    logger.info("============Topic:"+ entry.getKey()); 
       
                    LinkedHashMap<Integer, String> messages = (LinkedHashMap<Integer, String>) entry.getValue(); 
       
                    Set<Integer> keys = messages.keySet(); 
       
                    for(Integer i : keys) 
       
                        logger.info("======Partition:"+ i); 
       
                    Collection<String> values = messages.values(); 
       
                    for(Iterator<String> iterator = values.iterator(); iterator.hasNext();) { 
       
                        String message ="["+iterator.next()+"]"; 
       
                        logger.info("=====message:"+ message); 
       
                        List<UserDto> userList = JSON.parseArray(message, UserDto.class);  
       
                        logger.info("=====userList.size:"+ userList.size()); 
       
                    } 
       
                } 
       
            } 
       
        }

　　6、pom

 
        <dependencies> 
       
                <dependency> 
       
                    <groupId>org.springframework.boot</groupId> 
       
                    <artifactId>spring-boot-starter-web</artifactId> 
       
                </dependency> 
       
                <dependency> 
       
                    <groupId>org.springframework.integration</groupId> 
       
                    <artifactId>spring-integration-kafka</artifactId> 
       
                    <version>1.3.0.RELEASE</version> 
       
                </dependency> 
       
                <dependency> 
       
                    <groupId>junit</groupId> 
       
                    <artifactId>junit</artifactId> 
       
                    <version>4.11</version> 
       
                    <scope>test</scope> 
       
                </dependency> 
       
                <dependency> 
       
                    <groupId> org.apache.avro</groupId> 
       
                    <artifactId>avro</artifactId> 
       
                    <version>1.7.7</version> 
       
                </dependency> 
       
                <dependency> 
       
                    <groupId>com.alibaba</groupId> 
       
                    <artifactId>fastjson</artifactId> 
       
                    <version>1.2.7</version> 
       
                </dependency> 
       
            </dependencies>

　六、源代码地址：https://github.com/sunney2010/kafka-demo

七、遇到的问题

1、消费端口收不到消息

spring-kafka-consumer.xml的auto-startup设置为true

Kafka开源项目指南

Kafka开源项目指南提供详尽教程，助开发者掌握其架构、配置和使用，实现高效数据流管理和实时处理。它高性能、可扩展，适合日志收集和实时数据处理，通过持久化保障数据安全，是企业大数据生态系统的核心。

更多推荐

Kafka入门（一）概述、部署与API的简单使用

Kafka概述、部署与API的简单使用

Kafka开源项目指南

基于canal和kafka同步，实现binlog同步ElasticSearch

文章目录前言elasticsearch 安装canal安装canal-adapter 安装及配置mysql 安装zk及kafaka安装查看效果注意事项前言中间件版本elasticsearch7.5.2canal1.1.4client-adapter1.1.5-alpha-1zookeeper3.4.13kafka2.6.0mysql5.7.31elasticsearch 安装{"settings"

Kafka开源项目指南

基于 Iceberg 的湖仓一体架构在 B 站的实践

背景在B站，每天都有PB级的数据注入到大数据平台，经过离线或实时的ETL建模后，提供给下游的分析、推荐及预测等场景使用。面对如此大规模的数据，如何高效低成本地满足下游数据的分析需求，一直是我们重点的工作方向。我们之前的数据处理流程基本上是这样的：采集端将客户端埋点、服务端埋点、日志、业务数据库等数据收集到HDFS、Kafka等存储系统中，然后通过Hive、Spark、Fl...