提交线上使用版本

src/main/java/cn/ac/iie/bolt/CompletionBolt.java

@@ -0,0 +1,72 @@
+package cn.ac.iie.bolt;
+
+import cn.ac.iie.common.FlowWriteConfig;
+import cn.ac.iie.utils.general.SnowflakeId;
+import cn.ac.iie.utils.hbase.HBaseUtils;
+import cn.ac.iie.utils.system.TupleUtils;
+import com.zdjizhi.utils.FormatUtils;
+import com.zdjizhi.utils.StringUtil;
+import org.apache.log4j.Logger;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.BasicOutputCollector;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseBasicBolt;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static cn.ac.iie.utils.general.TransFormUtils.dealCommonMessage;
+
+/**
+ * @author qidaijie
+ */
+
+public class CompletionBolt extends BaseBasicBolt {
+    private final static Logger logger = Logger.getLogger(CompletionBolt.class);
+    private static final long serialVersionUID = 9006119186526123734L;
+    private static final String IS = "yes";
+
+
+    @Override
+    public void prepare(Map stormConf, TopologyContext context) {
+
+    }
+
+
+    @Override
+    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
+        try {
+//            if (TupleUtils.isTick(tuple)) {
+//                if (IS.equals(FlowWriteConfig.NEED_COMPLETE_SUBID)) {
+//                    HBaseUtils.change();
+//                }
+//            } else {
+                String message = tuple.getString(0);
+                if (StringUtil.isNotBlank(message)) {
+                    basicOutputCollector.emit(new Values(dealCommonMessage(message)));
+                }
+//            }
+        } catch (Exception e) {
+            logger.error(FlowWriteConfig.KAFKA_TOPIC + "接收/解析过程出现异常");
+            e.printStackTrace();
+        }
+    }
+
+//    @Override
+//    public Map<String, Object> getComponentConfiguration() {
+//        Map<String, Object> conf = new HashMap<String, Object>(16);
+//        conf.put(org.apache.storm.Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,
+//                FlowWriteConfig.HBASE_TICK_TUPLE_FREQ_SECS);
+//        return conf;
+//    }
+
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+        outputFieldsDeclarer.declare(new Fields("tsgLog"));
+    }
+
+}