大数据
by smallnest

Spring 集成 Kafka

spring-integration-kafka是Spring官方提供的一个Spring集成框架的扩展，用来为使用Spring框架的应用程序提供Kafka框架的集成。
当前spring-integration-kafka仅提供Kafka 0.8的集成，低版本的Kafka并不支持。

新的文章介绍了代码实践： Kafka和Spring集成实践

spring-integration-kafka仅仅支持两个组件，分别对应Producer和 High Level Consumer。 它们分别是：

  • Outbound Channel Adapter
  • Inbound Channel Adapter based on the High level consumer API

其它的Kafka的特性比如Simple Consumer API。 所以使用spring-integration-kafka你并不能指定特定的offset来读取数据，或者进行更灵活的定制。

总的来说，spring-integration-kafka还处于很低级的阶段，和整体的Spring framework/Spring Integration Framework还不是很完美的整合。如果你现在就使用它，可能在开发的过程中会遇到些问题。

Outbound Channel Adapter:

用来向Kafka集群发送消息。消息读取于Spring Integration channel。当前的版本需要你指定Topic和MessageKey。

1
2
3
4
5
6
final MessageChannel channel = ctx.getBean("inputToKafka", MessageChannel.class);
channel.send(
        MessageBuilder.withPayload(payload)
                .setHeader("messageKey", "key")
                .setHeader("topic", "test").build());

channel是这样配置的：

1
2
3
4
5
6
<int-kafka:outbound-channel-adapter id="kafkaOutboundChannelAdapter"
                                     kafka-producer-context-ref="kafkaProducerContext"
                                     auto-startup="false"
                                     channel="inputToKafka">
     <int:poller fixed-delay="1000" time-unit="MILLISECONDS" receive-timeout="0" task-executor="taskExecutor"/>
 </int-kafka:outbound-channel-adapter>

上面一个很关键的属性是kafka-producer-context-ref， 它用来配置Producer。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
<int-kafka:producer-context id="kafkaProducerContext">
    <int-kafka:producer-configurations>
        <int-kafka:producer-configuration broker-list="localhost:9092"
                   key-class-type="java.lang.String"
                   value-class-type="java.lang.String"
                   topic="test1"
                   value-encoder="kafkaEncoder"
                   key-encoder="kafkaEncoder"
                   compression-codec="default"/>
        <int-kafka:producer-configuration broker-list="localhost:9092"
                   topic="test2"
                   compression-codec="default"
                   async="true"/>
        <int-kafka:producer-configuration broker-list="localhost:9092"
                    topic="regextopic.*"
                    compression-codec="default"/>
    </int-kafka:producer-configurations>
</int-kafka:producer-context>

可以看到， Spring将很多Kafka native codes中的配置抽象成SPring bean，通过Spring配置的方式生成Spring beans。
每个producer-configuration最终转换成一个Kafka producer。每个Topic都对应一个Producer。上面的例子会产生两个Producer，一个对应 topic test1,另外一个对应topic test2。
每个Producer都可以配置下面的属性：

1
2
3
4
5
6
7
8
9
10
11
12
broker-list             List of comma separated brokers that this producer connects to
topic                   Topic name or Java regex pattern of topic name
compression-codec       Compression method to be used. Default is no compression. Supported compression codec are gzip and snappy.
                        Anything else would result in no compression
value-encoder           Serializer to be used for encoding messages.
key-encoder             Serializer to be used for encoding the partition key
key-class-type          Type of the key class. This will be ignored if no key-encoder is provided
value-class-type        Type of the value class. This will be ignored if no value-encoder is provided.
partitioner             Custom implementation of a Kafka Partitioner interface.
async                   True/False - default is false. Setting this to true would make the Kafka producer to use
                        an async producer
batch-num-messages      Numbe

value-encoder 和 key-encoder 可以引用其它的Spring bean。partitioner 也可以是一个实现Kafka Partitioner 接口的Spring bean。
这里有一个encoder的例子：

1
2
3
<bean id="kafkaEncoder" class="org.springframework.integration.kafka.serializer.avro.AvroSpecificDatumBackedKafkaEncoder">
    <constructor-arg value="com.company.AvroGeneratedSpecificRecord" />
</bean>

如果没有配置，将采用Kafka默认的encoder。 默认的Encoder将数据视为byte数组。
如果key和消息都是字符串， Kafka提供Spring Encoder.它需要一个VerifiableProperties 作为构造函数参数。spring-integration-kafka提供可一个Properties对象封装。
所以你可以配置属性如：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
<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="send.buffer.bytes">5242880</prop>
        </props>
    </property>
</bean>
<int-kafka:producer-context id="kafkaProducerContext" producer-properties="producerProperties">
    <int-kafka:producer-configurations>
        <int-kafka:producer-configuration ... > ... </int-kafka:producer-configuration>
        <int-kafka:producer-configuration ... > ... </int-kafka:producer-configuration>
        ...
    <int-kafka:producer-configurations>
</int-kafka:producer-context>

Inbound Channel Adapter

用来消费消息。 消息会被放入到channel中。 Kafka提供两种方式的consumer API: High Level Consumer 和 Simple Consumer. 对于client来说，如果不需要特别的控制，只是用来出来目前的消息，High Level consumer API更直接，更易用。 但是它不提供offset管理。 如果你想获取前面的消息，或者重新获取已获取的消息， 它做不到。 你需要使用Simple consumer API. Spring Integration Kafka inbound channel adapter 当前只支持High Level Consumer.

1
2
3
4
5
6
<int-kafka:inbound-channel-adapter id="kafkaInboundChannelAdapter"
                                       kafka-consumer-context-ref="consumerContext"
                                       auto-startup="false"
                                       channel="inputFromKafka">
        <int:poller fixed-delay="10" time-unit="MILLISECONDS" max-messages-per-poll="5"/>
</int-kafka:inbound-channel-adapter>

必须定义kafka-consumer-context-ref用来产生consumer。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
<int-kafka:consumer-context id="consumerContext"
                                consumer-timeout="4000"
                                zookeeper-connect="zookeeperConnect">
        <int-kafka:consumer-configurations>
            <int-kafka:consumer-configuration group-id="default"
                    value-decoder="valueDecoder"
                    key-decoder="valueDecoder"
                    max-messages="5000">
                <int-kafka:topic id="test1" streams="4"/>
                <int-kafka:topic id="test2" streams="4"/>
            </int-kafka:consumer-configuration>
            <int-kafka:consumer-configuration group-id="default3"
                     value-decoder="kafkaSpecificDecoder"
                     key-decoder="kafkaReflectionDecoder"
                     max-messages="10">
                <int-kafka:topic-filter pattern="regextopic.*" streams="4" exclude="false"/>
            </int-kafka:consumer-configuration>
        </int-kafka:consumer-configurations>
</int-kafka:consumer-context>

需要配置zookeeper connection:

1
2
3
<int-kafka:zookeeper-connect id="zookeeperConnect" zk-connect="localhost:2181" zk-connection-timeout="6000"
                    zk-session-timeout="6000"
                    zk-sync-time="2000" />