MetaQ对接SparkStreaming示例代码

由于JavaReceiverInputDStream<String> lines = ssc.receiverStream(Receiver<T> receiver) 中 没有直接对接MetaQ的工具，当然可以实用使用spark streaming已经有的工具进行转接，这里不建议，所以可以继承Receiver类重写onStart()方法

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;

}

下面该段代码可以减掉，若有需要转Object可以在此进行处理

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;

    }*/

}

接下来通过spark streaming进行metaq的消息处理

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();

        }

}