commit 027908d6cbb6a87dd8e94a73df5df2c2ab0405f8 Author: qidaijie Date: Fri Nov 8 11:46:29 2019 +0800 radius 写入hbase初始版本 diff --git a/pom.xml b/pom.xml new file mode 100644 index 0000000..35ee297 --- /dev/null +++ b/pom.xml @@ -0,0 +1,287 @@ + + 4.0.0 + + log-address-hbase + log-address-hbase + 0.0.1-SNAPSHOT + jar + + log-address-hbase + http://maven.apache.org + + + + nexus + Team Nexus Repository + http://192.168.10.125:8099/content/groups/public + + + + + + + + org.apache.maven.plugins + maven-shade-plugin + 2.4.2 + + + package + + shade + + + + + cn.ac.iie.topology.LogAddressRedisTopology + + + META-INF/spring.handlers + + + META-INF/spring.schemas + + + + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + 2.3.2 + + 1.8 + 1.8 + + + + + + properties + + **/*.properties + **/*.xml + + false + + + src/main/java + + log4j.properties + + false + + + + + + UTF-8 + 1.0.0 + 1.0.2 + 1.4.9 + 2.7.1 + + + + + + org.apache.kafka + kafka_2.11 + 1.0.0 + + + slf4j-log4j12 + org.slf4j + + + + + + org.apache.storm + storm-core + ${storm.version} + provided + + + + org.apache.storm + storm-kafka + ${storm.version} + + + slf4j-log4j12 + org.slf4j + + + log4j-over-slf4j + org.slf4j + + + + + + com.nis + nis-core + 1.0 + + + + info.monitorenter + cpdetector + 1.0.7 + + + + com.google.guava + guava + 18.0 + + + + redis.clients + jedis + 2.8.1 + + + + junit + junit + 3.8.1 + test + + + + + com.alibaba + fastjson + 1.2.47 + + + + com.zdjizhi + galaxy + 1.0.1 + + + slf4j-log4j12 + org.slf4j + + + log4j-over-slf4j + org.slf4j + + + + + + + org.apache.zookeeper + zookeeper + 3.4.9 + + + slf4j-log4j12 + org.slf4j + + + log4j-over-slf4j + org.slf4j + + + + + + + org.apache.hbase + hbase-client + ${hbase.version} + + + slf4j-log4j12 + org.slf4j + + + log4j-over-slf4j + org.slf4j + + + + + + + org.apache.hbase + hbase-server + ${hbase.version} + + + slf4j-log4j12 + org.slf4j + + + log4j-over-slf4j + org.slf4j + + + + + + + org.apache.hadoop + hadoop-common + ${hadoop.version} + + + slf4j-log4j12 + org.slf4j + + + log4j-over-slf4j + org.slf4j + + + + + + + org.apache.hadoop + hadoop-client + ${hadoop.version} + + + slf4j-log4j12 + org.slf4j + + + log4j-over-slf4j + org.slf4j + + + + + + + org.apache.hadoop + hadoop-hdfs + ${hadoop.version} + + + slf4j-log4j12 + org.slf4j + + + log4j-over-slf4j + org.slf4j + + + + + + + diff --git a/properties/address_routine.properties b/properties/address_routine.properties new file mode 100644 index 0000000..7692862 --- /dev/null +++ b/properties/address_routine.properties @@ -0,0 +1,41 @@ +#管理kafka地址 +bootstrap.servers=192.168.40.119:9092,192.168.40.122:9092,192.168.40.123:9092 +#从kafka哪里开始读：earliest/latest +auto.offset.reset=latest + +#hbase zookeeper地址 +hbase.zookeeper.servers=192.168.40.203:2186 + +#hbase tablename +hbase.table.name=subcriber_info + +#tick时钟频率 +topology.tick.tuple.freq.secs=50 + +topology.config.max.spout.pending=500000 + +topology.num.acks=0 + +#redis过期时间 +expiration.time=604800 + +#用于过滤对准用户名 +check.ip.scope=10,100,192 + +#kafka broker下的topic名称 +kafka.topic=RADIUS-LOG + +#kafka消费group id +group.id=account-to-hbase + +#storm topology workers +topology.workers=1 + +#storm spout parallelism +spout.parallelism=10 + +#storm bolt parallelism +format.bolt.parallelism=10 + +#1=单机逐条写入 2=集群逐条写入 3=集群批量写入 +redis.model=1 diff --git a/properties/core-site.xml b/properties/core-site.xml new file mode 100644 index 0000000..66742f0 --- /dev/null +++ b/properties/core-site.xml @@ -0,0 +1,64 @@ + + + + + + + + + fs.defaultFS + hdfs://ns1 + + + hadoop.tmp.dir + file:/opt/hadoop/tmp + + + io.file.buffer.size + 131702 + + + hadoop.proxyuser.root.hosts + * + + + hadoop.proxyuser.root.groups + * + + + hadoop.logfile.size + 10000000 + The max size of each log file + + + + hadoop.logfile.count + 1 + The max number of log files + + + ha.zookeeper.quorum + 192.168.40.202:2181,192.168.40.203:2181,192.168.40.206:2181 + + +io.compression.codecs +com.hadoop.compression.lzo.LzoCodec,com.hadoop.compression.lzo.LzopCodec + + +io.compression.codec.lzo.class +com.hadoop.compression.lzo.LzoCodec + + + diff --git a/properties/hbase-site.xml b/properties/hbase-site.xml new file mode 100644 index 0000000..b81715c --- /dev/null +++ b/properties/hbase-site.xml @@ -0,0 +1,81 @@ + + + + + + hbase.rootdir + hdfs://ns1/hbase/hbase-1.4.9 + + + hbase.cluster.distributed + true + + + hbase.zookeeper.quorum + 192.168.40.202,192.168.40.203,192.168.40.206 + + +hbase.master.info.port +60010 + + + + phoenix.schema.isNamespaceMappingEnabled + true + + + phoenix.schema.mapSystemTablesToNamespace + true + + + hbase.client.keyvalue.maxsize + 99428800 + + + hbase.server.keyvalue.maxsize + 99428800 + + + hbase.regionserver.wal.codec + org.apache.hadoop.hbase.regionserver.wal.IndexedWALEditCodec + + + phoenix.query.timeoutMs + 1800000 + + + hbase.regionserver.lease.period + 1200000 + + + hbase.rpc.timeout + 1200000 + + + hbase.client.scanner.caching + 1000 + + + hbase.client.scanner.timeout.period + 1200000 + + diff --git a/properties/hdfs-site.xml b/properties/hdfs-site.xml new file mode 100644 index 0000000..5bf742b --- /dev/null +++ b/properties/hdfs-site.xml @@ -0,0 +1,116 @@ + + + + + + + + + dfs.namenode.name.dir + file:/home/bigdata/hadoop/dfs/name + + + dfs.datanode.data.dir + file:/home/bigdata/hadoop/dfs/data + + + dfs.replication + 2 + + + dfs.namenode.secondary.http-address + 192.168.40.202:9001 + + + dfs.webhdfs.enabled + true + + + dfs.permissions + false + + + dfs.permissions.enabled + false + + + dfs.nameservices + ns1 + + + dfs.blocksize + 134217728 + + + dfs.ha.namenodes.ns1 + nn1,nn2 + + + + dfs.namenode.rpc-address.ns1.nn1 + 192.168.40.202:9000 + + + + dfs.namenode.http-address.ns1.nn1 + 192.168.40.202:50070 + + + + dfs.namenode.rpc-address.ns1.nn2 + 192.168.40.203:9000 + + + + dfs.namenode.http-address.ns1.nn2 + 192.168.40.203:50070 + + + + dfs.namenode.shared.edits.dir + qjournal://192.168.40.203:8485;192.168.40.206:8485;192.168.40.202:8485/ns1 + + + + dfs.journalnode.edits.dir + /home/bigdata/hadoop/journal + + + + dfs.client.failover.proxy.provider.ns1 + org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider + + + + dfs.ha.fencing.methods + sshfence + + + + dfs.ha.fencing.ssh.private-key-files + /root/.ssh/id_rsa + + + + dfs.ha.fencing.ssh.connect-timeout + 30000 + + + + dfs.ha.automatic-failover.enabled + true + + + diff --git a/src/main/java/cn/ac/iie/bolt/SubcribeIdBolt.java b/src/main/java/cn/ac/iie/bolt/SubcribeIdBolt.java new file mode 100644 index 0000000..d71542a --- /dev/null +++ b/src/main/java/cn/ac/iie/bolt/SubcribeIdBolt.java @@ -0,0 +1,159 @@ +package cn.ac.iie.bolt; + +import cn.ac.iie.common.AddressConfig; +import cn.ac.iie.utils.TupleUtils; +import com.alibaba.fastjson.JSONObject; +import com.zdjizhi.utils.StringUtil; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellUtil; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.*; +import org.apache.hadoop.hbase.util.Bytes; +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.io.IOException; +import java.util.*; + +/** + * @author qidaijie + */ +public class SubcribeIdBolt extends BaseBasicBolt { + private static Logger logger = Logger.getLogger(SubcribeIdBolt.class); + private static Map subIdMap; + private List putList; + private static Connection connection; + + static { + // 管理Hbase的配置信息 + Configuration configuration = HBaseConfiguration.create(); + // 设置zookeeper节点 + configuration.set("hbase.zookeeper.quorum", AddressConfig.HBASE_ZOOKEEPER_SERVERS); + try { + connection = ConnectionFactory.createConnection(configuration); + } catch (IOException e) { + e.printStackTrace(); + } + } + + @Override + public void prepare(Map stormConf, TopologyContext context) { + subIdMap = new HashMap<>(16); + putList = new ArrayList<>(); + getAll(); + } + + @Override + public void execute(Tuple tuple, BasicOutputCollector collector) { + try { + if (TupleUtils.isTick(tuple)) { + insertData(putList); + } else { + String message = tuple.getString(0); + if (StringUtil.isNotBlank(message)) { + JSONObject jsonObject = JSONObject.parseObject(message); + String ip = jsonObject.getString("framed_ip"); + String account = jsonObject.getString("account"); + dataValidation(ip, account, putList); + if (putList.size() == AddressConfig.LIST_SIZE_MAX) { + insertData(putList); + } + } + } + } catch (Exception e) { + logger.error("Radius写入Redis出现异常", e); + } + + } + + @Override + public Map getComponentConfiguration() { + Map conf = new HashMap(16); + conf.put(org.apache.storm.Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, + AddressConfig.TOPOLOGY_TICK_TUPLE_FREQ_SECS); + return conf; + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + + } + + /** + * 获取所有的 key value + */ + private static void getAll() { + try { + Table table = connection.getTable(TableName.valueOf("sub:" + AddressConfig.HBASE_TABLE_NAME)); + Scan scan2 = new Scan(); + ResultScanner scanner = table.getScanner(scan2); + for (Result result : scanner) { + Cell[] cells = result.rawCells(); + for (Cell cell : cells) { + subIdMap.put(Bytes.toString(CellUtil.cloneRow(cell)), Bytes.toString(CellUtil.cloneValue(cell))); + } + } + scanner.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + + /** + * 写入数据到hbase + */ + private static void insertData(List putList) { + Table table = null; + try { + table = connection.getTable(TableName.valueOf("sub:" + AddressConfig.HBASE_TABLE_NAME)); + table.put(putList); + putList.clear(); + } catch (IOException e) { + e.printStackTrace(); + } finally { + try { + if (table != null) { + table.close(); + } + } catch (IOException e) { + e.printStackTrace(); + } + } + + } + + /** + * 验证数据并与内存中的对比 + * + * @param ip framed_ip + * @param account account + */ + private static void dataValidation(String ip, String account, List putList) { + if (StringUtil.isNotBlank(ip) && StringUtil.isNotBlank(account)) { +// String s = ip.split("\\.")[0]; +// if (!AddressConfig.CHECK_IP_SCOPE.contains(s)) { + if (subIdMap.containsKey(ip)) { + if (!subIdMap.get(ip).equals(account)) { + Put put = new Put(ip.getBytes()); + put.addColumn("subscribe_id".getBytes(), "account".getBytes(), account.getBytes()); + putList.add(put); + subIdMap.put(ip, account); + } + } else { + Put put = new Put(ip.getBytes()); + put.addColumn("subscribe_id".getBytes(), "account".getBytes(), account.getBytes()); + putList.add(put); + subIdMap.put(ip, account); + } +// } + } + } +} diff --git a/src/main/java/cn/ac/iie/bolt/ToHBaseBolt.java b/src/main/java/cn/ac/iie/bolt/ToHBaseBolt.java new file mode 100644 index 0000000..fa584b0 --- /dev/null +++ b/src/main/java/cn/ac/iie/bolt/ToHBaseBolt.java @@ -0,0 +1,109 @@ +package cn.ac.iie.bolt; + +import cn.ac.iie.common.AddressConfig; +import cn.ac.iie.utils.TupleUtils; +import com.zdjizhi.utils.StringUtil; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.*; +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.Tuple; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * @author qidaijie + */ +public class ToHBaseBolt extends BaseBasicBolt { + private static Logger logger = Logger.getLogger(ToHBaseBolt.class); + private static List putList; + private static Connection connection; + + static { + // 管理Hbase的配置信息 + Configuration configuration = HBaseConfiguration.create(); + // 设置zookeeper节点 + configuration.set("hbase.zookeeper.quorum", AddressConfig.HBASE_ZOOKEEPER_SERVERS); + try { + connection = ConnectionFactory.createConnection(configuration); + } catch (IOException e) { + e.printStackTrace(); + } + } + + @Override + public void prepare(Map stormConf, TopologyContext context) { + putList = new ArrayList<>(); + } + + @Override + public void execute(Tuple tuple, BasicOutputCollector collector) { + try { + if (TupleUtils.isTick(tuple)) { + insertData(); + putList.clear(); + } else { + String message = tuple.getString(0); + if (StringUtil.isNotBlank(message)) { + String[] split = message.split("-"); + Put put = new Put(split[0].getBytes()); + put.addColumn("subscribe_id".getBytes(), "account".getBytes(), split[1].getBytes()); + putList.add(put); + if (putList.size() == AddressConfig.LIST_SIZE_MAX) { + insertData(); + putList.clear(); + } + } + } + } catch (Exception e) { + logger.error("Radius写入Redis出现异常", e); + } + + } + + @Override + public Map getComponentConfiguration() { + Map conf = new HashMap(16); + conf.put(org.apache.storm.Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, + AddressConfig.TOPOLOGY_TICK_TUPLE_FREQ_SECS); + return conf; + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + + } + + + /** + * 写入数据到hbase + */ + private static void insertData() { + Table table = null; + try { + table = connection.getTable(TableName.valueOf("sub:" + AddressConfig.HBASE_TABLE_NAME)); + table.put(putList); + } catch (IOException e) { + e.printStackTrace(); + } finally { + try { + if (table != null) { + table.close(); + } + } catch (IOException e) { + e.printStackTrace(); + } + } + + } + +} diff --git a/src/main/java/cn/ac/iie/common/AddressConfig.java b/src/main/java/cn/ac/iie/common/AddressConfig.java new file mode 100644 index 0000000..2f851c9 --- /dev/null +++ b/src/main/java/cn/ac/iie/common/AddressConfig.java @@ -0,0 +1,38 @@ +package cn.ac.iie.common; + +import cn.ac.iie.utils.AddressConfigurations; + +import java.io.Serializable; + +/** + * @author qidaijie + */ +public class AddressConfig implements Serializable { + private static final long serialVersionUID = -8326385159484059324L; + + public static final String SEGMENTATION = ","; + public static final int LIST_SIZE_MAX = 5000; + + /*** + * kafka and system + */ + public static final String BOOTSTRAP_SERVERS = AddressConfigurations.getStringProperty(0, "bootstrap.servers"); + public static final Integer SPOUT_PARALLELISM = AddressConfigurations.getIntProperty(0, "spout.parallelism"); + public static final Integer FORMAT_BOLT_PARALLELISM = AddressConfigurations.getIntProperty(0, "format.bolt.parallelism"); + public static final String GROUP_ID = AddressConfigurations.getStringProperty(0, "group.id"); + public static final String KAFKA_TOPIC = AddressConfigurations.getStringProperty(0, "kafka.topic"); + public static final String AUTO_OFFSET_RESET = AddressConfigurations.getStringProperty(0, "auto.offset.reset"); + public static final Integer TOPOLOGY_TICK_TUPLE_FREQ_SECS = AddressConfigurations.getIntProperty(0, "topology.tick.tuple.freq.secs"); + public static final Integer TOPOLOGY_CONFIG_MAX_SPOUT_PENDING = AddressConfigurations.getIntProperty(0, "topology.config.max.spout.pending"); + public static final Integer TOPOLOGY_NUM_ACKS = AddressConfigurations.getIntProperty(0, "topology.num.acks"); + + public static final Integer TOPOLOGY_WORKERS = AddressConfigurations.getIntProperty(0, "topology.workers"); + + public static final Integer EXPIRATION_TIME = AddressConfigurations.getIntProperty(0, "expiration.time"); + public static final Integer REDIS_MODEL = AddressConfigurations.getIntProperty(0, "redis.model"); + + public static final String CHECK_IP_SCOPE = AddressConfigurations.getStringProperty(0, "check.ip.scope"); + + public static final String HBASE_ZOOKEEPER_SERVERS = AddressConfigurations.getStringProperty(0, "hbase.zookeeper.servers"); + public static final String HBASE_TABLE_NAME = AddressConfigurations.getStringProperty(0, "hbase.table.name"); +} \ No newline at end of file diff --git a/src/main/java/cn/ac/iie/spout/CustomizedKafkaSpout.java b/src/main/java/cn/ac/iie/spout/CustomizedKafkaSpout.java new file mode 100644 index 0000000..96c8330 --- /dev/null +++ b/src/main/java/cn/ac/iie/spout/CustomizedKafkaSpout.java @@ -0,0 +1,78 @@ +package cn.ac.iie.spout; + +import cn.ac.iie.common.AddressConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.log4j.Logger; +import org.apache.storm.spout.SpoutOutputCollector; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.topology.OutputFieldsDeclarer; +import org.apache.storm.topology.base.BaseRichSpout; +import org.apache.storm.tuple.Fields; +import org.apache.storm.tuple.Values; + +import java.util.Arrays; +import java.util.Map; +import java.util.Properties; + +/** + * @author qidaijie + */ +public class CustomizedKafkaSpout extends BaseRichSpout { + private static final long serialVersionUID = 2934528972182398950L; + private KafkaConsumer consumer; + private SpoutOutputCollector collector = null; + private TopologyContext context = null; + private final static Logger logger = Logger.getLogger(CustomizedKafkaSpout.class); + + + private static Properties createConsumerConfig() { + Properties props = new Properties(); + props.put("bootstrap.servers", AddressConfig.BOOTSTRAP_SERVERS); + props.put("group.id", AddressConfig.GROUP_ID); + props.put("auto.offset.reset", AddressConfig.AUTO_OFFSET_RESET); + props.put("session.timeout.ms", "60000"); + props.put("max.poll.records", 3000); + props.put("max.partition.fetch.bytes", 31457280); + props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + return props; + } + + @Override + public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { + // TODO Auto-generated method stub + this.collector = collector; + this.context = context; + Properties prop = createConsumerConfig(); + this.consumer = new KafkaConsumer<>(prop); + this.consumer.subscribe(Arrays.asList(AddressConfig.KAFKA_TOPIC)); + } + + @Override + public void close() { + consumer.close(); + } + + @Override + public void nextTuple() { + try { + // TODO Auto-generated method stub + ConsumerRecords records = consumer.poll(10000L); + Thread.sleep(300); + for (ConsumerRecord record : records) { + this.collector.emit(new Values(record.value())); + } + } catch (Exception e) { + logger.error("kfaka-spout 发送数据出现异常" + e); + e.printStackTrace(); + } + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + // TODO Auto-generated method stub + declarer.declare(new Fields("source")); + } +} diff --git a/src/main/java/cn/ac/iie/topology/LogAddressRedisTopology.java b/src/main/java/cn/ac/iie/topology/LogAddressRedisTopology.java new file mode 100644 index 0000000..491cc97 --- /dev/null +++ b/src/main/java/cn/ac/iie/topology/LogAddressRedisTopology.java @@ -0,0 +1,83 @@ +package cn.ac.iie.topology; + +import cn.ac.iie.bolt.SubcribeIdBolt; +import cn.ac.iie.bolt.ToHBaseBolt; +import cn.ac.iie.common.AddressConfig; +import cn.ac.iie.spout.CustomizedKafkaSpout; +import org.apache.storm.Config; +import org.apache.storm.generated.AlreadyAliveException; +import org.apache.storm.generated.AuthorizationException; +import org.apache.storm.generated.InvalidTopologyException; +import org.apache.storm.topology.TopologyBuilder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author qidaijie + */ +public class LogAddressRedisTopology { + private static final Logger logger = LoggerFactory.getLogger(LogAddressRedisTopology.class); + private final String topologyName; + private final Config topologyConfig; + private TopologyBuilder builder; + + public LogAddressRedisTopology() { + this(LogAddressRedisTopology.class.getSimpleName()); + } + + public LogAddressRedisTopology(String topologyName) { + this.topologyName = topologyName; + topologyConfig = createTopologConfig(); + } + + private Config createTopologConfig() { + Config conf = new Config(); + conf.setDebug(false); + conf.setMessageTimeoutSecs(120); + conf.setTopologyWorkerMaxHeapSize(500); + conf.setMaxSpoutPending(AddressConfig.TOPOLOGY_CONFIG_MAX_SPOUT_PENDING); + if (AddressConfig.TOPOLOGY_NUM_ACKS == 0) { + conf.setNumAckers(0); + } + return conf; + } + + public void runLocally() throws InterruptedException { + topologyConfig.setMaxTaskParallelism(1); + StormRunner.runTopologyLocally(builder, topologyName, topologyConfig, 600); + } + + public void runRemotely() throws AlreadyAliveException, InvalidTopologyException, AuthorizationException { + topologyConfig.setNumWorkers(AddressConfig.TOPOLOGY_WORKERS); + topologyConfig.put(Config.TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE, 8);//设置过高会导致很多问题，如心跳线程饿死、吞吐量大幅下跌 + StormRunner.runTopologyRemotely(builder, topologyName, topologyConfig); + } + + private void buildTopology() { + builder = new TopologyBuilder(); + builder.setSpout("CustomizedKafkaSpout", new CustomizedKafkaSpout(), AddressConfig.SPOUT_PARALLELISM); + builder.setBolt("SubcribeIdBolt", new SubcribeIdBolt(), AddressConfig.FORMAT_BOLT_PARALLELISM).localOrShuffleGrouping("CustomizedKafkaSpout"); +// builder.setBolt("ToHBaseBolt", new ToHBaseBolt(), 1).localOrShuffleGrouping("SubcribeIdBolt"); + + } + + public static void main(String[] args) throws Exception { + LogAddressRedisTopology csst = null; + boolean runLocally = true; + if (args.length >= 2 && "remote".equalsIgnoreCase(args[1])) { + runLocally = false; + csst = new LogAddressRedisTopology(args[0]); + } else { + csst = new LogAddressRedisTopology(); + } + csst.buildTopology(); + + if (runLocally) { + logger.info("执行本地模式..."); + csst.runLocally(); + } else { + logger.info("执行远程部署模式..."); + csst.runRemotely(); + } + } +} diff --git a/src/main/java/cn/ac/iie/topology/StormRunner.java b/src/main/java/cn/ac/iie/topology/StormRunner.java new file mode 100644 index 0000000..d2d4ab9 --- /dev/null +++ b/src/main/java/cn/ac/iie/topology/StormRunner.java @@ -0,0 +1,32 @@ +package cn.ac.iie.topology; + + +import org.apache.storm.Config; +import org.apache.storm.LocalCluster; +import org.apache.storm.StormSubmitter; +import org.apache.storm.generated.AlreadyAliveException; +import org.apache.storm.generated.AuthorizationException; +import org.apache.storm.generated.InvalidTopologyException; +import org.apache.storm.topology.TopologyBuilder; + +public final class StormRunner{ + private static final int MILLS_IN_SEC = 1000; + + private StormRunner() {} + + public static void runTopologyLocally(TopologyBuilder builder, String topologyName, Config conf, int runtimeInSeconds) throws InterruptedException { + + LocalCluster localCluster = new LocalCluster(); + localCluster.submitTopology(topologyName, conf, builder.createTopology()); + Thread.sleep((long) runtimeInSeconds * MILLS_IN_SEC); + localCluster.shutdown(); + + } + + public static void runTopologyRemotely(TopologyBuilder builder, String topologyName, Config conf ) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException { + + StormSubmitter.submitTopologyWithProgressBar(topologyName, conf, builder.createTopology()); + } + + +} diff --git a/src/main/java/cn/ac/iie/utils/AddressConfigurations.java b/src/main/java/cn/ac/iie/utils/AddressConfigurations.java new file mode 100644 index 0000000..ebdfcc7 --- /dev/null +++ b/src/main/java/cn/ac/iie/utils/AddressConfigurations.java @@ -0,0 +1,62 @@ +package cn.ac.iie.utils; + +import java.util.Properties; + + + +public final class AddressConfigurations { + + private static Properties propCommon = new Properties(); +// private static Properties propService = new Properties(); + + + public static String getStringProperty(Integer type, String key) { + if (type == 0) { + return propCommon.getProperty(key); +// } else if (type == 1) { +// return propService.getProperty(key); + } else { + return null; + } + } + + + public static Integer getIntProperty(Integer type, String key) { + if (type == 0) { + return Integer.parseInt(propCommon.getProperty(key)); +// } else if (type == 1) { +// return Integer.parseInt(propService.getProperty(key)); + } else { + return null; + } + } + + public static Long getLongProperty(Integer type, String key) { + if (type == 0) { + return Long.parseLong(propCommon.getProperty(key)); +// } else if (type == 1) { +// return Long.parseLong(propService.getProperty(key)); + } else { + return null; + } + } + + public static Boolean getBooleanProperty(Integer type, String key) { + if (type == 0) { + return "true".equals(propCommon.getProperty(key).toLowerCase().trim()); +// } else if (type == 1) { +// return propService.getProperty(key).toLowerCase().trim().equals("true"); + } else { + return null; + } + } + + static { + try { + propCommon.load(AddressConfigurations.class.getClassLoader().getResourceAsStream("address_routine.properties")); + } catch (Exception e) { + propCommon = null; + System.err.println("配置加载失败"); + } + } +} diff --git a/src/main/java/cn/ac/iie/utils/TupleUtils.java b/src/main/java/cn/ac/iie/utils/TupleUtils.java new file mode 100644 index 0000000..c0dc410 --- /dev/null +++ b/src/main/java/cn/ac/iie/utils/TupleUtils.java @@ -0,0 +1,21 @@ +package cn.ac.iie.utils; + +import org.apache.storm.Constants; +import org.apache.storm.tuple.Tuple; + +/** + * @author Administrator + */ +public final class TupleUtils { + /** + * 判断是否系统自动发送的Tuple + * + * @param tuple + * @return + */ + public static boolean isTick(Tuple tuple) { + return tuple != null + && Constants.SYSTEM_COMPONENT_ID.equals(tuple.getSourceComponent()) + && Constants.SYSTEM_TICK_STREAM_ID.equals(tuple.getSourceStreamId()); + } +} diff --git a/src/main/java/log4j.properties b/src/main/java/log4j.properties new file mode 100644 index 0000000..dfd50f1 --- /dev/null +++ b/src/main/java/log4j.properties @@ -0,0 +1,25 @@ +#Log4j +log4j.rootLogger=info,console,file +# 控制台日志设置 +log4j.appender.console=org.apache.log4j.ConsoleAppender +log4j.appender.console.Threshold=error +log4j.appender.console.layout=org.apache.log4j.PatternLayout +log4j.appender.console.layout.ConversionPattern=[%d{yyyy-MM-dd HH\:mm\:ss}] [%-5p] [Thread\:%t] %l %x - <%m>%n + +# 文件日志设置 +log4j.appender.file=org.apache.log4j.DailyRollingFileAppender +log4j.appender.file.Threshold=error +log4j.appender.file.encoding=UTF-8 +log4j.appender.file.Append=true +#路径请用相对路径，做好相关测试输出到应用目下 +log4j.appender.file.file=${nis.root}/log/galaxy-name.log +log4j.appender.file.DatePattern='.'yyyy-MM-dd +log4j.appender.file.layout=org.apache.log4j.PatternLayout +#log4j.appender.file.layout.ConversionPattern=%d{HH:mm:ss} %X{ip} [%t] %5p %c{1} %m%n +log4j.appender.file.layout.ConversionPattern=[%d{yyyy-MM-dd HH\:mm\:ss}] [%-5p] %X{ip} [Thread\:%t] %l %x - %m%n +#MyBatis 配置，com.nis.web.dao是mybatis接口所在包 +log4j.logger.com.nis.web.dao=debug +#bonecp数据源配置 +log4j.category.com.jolbox=debug,console + + diff --git a/src/test/java/cn/ac/iie/test/SubcribeIdBolt.java b/src/test/java/cn/ac/iie/test/SubcribeIdBolt.java new file mode 100644 index 0000000..add969a --- /dev/null +++ b/src/test/java/cn/ac/iie/test/SubcribeIdBolt.java @@ -0,0 +1,161 @@ +package cn.ac.iie.test; + +import cn.ac.iie.common.AddressConfig; +import cn.ac.iie.utils.TupleUtils; +import com.alibaba.fastjson.JSONObject; +import com.zdjizhi.utils.StringUtil; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellUtil; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.*; +import org.apache.hadoop.hbase.util.Bytes; +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.Tuple; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * @author qidaijie + */ +public class SubcribeIdBolt extends BaseBasicBolt { + private static Logger logger = Logger.getLogger(SubcribeIdBolt.class); + private static Map subIdMap; + private List putList; + private static Connection connection; + + static { + // 管理Hbase的配置信息 + Configuration configuration = HBaseConfiguration.create(); + // 设置zookeeper节点 + configuration.set("hbase.zookeeper.quorum", AddressConfig.HBASE_ZOOKEEPER_SERVERS); + try { + connection = ConnectionFactory.createConnection(configuration); + } catch (IOException e) { + e.printStackTrace(); + } + } + + @Override + public void prepare(Map stormConf, TopologyContext context) { + subIdMap = new HashMap<>(16); + putList = new ArrayList<>(); + getAll(); + } + + @Override + public void execute(Tuple tuple, BasicOutputCollector collector) { + try { + if (TupleUtils.isTick(tuple)) { + insertData(putList); + } else { + String message = tuple.getString(0); + if (StringUtil.isNotBlank(message)) { + JSONObject jsonObject = JSONObject.parseObject(message); + String ip = jsonObject.getString("framed_ip"); + String account = jsonObject.getString("account"); + dataValidation(ip, account, putList); + if (putList.size() == AddressConfig.LIST_SIZE_MAX) { + insertData(putList); + } + } + } + } catch (Exception e) { + logger.error("Radius写入Redis出现异常", e); + } + + } + + @Override + public Map getComponentConfiguration() { + Map conf = new HashMap(16); + conf.put(org.apache.storm.Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, + AddressConfig.TOPOLOGY_TICK_TUPLE_FREQ_SECS); + return conf; + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + + } + + /** + * 获取所有的 key value + */ + private static void getAll() { + try { + Table table = connection.getTable(TableName.valueOf("sub:" + AddressConfig.HBASE_TABLE_NAME)); + Scan scan2 = new Scan(); + ResultScanner scanner = table.getScanner(scan2); + for (Result result : scanner) { + Cell[] cells = result.rawCells(); + for (Cell cell : cells) { + subIdMap.put(Bytes.toString(CellUtil.cloneRow(cell)), Bytes.toString(CellUtil.cloneValue(cell))); + } + } + scanner.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + + /** + * 写入数据到hbase + */ + private static void insertData(List putList) { + Table table = null; + try { + table = connection.getTable(TableName.valueOf("sub:" + AddressConfig.HBASE_TABLE_NAME)); + table.put(putList); + logger.error("写入hbase数目：" + putList.size()); + putList.clear(); + } catch (IOException e) { + e.printStackTrace(); + } finally { + try { + if (table != null) { + table.close(); + } + } catch (IOException e) { + e.printStackTrace(); + } + } + + } + + /** + * 验证数据并与内存中的对比 + * + * @param ip framed_ip + * @param account account + */ + private static void dataValidation(String ip, String account, List putList) { + if (StringUtil.isNotBlank(ip) && StringUtil.isNotBlank(account)) { +// String s = ip.split("\\.")[0]; +// if (!AddressConfig.CHECK_IP_SCOPE.contains(s)) { + if (subIdMap.containsKey(ip)) { + if (!subIdMap.get(ip).equals(account)) { + Put put = new Put(ip.getBytes()); + put.addColumn("subscribe_id".getBytes(), "account".getBytes(), account.getBytes()); + putList.add(put); + subIdMap.put(ip, account); + } + } else { + Put put = new Put(ip.getBytes()); + put.addColumn("subscribe_id".getBytes(), "account".getBytes(), account.getBytes()); + putList.add(put); + subIdMap.put(ip, account); + } +// } + } + } +} diff --git a/src/test/java/cn/ac/iie/test/SubcribeIdBoltone.java b/src/test/java/cn/ac/iie/test/SubcribeIdBoltone.java new file mode 100644 index 0000000..00b54a8 --- /dev/null +++ b/src/test/java/cn/ac/iie/test/SubcribeIdBoltone.java @@ -0,0 +1,117 @@ +package cn.ac.iie.test; + +import cn.ac.iie.common.AddressConfig; +import com.alibaba.fastjson.JSONObject; +import com.zdjizhi.utils.StringUtil; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellUtil; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.*; +import org.apache.hadoop.hbase.util.Bytes; +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.io.IOException; +import java.util.HashMap; +import java.util.Map; + +/** + * @author qidaijie + */ +public class SubcribeIdBoltone extends BaseBasicBolt { + private static Logger logger = Logger.getLogger(SubcribeIdBoltone.class); + private static Map subIdMap; + private static Connection connection; + + static { + // 管理Hbase的配置信息 + Configuration configuration = HBaseConfiguration.create(); + // 设置zookeeper节点 + configuration.set("hbase.zookeeper.quorum", AddressConfig.HBASE_ZOOKEEPER_SERVERS); + try { + connection = ConnectionFactory.createConnection(configuration); + } catch (IOException e) { + e.printStackTrace(); + } + } + + @Override + public void prepare(Map stormConf, TopologyContext context) { + subIdMap = new HashMap<>(16); + getAll(); + } + + @Override + public void execute(Tuple tuple, BasicOutputCollector collector) { + try { + String message = tuple.getString(0); + if (StringUtil.isNotBlank(message)) { + JSONObject jsonObject = JSONObject.parseObject(message); + String ip = jsonObject.getString("framed_ip"); + String account = jsonObject.getString("account"); +// dataValidation(ip, account, collector); + collector.emit(new Values(ip + "-" + account)); + } + } catch (Exception e) { + logger.error("Radius写入Redis出现异常", e); + } + + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + declarer.declare(new Fields("connLog")); + } + + /** + * 获取所有的 key value + */ + private static void getAll() { + try { + Table table = connection.getTable(TableName.valueOf("sub:" + AddressConfig.HBASE_TABLE_NAME)); + Scan scan2 = new Scan(); + ResultScanner scanner = table.getScanner(scan2); + for (Result result : scanner) { + Cell[] cells = result.rawCells(); + for (Cell cell : cells) { + subIdMap.put(Bytes.toString(CellUtil.cloneRow(cell)), Bytes.toString(CellUtil.cloneValue(cell))); + } + } + scanner.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + + + /** + * 验证数据并与内存中的对比 + * + * @param ip framed_ip + * @param account account + */ + private static void dataValidation(String ip, String account, BasicOutputCollector collector) { + if (StringUtil.isNotBlank(ip) && StringUtil.isNotBlank(account)) { +// String s = ip.split("\\.")[0]; +// if (!AddressConfig.CHECK_IP_SCOPE.contains(s)) { + if (subIdMap.containsKey(ip)) { + if (!subIdMap.get(ip).equals(account)) { + subIdMap.put(ip, account); + collector.emit(new Values(ip + "-" + account)); + } + } else { + subIdMap.put(ip, account); + collector.emit(new Values(ip + "-" + account)); + } +// } + } + } +} diff --git a/src/test/java/cn/ac/iie/test/subTest.java b/src/test/java/cn/ac/iie/test/subTest.java new file mode 100644 index 0000000..468517a --- /dev/null +++ b/src/test/java/cn/ac/iie/test/subTest.java @@ -0,0 +1,32 @@ +package cn.ac.iie.test; + +import cn.ac.iie.common.AddressConfig; +import com.zdjizhi.utils.StringUtil; + +public class subTest { + + + public static void main(String[] args) { + boolean validation = dataValidation("1.168.40.123", "abcd"); + System.out.println(validation); + + if ("10,100,192".contains("1")){ + System.out.println("yes"); + }else{ + System.out.println("no"); + } + + String s = "192.168.40.125-abcd"; + System.out.println(s.split("-")[0]); + } + + + private static boolean dataValidation(String ip, String account) { + if (StringUtil.isNotBlank(ip) && StringUtil.isNotBlank(account)) { + String s = ip.split("\\.")[0]; + return !AddressConfig.CHECK_IP_SCOPE.contains(s); + } + return false; + } + +}