技术标签: Java jstorm jstorm入门 源代码
最近接到任务说要使用jstorm处理业务,之前没接触过,只能硬着头皮来,接下来谈谈我这两天的收获
1,怎么了解jstorm,这个答案没什么固定的,但是我个人比较喜欢去看官方的文档,如果官方的文档实在找不到了,再去baidu,baidu不到那就只能翻墙google了,如果还是找不到。。没办法,看源代码吧;
对于官方文档,我现在能找到的只有:
https://github.com/alibaba/jstorm/wiki/JStorm-Chinese-Documentation
该地址是jstorm的中文文档,上面对jstorm做了简要的概述,内容并不是很多,看完了, 至少能对jstorm有个大体的了解
因为我看博客的时候,喜欢在博客上直接找到答案,所以我在这里赘述一下
根据ali的文档得知,jStorm是一个分布式实时计算引擎,jstorm是 从storm发展而来,由ali爸爸开源的,是一个类似Hadoop MapReduce的系统,用户创建任务后,提交给jstorm系统,JStorm系统会将这个任务跑起来,7*24不间断,如果一个worker发生一碗,立即分配一个worker来替换失效的worker(这段文字摘自 alibaba的github),从系统的角度来说,jstorm是类似Mapreduce的调度系统,从数据的角度来说,Jstorm是一套基于流水线的消息处理机制;
说到流水线,你能想到什么?至少从流水两个字,我想到了管道,其实个人理解jstorm的流水线很像是管道,想象一下,一个主管道可以分出多个支流管道,也可以接一个弯管或者其他什么不一样的管,来看张图
这个图完全是偷ali的,在看另一张
也是偷的,从这两张图就可以形象的看出jstorm的思想了
接下来解释几个重要的概念:
1,topology
我的英文差,只能音译了。。拓扑结构图,其实也差不多是这个意思
jstorm所执行的任务,其实就是一个topology,一个topology可以包含多个spout,多个bolt,多级bolt
2,spout
流的来源
3,bolt
可以说是流的去处(这里可以进行业务处理)
简单来说:在jstorm运行环境正常的情况下,开发人员要做的就是
建立一个topology,建立若干spout,建立若干bolt,然后将spout加入到topology,将bolt也加入到topology,这时组成了一个相对完整的拓扑结构图,然后提交给jtorm,齐活!
说起来比较简单,那么实际写起来呢,来看下面的代码
注意️:有个前提是jstorm运行环境正常,
首先是配置文件:看过其他博客的人知道有*.ynml的文件,主要是内容是对拓扑结构的配置,这个建议新手可以直接忽略,先不做修改,仔细研读官方的文档之后再尝试
经过上面的解释,我们知道一个拓扑结构包涵哪些东西了。其实代码也是差不多的
首先:一个TestSpout.java
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
public class TestSpout extends BaseRichSpout {
private static final long serialVersionUID = 4924189748702648696L;
private static final Logger LOG = LoggerFactory.getLogger(TestSpout.class);
SpoutOutputCollector collector;
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.collector = collector;
}
@Override
public void nextTuple() {
this.collector.emit(new Values("test123"));
// this.collector.emit("VEHICLEINFO", new Values("test123"));
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// TODO Auto-generated method stub
declarer.declare(new Fields("VEHICLE"));
// declarer.declareStream("VEHICLEINFO", new Fields("VEHICLE"));
}
}
继承自BaseRichSpout,需要覆盖三个方法,分别来说下:
open:这个应该是在创建拓扑结构,提交的时候会调用一次,具体我没细研究,基本都是这么写
nextTuple:发送数据的源头,发送的方式是调用SpoutOutputCollector.emit方法,数据的方式就是用Values做封装,这里注意下:jstorm的流支持任何可序列化的数据类型,包括bean,只要是可序列化的就可以,数据的来源,看你的应用场景,可以来自数据库,也可以来自mq,或者是Kafka等等,我们做测试,直接写死就好了
declareOutputFields: 这里一开始一直没搞明白是干嘛的,通过无数次失败,发现,其实他是emit对应的,采用未注视的方式提交的时候,在此方法中 new Fields的值启示就是对应Values的值,是一个建值对,在bolt里面可以通过input.getValueByFiled()的方式获取 (get方法名我写了个大概,可能写错,各位亲自己 点 一下就出来了)
那么注视的掉的是什么情况呢,首先注视掉的emit第一个参数是streamId,就相当于我指定要发给谁,本方法中注视掉的是和emit对应的
2,来一个TestBolt
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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;
public class TestBolt extends BaseRichBolt {
private static final long serialVersionUID = 1804939402092249890L;
private static final Logger LOG = LoggerFactory.getLogger(TestBolt.class);
OutputCollector collector;
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
}
@Override
public void execute(Tuple input) {
String s = input.getString(0);
String sf = input.getStringByField("VEHICLE");
LOG.error("receive MSG----------------------" + s + " ---- " + sf);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
}
这个只有方法execute可以说一说,就是获取数据,tuple就是一个数据流,里面有你的数据,你可以通过get参数序号的方式,也可以用Field方式,而这个getValueByField的值就是上面spout中定义的,有兴趣的自己回顾一下
3,来一个TestTopology
import java.util.HashMap;
import java.util.Map;
import com.zuche.jstom.example.topology.spout.TestSpout;
import com.zuche.jstom.example.topology.spout.bolt.TestBolt;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
public class TestTopology {
public static void main(String[] args) {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("testspout", new TestSpout(), 1);
builder.setBolt("testbolt", new TestBolt(), 1).shuffleGrouping("testspout");
// builder.setBolt("testbolt", new TestBolt(), 1).shuffleGrouping("testspout", "VEHICLEINFO");
Map<Object,Object> conf = new HashMap<Object, Object>();
conf.put(Config.TOPOLOGY_WORKERS, 3);
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("testTopology", conf, builder.createTopology());
}
}
这个就是拼装topology的过程,仔细看的发现有个注视的,这个注视的就是上面说的,我指定这个bolt处理哪个streamId的数据,和前面的spout对应起来,仔细看看
接下里就是直接跑就行了
这里说下:如果本地搭建了环境,跑的时候发现有个slf4j的版本不支持的错误信息,是因为jstorm使用了1.4的版本slf4j,你把她们exclude掉,然后自己引入一个高版本的就行了
那么来说说多个bolt的情况
其实就是在上面的代码再加一行代码:
builder.setBolt("testbolt2", new TestBolt2(), 1).shuffleGrouping("testbolt");
记得自己写个TestBolt2,这是数据流就是 testspout > testbolt > testbolt2
这里要说下shuffleGrouping,jstorm中有几个类似后缀的方法,具体的含义大家看文档,我就不多说了,他们的功能并不相同
可以说为了实现同一个目的,可以有好几种写法,但是具体哪个好,还是大家自己体会,我也是新手,不懂。。
举个例子:
builder.setSpout(LocalSpout.LOCAL_SPOUT_NAME, new LocalSpout(), 1);
builder.setBolt(FenceBolt.FENCE_BOLT_NAME, new FenceBolt(), 1).shuffleGrouping(LocalSpout.LOCAL_SPOUT_NAME);
// builder.setBolt(AlarmBolt.ALARM_BOLT_NAME, new AlarmBolt(), 1).shuffleGrouping(FenceBolt.FENCE_BOLT_NAME);
// builder.setBolt(AlarmBolt.ALARM_BOLT_NAME, new AlarmBolt(), 1).fieldsGrouping(FenceBolt.FENCE_BOLT_NAME , new Fields("FENCEVEHICLE"));
builder.setBolt(AlarmBolt.ALARM_BOLT_NAME, new AlarmBolt(), 1).shuffleGrouping(FenceBolt.FENCE_BOLT_NAME, FenceBolt.FENCE_STREAM_ID);
一个spout,两个连续的bolt,第二个bolt的配置方式,三种写法都可以
深层次的区别我还没体会出来,等我体会出来了,再和大家交流
到此,本文结束,有时间话,我会再写一篇介绍概念的
原始数据如下:u1 a,d,b,cu2 a,a,cu3 b,du4 a,d,cu5 a,b,c计算公式使用:sim = U(i)∩U(j) / (U(i)∪U(j))其中: (U(i)∪U(j)) = U(i) + U(j) - U(i)∩U(j) 原始的Hadoop实现需要5轮MR,优化后只需要两轮就可以完成。之前的轮数过多,主要在于计算(U(i)∪U(j
前段时间,有个朋友又出去面试了,这次他面试目标比较清晰,面的都是业务量大、业务比较核心的部门。前前后后去了不少公司,几家大厂里,他说给他印象最深的是字节。3 轮技术面 + 1 轮 HR 面,他最终拿到了 45k*18薪 的 Offer。第一轮主要考察 Java 基础,二、三轮注重对应⽤技术的掌握,以及对过往项目的业务理解。之所以令他印象深刻,是因为每轮都聊了一个半小时以上,确实让他受益匪浅。如果这些你回答不出来,那么我建议你刷一遍题库。我这里有一份Java架构成长笔记和滴滴、美团等大厂的面试真题,帮_字节面试几轮
展开全部题目中数据很大于是我们有了高精度算法思路是什么呢:一个数组里面每一位存的是十进制数三位(也62616964757a686964616fe58685e5aeb931333363383433可以只存一位,但是比较慢)也可以理解成一千进制这样能模拟大整数的乘法、加法#include#include#definemax(a,b)(a>b?a:b)intn;FILE*input,*..._c语言怎么算阶乘相加fac
oracle中的交集、并集、差集_oracle并集,交集,差集
'''Created on 2013-9-27@author: liangqianwu'''#_*_ coding:utf-8_*_from Tkinter import * def frame(root,side): w=Frame(root) w.pack(side=side,expand=YES,fill=BOTH) return wdef butt
mat为matlab常用存储数据的文件格式,python的scipy.io模块中包含保存和加载mat格式文件的API。_scipy.io.loadmat 数据是ndarray
bitcount(unsigned x):统计x中值为1的二进制位数 将x声明为无符号类型是为了保证将x右移时,无论该程序在什么机器上运行,左边空出的位都是0(而不是符号位)填补。1、通过右移操作实现#include <stdio.h>//通过右移操作实现int bitcount(unsigned x) { int b; for(b = 0; x != 0; x >>= 1) { if(x & 01) { b++; } } ._c++ bitcount
package com.it.Stringjiequ;/*需求:实现手机号屏蔽如157****1234,实现:涉及到字符串截取1.首先调用subString()方法确定开始/结束索引截取前三位,然后继续调用subString()方法确定开始,直接截取到最后2.完成字符串的拼接即可** */import java.util.Scanner;public class Stringshoujihaopingbi { public static void main(String[] arg_substring截取手机号
QThread类为我们提供了一种平台无关的管理线程的方式。一个QThread对象管理应用程序中的一个线程,该线程从run()函数开始执行。并且,默认情况下,我们可以在run()函数中通过调用QThread::exec()函数来在当前线程中开启一个事件循环。
之前向LaTeX插入图片的时候,一直使用.jpg格式,编译出来效果很不好,很多地方稍微放大就显得很模糊。后来采用.eps(Encapsulated Post Script)格式的图片,因为是矢量图,即使放大也不会失真,效果很好,推荐使用。下面介绍如何在LaTeX中插入.eps格式的图片。第一步:生成.eps格式的图片1、利用visio画图,另存为pdf格式的图片利用Adobe ..._latex怎么插入eps
1、配置Android的NTP服务器地址目录:frameworks/base/core/res/res/values/config.xml1725 <!-- Remote server that can provide NTP responses. -->1726 <string translatable="false" name="config_ntpServer">cn.ntp.org.cn</string>cn.ntp.org.cn ..._安卓时间服务器地址
1、打开文件fid = fopen('filename', 'permission')fid在正常情况下应该返回一个非负整数,称为文件标识,如果返回的文件标识为-1,则表示fopen无法打开该文件,原因可能是该文件不存在,也可能是用户无权限打开此文件。filename是要打开的文件名称,permission则表示要对文件进行处理的方式.'r':只读文件(reading)'w':只...