由于项目需要,写了版针对业务的自动化测试代码,主要应用场景在于由于业务日趋复杂,一些公共代码的改动,担心会影响已有业务。还没进行重写,但知识点还是不少的与大家分享实践下。首先,介绍下整个流处理的业务流程。

  首先 从网管实时接入数据到kafka,然后消息接入 进行预处理(这个过程是通过jetty框架,直接用servlet启动的项目,因为考虑到tomcat的并发不够,所以这样用。)随后预处理完 传入kafka,然后storm的不同的topo根据不同的传入类型,进行接入消息的规则匹配,规则是存在于前台的项目中,定时刷入redis(1分钟1刷) 随后加载用户卡数据、用户信息等(这些数据是每晚通过跑mapreduce任务生成的大宽表,直接导入redis),通过redis加载效率非常高,满足实时性(如果redis中不存在数据的情况下,会连接hbase,再进行查询) 随后进行业务处理(有时有些会调各个网管的接口,获取相应业务数据),随后将封装好的数据发总致下游通知拓扑,通知拓扑通过webservice或者restTemple发送值各个其他平台,比如微信,支付宝,短信等,最终将整个运行日志写入hbase。

  首先准备下一些需要的公共类,kafkaclient:

private Properties properties;

	private String defaultTopic;

	private KafkaProducer<K, V> producer;

	public void setProperties(Properties properties) {
this.properties = properties;
} public void setDefaultTopic(String defaultTopic) {
this.defaultTopic = defaultTopic;
} public void setProducer(KafkaProducer<K, V> producer) {
this.producer = producer;
} public void init() {
if (properties == null) {
throw new NullPointerException("kafka properties is null.");
}
this.producer = new KafkaProducer<K, V>(properties);
} public void syncSend(V value) {
ProducerRecord<K, V> producerRecord = new ProducerRecord<K, V>(defaultTopic, value);
try {
producer.send(producerRecord).get();
} catch (Exception e) {
throw new RuntimeException(e);
}
} public void asyncSend(V value) {
ProducerRecord<K, V> producerRecord = new ProducerRecord<K, V>(defaultTopic, value);
producer.send(producerRecord);
}

  HbaseUtil:

private static final Logger logger = LoggerFactory.getLogger(HbaseResult.class);

	private Gson gson = new Gson();

	private HConnection connection = null;

	private Configuration conf = null;

	private String logFile = "D:/error.txt";

	public void init() throws IOException {
logger.info("start init HBasehelper...");
conf = HBaseConfiguration.create();
connection = HConnectionManager.createConnection(conf);
logger.info("init HBasehelper successed!");
} public synchronized HConnection getConnection() throws IOException {
if (connection == null) {
connection = HConnectionManager.createConnection(conf);
}
return connection;
} private synchronized void closeConnection() {
if (connection != null) {
try {
connection.close();
} catch (IOException e) {
}
}
connection = null;
}

  kafkaClient主要负责将读取报文的信息发送至kafka,随之又topo自行运算,最终使用通过调用hbaseUtil,对相应字段的比对查询。

那么下面对整个自动化测试的流程进行说明:

  一、导入前台活动  由于是自动化测试,我们不可能每次都手工上下线,或在页面配置启用某个活动,所以通过直接调用前台系统 导入功能 的方法,将活动配置写入mysql数据库,并进行状态的切换。s

   List<String> codeList = new ArrayList<String>();
List<String> activityIdList = new ArrayList<String>();
try {
FileBody bin = new FileBody(new File("D:/activityTest/activity.ac"));
InputStream in = bin.getInputStream();
BufferedReader br = new BufferedReader(new InputStreamReader(in));
String tr = null;
while((tr = br.readLine())!=null){
HttpPost httppost = new HttpPost("http://*********:8088/***/***/***/import");
CloseableHttpClient httpclient = HttpClients.createDefault();
    ObjectMapper mapper = new ObjectMapper();
ActivityConfig cloneActivity = null;
cloneActivity = mapper.readValue(tr.toString(),ActivityConfig.class);
List<ActivityConfig> cloneActivitys = new ArrayList<ActivityConfig>();//存放所有的活动
cloneActivitys.add(cloneActivity); for (ActivityConfig cloneActivity1 : cloneActivitys) {
  String code = cloneActivity1.getCode();
codeList.add(code);
}
HttpEntity reqEntity = MultipartEntityBuilder.create()
.addPart("file", bin)
.build();
httppost.setEntity(reqEntity);
System.out.println("executing request " + httppost.getRequestLine());
CloseableHttpResponse response = httpclient.execute(httppost);
System.out.println(response.getStatusLine());
HttpEntity resEntity = response.getEntity();
if (resEntity != null) {
System.out.println("Response content length: " + resEntity.getContentLength());
}
EntityUtils.consume(resEntity);
response.close();
httpclient.close();
}
for(String code : codeList){
  String code1 = "'" + code + "'";
if(StringUtils.isNotEmpty(activityCode)){
  activityCode.append(",");
}
activityCode.append(code1);
}
      }
      return activityIdList;
    ]

二、读取准备好的报文数据(xml形式需通过解析,数据分隔符格式读取后直接发送至kafka)

public String readTxt() throws IOException{
StringBuffer sendMessage = new StringBuffer();
BufferedReader br = null;
try {
br = new BufferedReader(
new InputStreamReader(new FileInputStream(MessageText), "UTF-8"));
String line = "";
while((line = br.readLine()) != null){
if (line.contains("<?xml")) {
int beginIndex = line.indexOf("<?xml");
line = line.substring(beginIndex);
}
sendMessage.append(line);
}
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
} catch (FileNotFoundException e) {
e.printStackTrace();
}finally{
br.close();
}
return sendMessage.toString();
}

三、下来,我们需要将解析后的报文数据写入hbase的相应用户宽表、卡宽表中,以便storm拓扑中进行用户数据的加载,这里的rowkey为预分区过的。

HbaseResult baseHelper = new HbaseResult();
baseHelper.init();
tableName = "CARD****";
rowkey = HTableManager.generatRowkey(cardNo);
data.put("*****", "10019");
data.put("*****", cardNo);
data.put("*****", certNo);
data.put("*****", "A");
data.put("*****", "1019");
data.put("*****", supplementCardNo);
data.put("*****", "10020");
data.put("*****", certNo);
data.put("*****", cardType);
data.put("*****", cardType);
data.put("*****", cardNo.substring(12,16));
data.put("*****", "F");
data.put("*****", "ysy");
Put put = new Put(Bytes.toBytes(rowkey));
for (Entry<String, String> rs : data.entrySet()) {
put.add(HTableManager.DEFAULT_FAMILY_NAME, Bytes.toBytes(rs.getKey()), Bytes.toBytes(rs.getValue()));
}
baseHelper.put(tableName, put);
System.out.println("rowkey:"+rowkey);
data.clear();

四、随后就可进行消息的发送(发送至集群的kafka)

KafkaInit();
FSTConfiguration fstConf = FSTConfiguration.getDefaultConfiguration();
kafkaClient.syncSend(fstConf.asByteArray(kafkaData));

五、最终进行发送数据的字段对比(通过报文中的,预设的数据字段 与 最终输出的字段或结果进行对比,随后追加写入输出文件)

Result result = baseHelper.getResult("EVENT_LOG_DH", messageKey);
//对比字段
baseHelper.compareData(dataMap, result,activityCode); public Result getResult(String tableName, String rowKey) throws IOException {
Get get = new Get(Bytes.toBytes(rowKey));
Result result = null;
HTableInterface tableInterface = null;
try {
tableInterface = getConnection().getTable(tableName);
result = tableInterface.get(get);
return result;
} catch (Exception e) {
closeConnection();
logger.error("", e);
} finally {
if (tableInterface != null) {
tableInterface.close();
}
} public void compareData(Map<String,Object> messageData, Result res,List<String> activityCode) throws IOException{
List<String> Messages = new ArrayList<String>();
for (Cell cell : res.rawCells()) {
String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
if(Bytes.toString(CellUtil.cloneQualifier(cell)).equalsIgnoreCase("VARIABLESETS")){
System.out.println(qualifier + "[" + new Gson().fromJson(Bytes.toString(CellUtil.cloneValue(cell)), Map.class) + "]");
@SuppressWarnings("unchecked")
Map<String,Object> data = gson.fromJson(Bytes.toString(CellUtil.cloneValue(cell)), Map.class);
String message = "";
for(String datakey : data.keySet()){
if(messageData.containsKey(datakey)){ String dataValue = getString(data,datakey);
String messageValue = getString(messageData,datakey); if(datakey.equals("dh22")){
dataValue = dataValue.substring(0,dataValue.indexOf("."));
messageValue = messageValue.substring(0,messageValue.indexOf("."));
}
if(dataValue.equals(messageValue)){
message = datakey + " = " + dataValue + " 与报文中的 " + dataValue + "对比相同";
Messages.add(message);
}else{
message = datakey + " = " + dataValue + " 与报文中的 " + dataValue + "不一致!!!";
Messages.add(message);
}
}
} }
if(Bytes.toString(CellUtil.cloneQualifier(cell)).equalsIgnoreCase("NOTIFY__")){ }
} if(Messages.size() > 0){
StringBuffer sb = new StringBuffer();
for(String error : Messages){
sb.append(error).append("\n");
}
FileWriter fw = new FileWriter(logFile,true);
fw.write("\n----------------------");
fw.write(sb.toString());
fw.flush();
fw.close();
}else{
String sb = "没有对不上的字段呀";
FileWriter fw = new FileWriter(logFile);
fw.write(sb);
fw.flush();
fw.close();
}
}

六、清除导入的数据等信息,整个流程结束~

public void delHbaseData(String cardNo,String certNo) throws IOException{
String rowkeyCard = HTableManager.generatRowkey(cardNo) ;
String rowKeyUse = HTableManager.generatRowkey(certNo) ;
Delete delData = null;
HTableInterface tableInterface = null;
String tableName = "";
try {
tableInterface = getConnection().getTable(tableName);
tableInterface.delete(delData);
} return;
} catch (Exception e) {
closeConnection();
logger.error("", e);
} finally {
if (tableInterface != null) {
tableInterface.close();
}
}
}

  

初版storm项目全流程自动化测试代码实现的更多相关文章

  1. 用intellij Idea加载eclipse的maven项目全流程

    eclipse的maven项目目录 全流程 加载项目 打开intellij Idea file -> new -> module from existing Sources  选择.pom ...

  2. Vue开发项目全流程

    只记录vue项目开发流程,不说明怎样安装node和vue-cli等 确认安装 安装好node之后,可查看是否安装成功,有版本则安装成功.输入node -v 查看vue是否安装成功,有版本则安装成功.输 ...

  3. Web项目中手机注册短信验证码实现的全流程及代码

    最近在做只能净化器的后台用户管理系统,需要使用手机号进行注册,找了许久才大致了解了手机验证码实现流程,今天在此和大家分享一下. 我们使用的是榛子云短信平台, 官网地址:http://smsow.zhe ...

  4. VUE从入门到放弃(项目全流程)————VUE

    VUE从入门到放弃(第一天)--整体流程 先想想一个项目,vue项目是从什么到什么,然后再什么的?那是什么呢? 搭建 ( vue-cli) 代码内容 运行 封装 成品 一.搭建(脚手架vue-cli) ...

  5. vue项目搭建通过vue-cli包括组件路由的使用实现基本的前端项目全流程

    github上项目地址:https://github.com/comonly/javaweb_blog/tree/master/blog_diverse_frontend 具体搭建流程及实现方式:ht ...

  6. 【Heritrix基础教程之4】开始一个爬虫抓取的全流程代码分析

    在创建一个job后,就要开始job的运行,运行的全流程如下: 1.在界面上启动job 2.index.jsp 查看上述页面对应的源代码 <a href='"+request.getCo ...

  7. 基于Maven构建开发第一个Storm项目

    前面说过了Storm的测试项目,那么此时我们更想自己写一个小项目来练练手,首先我们自己的Windows系统上首先应该安装好maven,然后启动Eclipse for JavaEE版本,接下来开始建立项 ...

  8. 全球首个全流程跨平台界面开发套件,PowerUI分析

    一.       首个全流程跨平台界面开发套件,PowerUI正式发布 UIPower在DirectUI的基础上,自主研发全球首个全流程跨平台界面开发套件PowerUI(PUI)正式发布,PowerU ...

  9. react-create-app 构建react项目的流程以及需要注意的地方

    Hello 小伙伴们,如果觉得本文还不错,记得给个 star , 小伙伴们的 star 是我持续更新的动力!GitHub 地址 React 系列文章代码地址 一 目录 不折腾的前端,和咸鱼有什么区别 ...

随机推荐

  1. IOS常用第三方开源类库&组件

    1.AFNetworking AFNetworking 采用 NSURLConnection + NSOperation, 主要方便与服务端 API 进行数据交换, 操作简单, 功能强大, 现在许多人 ...

  2. 手把手windows64位配置安装python2.7

    这几天公司要用到python的一些算法,让我调研一番,之前对Python一次没接触的我在安装配置环境的时候由于版本的问题,折腾了好久,这里简单介绍一下我的安装方法,需要安装pyhton的朋友可以不再向 ...

  3. IOS 录像软件

    http://iphone.91.com/tutorial/cjjc/140430/21683219.html

  4. install-file -Dfile=J:\project01\workspace\service\lib\javapns-jdk16-163.jar -DgroupId=org.json -Dar

    今天在开发项目的时候发现了一个问题,所以通过博客来记录起来! 为了以后在问题的解决方面能得到借鉴! 问题的现象是这种: 这样会报错的.pom.xml文件他在编译.检查他的文件语法的时候是须要參考库中的 ...

  5. 使用apktool解包和打包apk

    使用apktool解包和打包apk 下载apktool工具 解包 apktool d xxx.apk -f 植入代码 使用apktool解包要植入代码的apk(下面称为A), 使用apktool解包包 ...

  6. svn笔记4属性Properties

    我们已经详细讲述了Subversion存储和检索版本库中不同版本的文件和目录的细节,并且用了好几个章节来论述这个工具的基本功能.如果对于版本化的支持到此为止,从版本控制的角度来看Subversion已 ...

  7. NPOI封装

    一.对于Excel的操作封装主要针对对于Excel的写入的封装,对于打印基类的封装主要引入了两个概念 1.打印区域模板类 2.打印集合类 如实例模板中需要打印的区域模块 如实例模板中需要打印的集合类 ...

  8. [Swift]LeetCode287. 寻找重复数 | Find the Duplicate Number

    Given an array nums containing n + 1 integers where each integer is between 1 and n (inclusive), pro ...

  9. Python 冒泡排序、归并排序、快速排序

    冒泡排序 原理: 代码: def bubble_sort2(arr):for j in range(len(arr) - 1, 0, -1): # [n-1, n-2, ....2, 1]for i ...

  10. Ubuntu16.04 静态IP设置

    为VMware虚拟机内安装的Ubuntu 16.04设置静态IP地址NAT方式 1.安装环境 VMware 12 Ubuntu 16.04 x86_64 2.在VMware中,配置网络环境 VMwar ...