package com.ljt.stormandkafka.kafkaAndStorm;

import org.apache.storm.kafka.KafkaSpout;
import org.apache.storm.kafka.SpoutConfig;
import org.apache.storm.kafka.ZkHosts;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.TopologyBuilder;

public class KafkaAndStormTopologyMain {
public static void main(String[] args) throws Exception{
TopologyBuilder topologyBuilder = new TopologyBuilder();
topologyBuilder.setSpout(“kafkaSpout”,
(IRichSpout) new KafkaSpout(new SpoutConfig(
new ZkHosts(“zk01:2181,zk02:2181,zk03:2181”),
“orderMq”,
“/myKafka”,
“kafkaSpout”)),1);
topologyBuilder.setBolt(“mybolt1”,new ParserOrderMqBolt(),1).shuffleGrouping(“kafkaSpout”);

    Config config = new Config();
    config.setNumWorkers(1);

    //3、提交任務  -----兩種模式 本地模式和叢集模式
    if (args.length>0) {
        StormSubmitter.submitTopology(args[0], config, topologyBuilder.createTopology());
    }else {
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("storm2kafka", config, topologyBuilder.createTopology());
    }
}
 

}

package com.ljt.stormandkafka.kafkaAndStorm;

import java.util.HashMap;
import java.util.Map;

import com.google.gson.Gson;
import com.ljt.stormandkafka.order.OrderInfo;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

/**
*
*

Title: ParserOrderMqBolt

*

功能描述::

*

Company: adteach

* @author 劉建濤 *
* @date 2017年8月23日下午7:44:04
* @version 1.0
*/
public class ParserOrderMqBolt extends BaseRichBolt {
private JedisPool pool;
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
//change “maxActive” -> “maxTotal” and “maxWait” -> “maxWaitMillis” in all examples
JedisPoolConfig config = new JedisPoolConfig();
//控制一個pool最多有多少個狀態為idle(空閒的)的jedis例項。
config.setMaxIdle(5);
//控制一個pool可分配多少個jedis例項，通過pool.getResource()來獲取；
//如果賦值為-1，則表示不限制；如果pool已經分配了maxActive個jedis例項，則此時pool的狀態為exhausted(耗盡)。
//在borrow一個jedis例項時，是否提前進行validate操作；如果為true，則得到的jedis例項均是可用的；
config.setMaxTotal(1000 * 100);
//表示當borrow(引入)一個jedis例項時，最大的等待時間，如果超過等待時間，則直接丟擲JedisConnectionException；
config.setMaxWaitMillis(30);
config.setTestOnBorrow(true);
config.setTestOnReturn(true);
/**
*如果你遇到 java.net.SocketTimeoutException: Read timed out exception的異常資訊
*請嘗試在構造JedisPool的時候設定自己的超時值. JedisPool預設的超時時間是2秒(單位毫秒)
*/
pool = new JedisPool(config, “127.0.0.1”, 6379);
}

@Override
public void execute(Tuple input) {
    Jedis jedis = pool.getResource();
    //獲取kafkaSpout傳送過來的資料，是一個json
    String string = new String((byte[]) input.getValue(0));
    //解析json
    OrderInfo orderInfo = (OrderInfo) new  Gson().fromJson(string, OrderInfo.class);
    //整個網站，各個業務線，各個品類，各個店鋪，各個品牌，每個商品
    //獲取整個網站的金額統計指標

// String totalAmount = jedis.get(“totalAmount”);
jedis.incrBy(“totalAmount”,orderInfo.getProductPrice());
//獲取商品所屬業務線的指標資訊
String bid = getBubyProductId(orderInfo.getProductId(),”b”);
// String bAmout = jedis.get(bid+”Amout”);
jedis.incrBy(bid+”Amount”,orderInfo.getProductPrice());
jedis.close();
}

private String getBubyProductId(String productId,String type) {

// key:value
//index:productID:info—->Map
// productId—–<各個業務線，各個品類，各個店鋪，各個品牌，每個商品>
Map