MetaQ对接SparkStreaming示例代码详解编程语言

import java.text.DateFormat; 
import java.text.SimpleDateFormat; 
import java.util.Date; 
import java.util.concurrent.Executor; 
 
import org.apache.spark.storage.StorageLevel; 
import org.apache.spark.streaming.receiver.Receiver; 
import org.slf4j.Logger; 
import org.slf4j.LoggerFactory; 
 
import com.dinpay.bdp.rcp.domain.Order; 
import com.taobao.metamorphosis.Message; 
import com.taobao.metamorphosis.client.MessageSessionFactory; 
import com.taobao.metamorphosis.client.MetaClientConfig; 
import com.taobao.metamorphosis.client.MetaMessageSessionFactory; 
import com.taobao.metamorphosis.client.consumer.ConsumerConfig; 
import com.taobao.metamorphosis.client.consumer.MessageConsumer; 
import com.taobao.metamorphosis.client.consumer.MessageListener; 
import com.taobao.metamorphosis.utils.ZkUtils.ZKConfig; 
 
public abstract class MetaQReceiver<T> extends Receiver<T>{ 
    private static final long serialVersionUID = -3240967436204273248L; 
    Logger logger=LoggerFactory.getLogger(MetaQReceiver.class); 
    private static final DateFormat df = new SimpleDateFormat("yyyyMMdd"); 
     
    private String zkConnect; 
    private String zkRoot; 
    private String topic; 
    private String group; 
 
    public MetaQReceiver(String zkConnect,String zkRoot,String topic,String group) { 
        super(StorageLevel.MEMORY_ONLY()); 
        this.zkConnect=zkConnect; 
        this.zkRoot=zkRoot; 
        this.topic=topic; 
        this.group=group; 
         
    } 
     
    @Override 
    public void onStart() { 
        try{ 
            final MetaClientConfig metaClientConfig = new MetaClientConfig(); 
            final ZKConfig zkConfig = new ZKConfig(); 
            zkConfig.zkConnect = this.zkConnect;// "127.0.0.1:2181"; 
            zkConfig.zkRoot = this.zkRoot;// "/meta"; 
            metaClientConfig.setZkConfig(zkConfig); 
            final MessageSessionFactory sessionFactory = new MetaMessageSessionFactory( 
                    metaClientConfig); 
            ConsumerConfig consumerConfig = new ConsumerConfig(group); 
            // 默认最大获取延迟为5秒，这里设置成100毫秒，请根据实际应用要求做设置。 
            consumerConfig.setMaxDelayFetchTimeInMills(100); 
            final MessageConsumer consumer = sessionFactory 
                    .createConsumer(consumerConfig); 
            // subscribe topic 
            consumer.subscribe(topic, 1024 * 1024, new MessageListener() { 
                @Override 
                public void recieveMessages(final Message message) { 
                    try{ 
                        //T t=message2Object(new String(message.getData(),"utf-8")); 
                        logger.info("Receive message " + new String(message.getData())); 
                        String orderJson = new String(message.getData()); 
                        Order order = ParameterDataUtil.getObject(orderJson, Order.class); 
                        String cardNo = order.getCard_no(); 
                        String yyyyMMdd = df.format(new Date()); 
                        String payclassId = order.getPayclass_id(); 
                        String cntKey = "DK_CNT_" + cardNo + "_" + payclassId + "_" + yyyyMMdd; 
                        logger.info(cntKey); 
                        System.out.println(cntKey); 
                        T result = (T) cntKey; 
                        if(result!=null){ 
                            store(result); 
                        } 
                    }catch(Exception e){ 
                        logger.error("message2Object error",e); 
                    } 
                } 
                @Override 
                public Executor getExecutor() { 
                    return null; 
                } 
            }); 
            consumer.completeSubscribe(); 
        }catch(Exception e){ 
            throw new RuntimeException("metaq error",e); 
        } 
    } 
     
    @Override 
    public void onStop() { 
    } 
     
    //public abstract T message2Object(String message) throws Exception; 
}

public class MetaQReceiverStreaming extends MetaQReceiver<String>{ 
     
    private static final long serialVersionUID = -2290689243756756929L; 
     
    public MetaQReceiverStreaming(String zkConnect, String zkRoot, String topic, String group) { 
        super(zkConnect, zkRoot, topic, group); 
    } 
 
    /*@Override 
    public String message2Object(String message) throws Exception { 
        return message; 
    }*/ 
     
}

ort java.util.Arrays; 
import java.util.List; 
 
import org.apache.log4j.Level; 
import org.apache.log4j.Logger; 
import org.apache.spark.SparkConf; 
import org.apache.spark.api.java.function.FlatMapFunction; 
import org.apache.spark.api.java.function.Function2; 
import org.apache.spark.api.java.function.PairFunction; 
import org.apache.spark.streaming.Durations; 
import org.apache.spark.streaming.api.java.JavaDStream; 
import org.apache.spark.streaming.api.java.JavaPairDStream; 
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream; 
import org.apache.spark.streaming.api.java.JavaStreamingContext; 
 
import com.dinpay.bdp.rcp.util.Constant; 
import com.dinpay.bdp.rcp.util.MetaQReceiverStreaming; 
import com.google.common.base.Optional; 
import com.sun.xml.bind.v2.runtime.reflect.opt.Const; 
 
import scala.Tuple2;   
/** 
 * @author ll 
 */ 
public class MetaqStreamingCount { 
     
     public static void main(String[] args) { 
         String zkConnect=Constant.METAZK; 
         String zkRoot="/meta"; 
         String topic=Constant.METATOPIC; 
         String group=Constant.METAGROUP;  
          
         //屏蔽日志 
         Logger.getLogger("org.apache.spark").setLevel(Level.OFF); 
         SparkConf sparkConf = new SparkConf().setAppName("MetaqStreamingCount").setMaster("local[2]");   
         JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(1));  
          
         JavaReceiverInputDStream<String> lines = ssc.receiverStream(new MetaQReceiverStreaming(zkConnect,zkRoot,topic,group)); 
          
         JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() { 
              
            @Override 
            public Iterable<String> call(String line) throws Exception { 
                return Arrays.asList(line.split(" ")); 
            } 
        }); 
          
        JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() { 
            @Override 
            public Tuple2<String, Integer> call(String word) { 
                return new Tuple2<>(word, 1); 
            } 
        }); 
         
        JavaPairDStream<String, Integer> wordsCount = pairs.updateStateByKey(new Function2<List<Integer>, Optional<Integer>,  
                Optional<Integer>>() {//对相同的Key，进行Value的累计（包括Local和Reducer级别同时Reduce） 
             
            @Override 
            public Optional<Integer> call(List<Integer> values, Optional<Integer> state){ 
                //第一个参数就是key传进来的数据，第二个参数是已经有的数据 
                Integer updateValue = 0;//如果第一次，state没有，updateValue为0，如果有就获取 
                if(state.isPresent()){ 
                    updateValue = state.get(); 
                } 
                //遍历batch传进来的数据可以一直加，随着时间的流式会不断的累加相同key的value结果 
                for (Integer value : values) { 
                    updateValue += value; 
                } 
                return Optional.of(updateValue);//返回更新的值 
            } 
        }); 
 
        wordsCount.print(); 
        //需要将结果保存到Codis中 
        ssc.checkpoint("checkpoint"); 
        ssc.start(); 
        ssc.awaitTermination(); 
        ssc.close(); 
         
        } 
}