spring-integration-kafka

单位，新突击做了一个项目，用的新技术比较多一些，说是新技术，只是我之前没有接触过，kafka是其中之一。
因为项目用到了spring，所以在使用很多开源框架的时候，都选择使用spring进行集成。而kafka的使用，我们是使用到了spring-integration-kafka。

1、pom.xml

项目使用maven进行构建，使用spring-integration-kafka的话，需要引入相关jar包：

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-kafka</artifactId>
    <version>2.0.0.M1</version>
</dependency>

spring选择的是4.1.6.RELEASE：

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

2、发送消息

2.1配置文件

<?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"
    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-2.0.xsd
        http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-4.2.xsd
        http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-4.1.xsd
        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-4.1.xsd">

<context:property-placeholder location="classpath:*.properties" />
<!-- 生产消息 -->
<int:channel id="inputToKafka"></int:channel>
<int-kafka:outbound-channel-adapter kafka-template="kafkaTemplate"  channel="inputToKafka" topic="test"/>
<bean id="kafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">
    <constructor-arg index="0" ref="producerFactory"></constructor-arg>
</bean>
<bean id="producerFactory" class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
    <constructor-arg>
        <!-- key 详见org.apache.kafka.clients.producer.ProducerConfig中的常量用于调优kafka参数 -->
        <map>
            <entry key="bootstrap.servers" value="${broker.address}"/>
            <entry key="key.serializer" value="org.apache.kafka.common.serialization.StringSerializer"></entry>
            <entry key="value.serializer" value="org.apache.kafka.common.serializatbroker.addression.StringSerializer"></entry>
        </map>
    </constructor-arg>
</bean>
</beans>

topic="test" 需要修改成对应的topoc的名称

kafka.properties：

broker.address=localhost:9092
broker.groupid=mygroup

2.2代码

package com.taikang.kafka.demo;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;

public class Server {
    
    public static void main(String[] args) {
        //ApplicationContext context =new ClassPathXmlApplicationContext("spring-produce.xml");
        final ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("/spring-produce.xml", Server.class);
        context.start();
        
        MessageChannel messageChannel=(MessageChannel) context.getBean("inputToKafka");
        for(int i=0;i<15;i++){
            Message<String> message=new GenericMessage<String>("test-------------");
            boolean flag=messageChannel.send(message);
            
            //boolean flag = messageChannel.send(message, 2 * 1000);
            System.out.println(flag);
        }
        
        try {
            System.out.println("Thread.sleep===========================");
            Thread.sleep(100000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        context.close();
    }

}

3、消息接收

3.1配置文件

<?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"
    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-2.0.xsd
        http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-4.2.xsd
        http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-4.1.xsd
        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-4.1.xsd">

    <int-kafka:message-driven-channel-adapter
id="kafkaListener"
listener-container="container1"
auto-startup="false"
phase="100"
send-timeout="5000"
channel="nullChannel"
error-channel="errorChannel" />

    <bean id="container1" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
    <constructor-arg>
    <bean class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
    <constructor-arg>
    <map>
        <entry key="bootstrap.servers" value="localhost:9092" />
        <entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"></entry>
                        <entry key="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"></entry>
                    </map>
    </constructor-arg>
    </bean>
    </constructor-arg>
    <constructor-arg name="topics" value="test" />
    </bean>
</beans>

3.2代码

package com.taikang.kafka.demo;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.messaging.Message;

public class Consumer {
    public static void main(String[] args) {
        ApplicationContext context =new ClassPathXmlApplicationContext("spring-consumer.xml");
        QueueChannel queueChannel=(QueueChannel) context.getBean("inputFromKafka");
        Message msg=null;
        while((msg=queueChannel.receive(-1))!=null){
            String map=(String) msg.getPayload();
            System.out.println(map);
        }
    }

}

上传github失败....回头再想办法传吧...不过想起来之前scrapy的安装文件说之后上传也没传。