Go语言视频零基础入门到精通

admin

出错在 KStreamFlatMapValues 方法执行时，由于json异常数据无法解析，结果生成的值为null,报错信息如下：

1. [code]2018-04-18 19:21:04,776 ERROR [app-8629d547-bcf1-487b-85e5-07d7e135e1e3-StreamThread-1] com.gw.stream.KStream103.lambda$main$1(100) | 捕获到异常：hello world hello world king
2. Exception in thread "app-8629d547-bcf1-487b-85e5-07d7e135e1e3-StreamThread-1" java.lang.NullPointerException
3.         at org.apache.kafka.streams.kstream.internals.KStreamFlatMapValues$KStreamFlatMapValuesProcessor.process(KStreamFlatMapValues.java:41)
4.         at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:46)
5.         at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:208)
6.         at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:124)
7.         at org.apache.kafka.streams.processor.internals.AbstractProcessorContext.forward(AbstractProcessorContext.java:174)
8.         at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:80)
9.         at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:224)
10.         at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:94)
11.         at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:411)
12.         at org.apache.kafka.streams.processor.internals.StreamThread.processAndMaybeCommit(StreamThread.java:918)
13.         at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:798)
14.         at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:750)
15.         at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:720)

复制代码

[/code]
问题解决方案：

对json解析的bean添加未知字段忽略

1. [code]
2. import java.util.List;
3. import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
5. @JsonIgnoreProperties(ignoreUnknown = true)
6. public class Bean103 {
8.     private List<String> key1;
9.     private List<List<String>> key2;
11.     public void setKey1(List<String> key1) {
12.         this.key1 = key1;
13.     }
14.     public List<String> getKey1() {
15.         return key1;
16.     }
18.     public void setKey2(List<List<String>> key2) {
19.         this.key2 = key2;
20.     }
21.     public List<List<String>> getKey2() {
22.         return key2;
23.     }
24. }

复制代码

[/code]
由于报空指针错误，所以解决空指针问题，即判断为null时创建一个空对象.

1. [code]return list == null ? new ArrayList<String>():list;

复制代码

[/code]
完整的示例代码如下:

1. [code]package com.gw.stream;
3. import java.util.ArrayList;
4. import java.util.List;
5. import java.util.Properties;
6. import java.util.stream.Collectors;
8. import org.apache.kafka.clients.consumer.ConsumerConfig;
9. import org.apache.kafka.common.serialization.Serdes;
10. import org.apache.kafka.streams.KafkaStreams;
11. import org.apache.kafka.streams.KeyValue;
12. import org.apache.kafka.streams.StreamsBuilder;
13. import org.apache.kafka.streams.StreamsConfig;
14. import org.apache.kafka.streams.kstream.KStream;
15. import org.apache.kafka.streams.kstream.Produced;
16. import org.apache.log4j.Logger;
18. import com.alibaba.fastjson.JSONObject;
20. public class KStream103 {
22.     private static Logger log = Logger.getLogger(KStream103.class);
24.     public static void main(String[] args) {
26.         if(args.length < 6){
27.             log.error("错误:参数个数不正确[application_id bootstarp_server groupid source_topic target_topic auto_offset_reset]");
28.             return ;
29.         }
30.         String application_id=args[0];
31.         String bootstarp_server = args[1];
32.         String groupid = args[2];
33.         String source_topic = args[3];
34.         String target_topic = args[4];
35.         String auto_offset_reset = args[5];
37.         Properties props = new Properties();
38.         // consumer group
39.         // 指定一个应用ID，会在指定的目录下创建文件夹，里面存放.lock文件
40.         props.put(StreamsConfig.APPLICATION_ID_CONFIG, application_id);
41.         props.put(StreamsConfig.STATE_DIR_CONFIG, "./tmp/");
42.         props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,bootstarp_server);
43.         // props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG,10485760);
44.         props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 2000);
45.         props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
46.         props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
47.         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, auto_offset_reset);
48.         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);  //自动提交
49.         props.put(ConsumerConfig.GROUP_ID_CONFIG, groupid);
50.         //针对时间异常解决方法
51.         props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, MyEventTimeExtractor.class);
54.         final String splitChar = "\001";
56.         StreamsBuilder builder = new StreamsBuilder();
57.         KStream<String, String> textLines = builder.stream(source_topic); // 接收第一个topic
58.         textLines.flatMapValues(value -> {
60.             Bean103 bean103 = null;
61.             List<String> list = null;
63.             try {
65.                 //这里是value的业务处理逻辑...最终返回的是一个list
67.             } catch (Exception e) {
68.                 log.error("捕获到异常：" + value);
69.                 log.error("error message:" + e.getMessage());
71.             }
72.             return list == null ? new ArrayList<String>():list;
74.         }).filter((k,v)-> v !=null).map((k, v) -> new KeyValue<>(k, v))
75.         .to(target_topic, Produced.with(Serdes.String(), Serdes.String()));
77.         KafkaStreams streams = new KafkaStreams(builder.build(), props);
78.         streams.start();
80.     }
82. }

复制代码

[/code]