- 浏览: 84001 次
- 性别:
- 来自: 郑州
文章分类
- 全部博客 (69)
- java (28)
- linux (6)
- redis (4)
- C# (3)
- 架构 (10)
- java ee (1)
- web (1)
- 操作系统 (7)
- sqlserver (1)
- android (2)
- Hadoop (12)
- 大数据 (21)
- 分布式 事务 消息 (10)
- linux mysql (1)
- 数据库 (3)
- 关于hadoop之bootshell使用 (2)
- 关于hbase---HTableInterfaceFactory (1)
- Spring (3)
- Hbase (5)
- jstorm (10)
- nginx (1)
- 分布式 (1)
- 区块链 (3)
- dubbo (1)
- nacos (1)
- 阿里 (1)
- go (3)
- 缓存 (1)
- memcached (1)
- ssdb (1)
- 源码 (1)
最新评论
-
想个可以用的名字:
楼主,能不能给发一份源代码,1300246542@qqq.co ...
spring+websocket的使用 -
wahahachuang5:
web实时推送技术使用越来越广泛,但是自己开发又太麻烦了,我觉 ...
websocket -
dalan_123:
前提是你用的是spring mvc 才需要加的1、在web.x ...
spring+websocket的使用 -
string2020:
CharacterEncodingFilter这个filter ...
spring+websocket的使用
一、作用
主要是通过结合zookeeper,在zookeeper指定的目录下创建不同的自己 代表不同的transaction state,并赋予不同的内容;通过zookeeper完成对partitions的transaction的管理
二、源码分析
package storm.trident.topology.state;
import backtype.storm.Config;
import backtype.storm.utils.Utils;
import backtype.storm.utils.ZookeeperAuthInfo;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.ProtectACLCreateModePathAndBytesable;
import org.apache.curator.framework.api.PathAndBytesable;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.json.simple.JSONValue;
// 事务状态
// 通过zookeeper来进行完成操作
public class TransactionalState {
// zookeeper 的监护器
CuratorFramework _curator;
// zookeeper 访问控制列表
List<ACL> _zkAcls = null;
// 构建用户的transaction的状态
public static TransactionalState newUserState(Map conf, String id) {
return new TransactionalState(conf, id, "user");
}
// 构建协调器的transaction的状态
public static TransactionalState newCoordinatorState(Map conf, String id) {
return new TransactionalState(conf, id, "coordinator");
}
// 在zookeeper上构建对应的transaction的state的目录
protected TransactionalState(Map conf, String id, String subroot) {
try {
conf = new HashMap(conf);
// zookeeper的root目录
String transactionalRoot = (String)conf.get(Config.TRANSACTIONAL_ZOOKEEPER_ROOT);
// zookeeper的transaction的state的根目录
String rootDir = transactionalRoot + "/" + id + "/" + subroot;
// 根据配置信息中的transactional servers 和 zookeeper servers获取servers 优先使用transactional的servers 否则使用zookeeper servers
List<String> servers = (List<String>) getWithBackup(conf, Config.TRANSACTIONAL_ZOOKEEPER_SERVERS, Config.STORM_ZOOKEEPER_SERVERS);
// 根据配置信息中的transactional ports和zookeeper port设置对应的ports
Object port = getWithBackup(conf, Config.TRANSACTIONAL_ZOOKEEPER_PORT, Config.STORM_ZOOKEEPER_PORT);
// 设定zookeeper权限控制
ZookeeperAuthInfo auth = new ZookeeperAuthInfo(conf);
// 根据config、servers、port、auth完成zookeeper监护器初始化 并启动
CuratorFramework initter = Utils.newCuratorStarted(conf, servers, port, auth);
// 获取zookeeper可用列表
_zkAcls = Utils.getWorkerACL(conf);
try {
// 在zookeeper上创建对应的transaction 根目录node
TransactionalState.createNode(initter, transactionalRoot, null, null, null);
} catch (KeeperException.NodeExistsException e) {
}
try {
// 创建对应的transaction state的node的子根目录
TransactionalState.createNode(initter, rootDir, null, _zkAcls, null);
} catch (KeeperException.NodeExistsException e) {
}
// 关闭初始化监护器
initter.close();
// 开启全局监护器
_curator = Utils.newCuratorStarted(conf, servers, port, rootDir, auth);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
protected static String forPath(PathAndBytesable<String> builder,
String path, byte[] data) throws Exception {
return (data == null)
? builder.forPath(path)
: builder.forPath(path, data);
}
protected static void createNode(CuratorFramework curator, String path,
byte[] data, List<ACL> acls, CreateMode mode) throws Exception {
ProtectACLCreateModePathAndBytesable<String> builder =
curator.create().creatingParentsIfNeeded();
if (acls == null) {
if (mode == null ) {
TransactionalState.forPath(builder, path, data);
} else {
TransactionalState.forPath(builder.withMode(mode), path, data);
}
return;
}
TransactionalState.forPath(builder.withACL(acls), path, data);
}
// 指定path设置data
public void setData(String path, Object obj) {
path = "/" + path;
byte[] ser;
try {
ser = JSONValue.toJSONString(obj).getBytes("UTF-8");
} catch (UnsupportedEncodingException e) {
throw new RuntimeException(e);
}
try {
if(_curator.checkExists().forPath(path)!=null) {
_curator.setData().forPath(path, ser);
} else {
TransactionalState.createNode(_curator, path, ser, _zkAcls,
CreateMode.PERSISTENT);
}
} catch(Exception e) {
throw new RuntimeException(e);
}
}
// 删除指定path
public void delete(String path) {
path = "/" + path;
try {
_curator.delete().forPath(path);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
// 罗列出指定path下所有的子节点
public List<String> list(String path) {
path = "/" + path;
try {
if(_curator.checkExists().forPath(path)==null) {
return new ArrayList<String>();
} else {
return _curator.getChildren().forPath(path);
}
} catch(Exception e) {
throw new RuntimeException(e);
}
}
public void mkdir(String path) {
setData(path, 7);
}
// 获取指定路径的data
public Object getData(String path) {
path = "/" + path;
try {
if(_curator.checkExists().forPath(path)!=null) {
return JSONValue.parse(new String(_curator.getData().forPath(path), "UTF-8"));
} else {
return null;
}
} catch(Exception e) {
throw new RuntimeException(e);
}
}
// 关闭监护器
public void close() {
_curator.close();
}
// 根据指定内容返回对应的结果
private Object getWithBackup(Map amap, Object primary, Object backup) {
Object ret = amap.get(primary);
if(ret==null) return amap.get(backup);
return ret;
}
}
主要是通过结合zookeeper,在zookeeper指定的目录下创建不同的自己 代表不同的transaction state,并赋予不同的内容;通过zookeeper完成对partitions的transaction的管理
二、源码分析
package storm.trident.topology.state;
import backtype.storm.Config;
import backtype.storm.utils.Utils;
import backtype.storm.utils.ZookeeperAuthInfo;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.ProtectACLCreateModePathAndBytesable;
import org.apache.curator.framework.api.PathAndBytesable;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.json.simple.JSONValue;
// 事务状态
// 通过zookeeper来进行完成操作
public class TransactionalState {
// zookeeper 的监护器
CuratorFramework _curator;
// zookeeper 访问控制列表
List<ACL> _zkAcls = null;
// 构建用户的transaction的状态
public static TransactionalState newUserState(Map conf, String id) {
return new TransactionalState(conf, id, "user");
}
// 构建协调器的transaction的状态
public static TransactionalState newCoordinatorState(Map conf, String id) {
return new TransactionalState(conf, id, "coordinator");
}
// 在zookeeper上构建对应的transaction的state的目录
protected TransactionalState(Map conf, String id, String subroot) {
try {
conf = new HashMap(conf);
// zookeeper的root目录
String transactionalRoot = (String)conf.get(Config.TRANSACTIONAL_ZOOKEEPER_ROOT);
// zookeeper的transaction的state的根目录
String rootDir = transactionalRoot + "/" + id + "/" + subroot;
// 根据配置信息中的transactional servers 和 zookeeper servers获取servers 优先使用transactional的servers 否则使用zookeeper servers
List<String> servers = (List<String>) getWithBackup(conf, Config.TRANSACTIONAL_ZOOKEEPER_SERVERS, Config.STORM_ZOOKEEPER_SERVERS);
// 根据配置信息中的transactional ports和zookeeper port设置对应的ports
Object port = getWithBackup(conf, Config.TRANSACTIONAL_ZOOKEEPER_PORT, Config.STORM_ZOOKEEPER_PORT);
// 设定zookeeper权限控制
ZookeeperAuthInfo auth = new ZookeeperAuthInfo(conf);
// 根据config、servers、port、auth完成zookeeper监护器初始化 并启动
CuratorFramework initter = Utils.newCuratorStarted(conf, servers, port, auth);
// 获取zookeeper可用列表
_zkAcls = Utils.getWorkerACL(conf);
try {
// 在zookeeper上创建对应的transaction 根目录node
TransactionalState.createNode(initter, transactionalRoot, null, null, null);
} catch (KeeperException.NodeExistsException e) {
}
try {
// 创建对应的transaction state的node的子根目录
TransactionalState.createNode(initter, rootDir, null, _zkAcls, null);
} catch (KeeperException.NodeExistsException e) {
}
// 关闭初始化监护器
initter.close();
// 开启全局监护器
_curator = Utils.newCuratorStarted(conf, servers, port, rootDir, auth);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
protected static String forPath(PathAndBytesable<String> builder,
String path, byte[] data) throws Exception {
return (data == null)
? builder.forPath(path)
: builder.forPath(path, data);
}
protected static void createNode(CuratorFramework curator, String path,
byte[] data, List<ACL> acls, CreateMode mode) throws Exception {
ProtectACLCreateModePathAndBytesable<String> builder =
curator.create().creatingParentsIfNeeded();
if (acls == null) {
if (mode == null ) {
TransactionalState.forPath(builder, path, data);
} else {
TransactionalState.forPath(builder.withMode(mode), path, data);
}
return;
}
TransactionalState.forPath(builder.withACL(acls), path, data);
}
// 指定path设置data
public void setData(String path, Object obj) {
path = "/" + path;
byte[] ser;
try {
ser = JSONValue.toJSONString(obj).getBytes("UTF-8");
} catch (UnsupportedEncodingException e) {
throw new RuntimeException(e);
}
try {
if(_curator.checkExists().forPath(path)!=null) {
_curator.setData().forPath(path, ser);
} else {
TransactionalState.createNode(_curator, path, ser, _zkAcls,
CreateMode.PERSISTENT);
}
} catch(Exception e) {
throw new RuntimeException(e);
}
}
// 删除指定path
public void delete(String path) {
path = "/" + path;
try {
_curator.delete().forPath(path);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
// 罗列出指定path下所有的子节点
public List<String> list(String path) {
path = "/" + path;
try {
if(_curator.checkExists().forPath(path)==null) {
return new ArrayList<String>();
} else {
return _curator.getChildren().forPath(path);
}
} catch(Exception e) {
throw new RuntimeException(e);
}
}
public void mkdir(String path) {
setData(path, 7);
}
// 获取指定路径的data
public Object getData(String path) {
path = "/" + path;
try {
if(_curator.checkExists().forPath(path)!=null) {
return JSONValue.parse(new String(_curator.getData().forPath(path), "UTF-8"));
} else {
return null;
}
} catch(Exception e) {
throw new RuntimeException(e);
}
}
// 关闭监护器
public void close() {
_curator.close();
}
// 根据指定内容返回对应的结果
private Object getWithBackup(Map amap, Object primary, Object backup) {
Object ret = amap.get(primary);
if(ret==null) return amap.get(backup);
return ret;
}
}
发表评论
-
nacos单机源码调试
2018-12-17 11:35 1173首先从github上获取对应的源码Nacos源码git cl ... -
jstorm源码之RotatingTransactionalState
2016-03-21 19:29 533一、作用 构建一个Rotationg transacti ... -
jstorm源码之PartitionedTridentSpoutExecutor
2016-03-21 19:28 847一、作用 Partition Spout对应的exec ... -
jstorm源码之 RichSpoutBatchExecutor
2016-03-21 19:28 0一、作用 RichSpoutBatchExecutor是IRi ... -
jstorm源码之RotatingMap
2016-03-21 19:27 836一、作用 基于LinkedList + HashM ... -
jstorm源码之 RichSpoutBatchExecutor
2016-03-21 19:24 566一、作用 RichSpoutBatchExecutor是IRi ... -
jstorm源码之TridentTopology
2016-03-16 18:12 2317在jstorm中对应TridentTopology的源码如下, ... -
jstorm操作命令
2016-03-15 18:04 2688启动ZOOPKEEPER zkServer.sh start ... -
JStorm之Supervisor简介
2016-03-15 18:02 1185一、简介Supervisor是JStorm中的工作节点,类似 ... -
JStorm介绍
2016-03-15 17:56 886一、简介Storm是开源的 ... -
mycat的使用---sqlserver和mysql
2016-01-11 14:33 8549数据库中间件mycat的使 ... -
jstorm安装
2015-12-03 19:43 1687关于jstorm单机安装可以 ... -
HBase系列一
2015-11-30 16:17 675关于hbase 一、客户端类 HTable 和 HTabl ... -
spring hadoop系列(六)---HbaseSystemException
2015-11-30 09:13 415一、源码 /** * HBase Data Access e ... -
spring hadoop系列(五)---spring hadoop hbase之HbaseSynchronizationManager
2015-11-27 18:16 843一、源码如下 /** * Synchronization m ... -
spring hadoop 系列(二)
2015-11-27 15:26 561一、源码分析 /** * * HbaseAccesso ... -
spring hadoop之batch处理(二)
2015-11-24 18:10 1471一、测试 public class MrBatchApp { ... -
spring hadoop之mapreduce batch
2015-11-24 15:51 583一、测试 // 定义hadoop configuration ... -
centos6.7 64位 伪分布 安装 cdh5.4.8 + jdk 8
2015-11-09 00:37 2269一、安装JAVA # 创建JAVA的目录 mkdir -p / ... -
spring hadoop系列二(MapReduce and Distributed cache)
2015-11-06 15:22 946关于MapReduce and Distributed Ca ...
相关推荐
jstorm源码解析之bolt异常处理方法.docx
Storm 源码分析 - 李明,王晓鹏
下面小编就为大家带来一篇jstorm源码解析之bolt异常处理方法。小编觉得挺不错的,现在就分享给大家,也给大家做个参考。一起跟随小编过来看看吧
JStorm原始码学习:主要包含Storm重新启动,Nimbus启动,Supervisor启动,Executor创建和启动 风暴编程模型 Nimbus:负责资源分配和任务调度。 主管:负责接受nimbus分配的任务,启动和停止属于自己管理的worker...
阿里中间件性能挑战赛初赛-RocketMQ+JStorm+Tair实时统计双11交易金额源码+学习说明.zip阿里中间件性能挑战赛初赛-RocketMQ+JStorm+Tair实时统计双11交易金额源码+学习说明.zip阿里中间件性能挑战赛初赛-RocketMQ+...
jstorm 2.1.1 javadoc
【大纲】 现状 Jstorm概叙 & 流式计算 为什么开发Jstorm 特性 Question and Answer.
jstorm集成kafka插件demo
jstorm框架介绍,包含架构图、jstorm安装部署以及配置、如何在jstorm框架里写业务代码。
jstorm storm 入门demo,包含本地模式 和 集群模式。小小的demo,仅包含4个class。
storm-core-1.0.3-sources.jar 源码文件,1.0.3版本
jstorm-2.2.1
jstorm简单example,仅供本人网络不通,大家无需下载,https://github.com/alibaba/jstorm有免费资源
JStorm 2.2.1 百度网盘下载,Alibaba JStorm 是一个强大的企业级流式计算引擎
java实现jstorm+kafka集成实例,代码可运行。内有readme说明文档
jstorm简单示例,十分适合入门,希望能帮到大家
JStorm是参考storm的实时流式计算框架,在网络IO、线程模型、资源调度、可用性及稳定性上做了持续改进,已被...经过4年发展,阿里巴巴JStorm集群已经成为世界上最大的集群之一,基于JStorm的应用数量超过1000个。 6198}
alibaba-jstorm
jstorm框架default.yaml参数配置项列表详解
Jstorm 、zookeeper集群部署详细操作步骤 Centos7系统