Kafka Streams 开发单词计数应用

1、添加依赖（IDEA工具操作）

        <groupId>org.apache.kafka</groupId>

        <artifactId>kafka-streams</artifactId>

        <version>2.0.0</version>

    </dependency>

2、编写代码（IDEA工具操作）

根据上述业务流程分析得出,单词数据通过自定义处处理器接收并执行相应业务计算，因此创建LogProcessor类，并且继承StreamsAPI中的P ocessor接口,在Processor接口中，定义了以下3个方法。

(1)init(ProcessorContextprocessorContext):初如治化上下文对象。

(2)process(Key,Value):每接收到一条消息时，都会调用该方法处理并更新状态进行存储。

(3)close():关闭处理器,这里可以做一些资源清理理工作。 Kafka Streams单词计数详细代码如下所示。

import org.apache.kafka.streams.processor.Processor;

import org.apache.kafka.streams.processor.ProcessorContext;

import java.util.HashMap;

public class LogProcessor implements Processor<byte[],byte[]> {

private ProcessorContext processorContext;

public void init(ProcessorContext processorContext) {

    this.processorContext = processorContext;

}

public void process(byte[] key, byte[] value) {

    String inputOri = new String(value);

    HashMap <String,Integer>map = new HashMap<String, Integer>();

    int times = 1;

    if (inputOri.contains("")){

        //截取字段

        String[] words = inputOri.split(" ");

        for (String word : words){

            if (map.containsKey(word)){

                map.put(word,map.get(word)+1);

            }else {

                map.put(word,times);

            }

        }

    }

    inputOri = map.toString();

    processorContext.forward(key,inputOri.getBytes());

}

public void close() {

}

}

单词计数的业务功能开发完之后，Kafka Streams 需要编写一个运行主程序的类App，来测试LogProcessor业务程序，具体代码如下：

import org.apache.kafka.streams.KafkaStreams;

import org.apache.kafka.streams.StreamsConfig;

import org.apache.kafka.streams.Topology;

import org.apache.kafka.streams.processor.Processor;

import org.apache.kafka.streams.processor.ProcessorSupplier;

import java.util.Properties;

public class App {

public static void main(String[] args) {

    //声明来源主题

    String fromTopic = "test1";

    //声明目标主题

    String toTopic = "test2";

    //设置参数信息

    Properties props = new Properties();

    props.put(StreamsConfig.APPLICATION_ID_CONFIG,"logProcessor");

    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,

            "master:9092,slave1:9092,slave2:9092");

    //实例化StreamsConfig对象

    StreamsConfig config = new StreamsConfig(props);

    //构建拓扑结构

    Topology topology = new Topology();

    //添加原处理节点，为原处理节点指定名称和他订阅的主题

    topology.addSource("SOURCE",fromTopic)

            //添加自定义处理节点，指定处理器类和上一节点的名称

    .addProcessor("PROCESSOR", new ProcessorSupplier() {

        public Processor get() {

            return new LogProcessor();

        }

    },"SOURCE")

            //添加目标处理节点，需要指定目标处理节点和上一节点的名称

            .addSink("SINK",toTopic,"PROCESSOR");

    //实例化KafkaStreams对象

    KafkaStreams streams = new KafkaStreams(topology,config);

    streams.start();

}

}

3、执行测试

代码编写完后，在master节点创建test1和test2主题，命令如下：

创建来源主题

*kafka-topics.sh --create *

**--topic test1 **

**--partitions 3 **

**--replication-factor 2 **

--zookeeper master:2181,slave1:2181,slave2:2181

创建目标主题

*kafka-topics.sh --create *

**--topic test2 **

**--partitions 3 **

**--replication-factor 2 **

--zookeeper master:2181,slave1:2181,slave2:2181

成功创建好目标主题后，分别在master节点和slave1节点启动生产者服务和消费者服务。

启动生产者服务命令如下：

*kafka-console-procuder.sh *

**--broker-list master:9092,slave1:9092,slave2:9092 **

--topic test1

启动消费者服务命令如下：

*kafka-console-consumer.sh *

**--from-beginning **

**--topic test2 **

--bootstrap-server master:9092,slave1:9092,slave2:9092

在生产者服务节点master输入数据：

到IDEA中运行创建好的App文件中的代码，运行结果如下：

以上两张图可以看到Kafka Streams所需的测试环境已经配置完成了。