如果用户量增加后为了解决吞吐量问题,需要引入集群,在openfire中提供了集群的支持,另外也实现了两个集群插件:hazelcast和clustering。为了了解情况集群的工作原理,我就沿着openfire的源代码进行了分析,也是一次学习的过程。
数据库因为对于openfire来说基本上是透明的,所以这块就交给数据库本身来实现。缓存数据缓存是存在内存里的,所以这部分是要同步的sessionsession在openfire并不需要所有实例同步,但是需要做用户路由缓存,否则发消息时找不到对应的会话。由此用户路由还是要同步的。
- 缓存接口
public interface Cache<K,V> extends java.util.Map<K,V>
如果不开启集群时缓存的默认缓存容器类是:public class DefaultCache<K, V> ,实际上DefaultCache就是用一个Hashmap来存数据的。
- 缓存工厂类
public class CacheFactory
/**
* Returns the named cache, creating it as necessary.
*
* @param name the name of the cache to create.
* @return the named cache, creating it as necessary.
*/
@SuppressWarnings("unchecked")
public static synchronized <T extends Cache> T createCache(String name) {
T cache = (T) caches.get(name);
if (cache != null) {
return cache;
}
cache = (T) cacheFactoryStrategy.createCache(name); log.info("Created cache [" + cacheFactoryStrategy.getClass().getName() + "] for " + name); return wrapCache(cache, name);
}
上面代码中会通过缓存工厂策略对象来创建一个缓存容器,最后warpCache方法会将此容器放入到caches中。
- 缓存工厂类的策略
- startup
public static synchronized void startup() {
if (isClusteringEnabled() && !isClusteringStarted()) {
initEventDispatcher();
CacheFactory.startClustering();
}
}
- 会使用集群的缓存工厂策略来启动,同时使自己加入到集群中。
- 开启一个线程用于同步缓存的状态
- shutdown
public static void doClusterTask(final ClusterTask<?> task) {
cacheFactoryStrategy.doClusterTask(task);
}
这里有个限定就是必须是ClusterTask派生的类才行,看看它的定义:
public interface ClusterTask<V> extends Runnable, Externalizable { V getResult(); }
主要是为了异步执行和序列化,异步是因为不能阻塞,而序列化当然就是为了能在集群中传送。
- 缓存策略工厂类(ClusteredCacheFactory)
public class ClusteredCacheFactory implements CacheFactoryStrategy {
首先是startCluster方法用于启动集群,主要完成几件事情:
- 设置缓存序列化工具类,ClusterExternalizableUtil。这个是用于集群间数据复制时的序列化工具
- 设置远程session定位器,RemoteSessionLocator,因为session不同步,所以它主要是用于多实例间的session读取
- 设置远程包路由器ClusterPacketRouter,这样就可以在集群中发送消息了
- 加载Hazelcast的实例设置NodeID,以及设置ClusterListener
/**
* Notification message indicating that this JVM has joined a cluster.
*/
@SuppressWarnings("unchecked")
public static synchronized void joinedCluster() {
cacheFactoryStrategy = clusteredCacheFactoryStrategy;
// Loop through local caches and switch them to clustered cache (copy content)
for (Cache cache : getAllCaches()) {
// skip local-only caches
if (localOnly.contains(cache.getName())) continue;
CacheWrapper cacheWrapper = ((CacheWrapper) cache);
Cache clusteredCache = cacheFactoryStrategy.createCache(cacheWrapper.getName());
clusteredCache.putAll(cache);
cacheWrapper.setWrappedCache(clusteredCache);
}
clusteringStarting = false;
clusteringStarted = true;
log.info("Clustering started; cache migration complete");
}
这里可以看到会读取所有的缓存容器并一个个的使用Wrapper包装一下,然后用同样的缓存名称去createCache一个新的Cache,这步使用的是切换后的集群缓存策略工厂,也就是说会使用ClusteredCacheFactory去创建新的缓存容器。最后再将cache写入到新的clusteredCache 里,这样就完成了缓存的切换。
public Cache createCache(String name) {
// Check if cluster is being started up
while (state == State.starting) {
// Wait until cluster is fully started (or failed)
try {
Thread.sleep(250);
}
catch (InterruptedException e) {
// Ignore
}
}
if (state == State.stopped) {
throw new IllegalStateException("Cannot create clustered cache when not in a cluster");
}
return new ClusteredCache(name, hazelcast.getMap(name));
}
这里使用的是ClusteredCache,而且最重要的是传入的第二个map参数换成了hazelcast的了,这样之后再访问这个缓存容器时已经不再是原先的本地Cache了,已经是hazelcast的map对象。hazelcast会自动对map的数据进行同步管理,这也就完成了缓存同步的功能。
- 集群计算
那就看hazelcast的实现吧,在ClusteredCacheFactory中doClusterTask举个例子吧:
public void doClusterTask(final ClusterTask task) {
if (cluster == null) { return; }
Set<Member> members = new HashSet<Member>();
Member current = cluster.getLocalMember();
for(Member member : cluster.getMembers()) {
if (!member.getUuid().equals(current.getUuid())) {
members.add(member);
}
}
if (members.size() > 0) {
// Asynchronously execute the task on the other cluster members
logger.debug("Executing asynchronous MultiTask: " + task.getClass().getName());
hazelcast.getExecutorService(HAZELCAST_EXECUTOR_SERVICE_NAME).submitToMembers(
new CallableTask<Object>(task), members);
} else {
logger.warn("No cluster members selected for cluster task " + task.getClass().getName());
}
}
过程就是,先获取到集群中的实例成员,当然要排除自己。然后hazelcast提供了ExecutorService来执行这个task,方法就是submiteToMembers。这样就提交了一个运算任务。只不过具体是如何分配计算并汇集结果倒真不太清楚。
总结
Openfire集群源码分析的更多相关文章
-
ZK集群源码解读
1.1. 集群模式 1.1.1. 数据同步总流程 1.1.1.1. OBSERVING 1.1.1.2. FOLLOWING 1.1.1.3. LEADING 1.1.2. 领导选举 1.1.2. ...
-
WordCount 远程集群源码
package test; import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop ...
-
quartz集群调度机制调研及源码分析---转载
quartz2.2.1集群调度机制调研及源码分析引言quartz集群架构调度器实例化调度过程触发器的获取触发trigger:Job执行过程:总结:附: 引言 quratz是目前最为成熟,使用最广泛的j ...
-
(1)quartz集群调度机制调研及源码分析---转载
quartz2.2.1集群调度机制调研及源码分析 原文地址:http://demo.netfoucs.com/gklifg/article/details/27090179 引言quartz集群架构调 ...
-
[转]RMI方式Ehcache集群的源码分析
RMI方式Ehcache集群的源码分析 Ehcache不仅支持基本的内存缓存,还支持多种方式将本地内存中的缓存同步到其他使用Ehcache的服务器中,形成集群.如下图所示: Ehcache支持 ...
-
RMI方式Ehcache集群的源码分析
Ehcache不仅支持基本的内存缓存,还支持多种方式将本地内存中的缓存同步到其他使用Ehcache的服务器中,形成集群.如下图所示: Ehcache支持多种集群方式,下面以RMI通信方式为例,来具体分 ...
-
分布式缓存技术之Redis_Redis集群连接及底层源码分析
目录 1. Jedis 单点连接 2. Jedis 基于sentinel连接 基本使用 源码分析 本次源码分析基于: jedis-3.0.1 1. Jedis 单点连接 当是单点服务时,Java ...
-
Dubbo 源码分析 - 集群容错之 LoadBalance
1.简介 LoadBalance 中文意思为负载均衡,它的职责是将网络请求,或者其他形式的负载"均摊"到不同的机器上.避免集群中部分服务器压力过大,而另一些服务器比较空闲的情况.通 ...
-
Dubbo 源码分析 - 集群容错之 Cluster
1.简介 为了避免单点故障,现在的应用至少会部署在两台服务器上.对于一些负载比较高的服务,会部署更多台服务器.这样,同一环境下的服务提供者数量会大于1.对于服务消费者来说,同一环境下出现了多个服务提供 ...
随机推荐
-
webpack那些事儿
webpack那些事儿01-webpack到底是什么 webpack那些事儿02-从零开始 webpack那些事儿03-热插拔 hot webpack那些事儿04-spa项目实战分析 webpack那 ...
-
Asp.net项目路径获取方法【转】
获取项目完整的绝对路径string path = System.AppDomain.CurrentDomain.BaseDirectory.ToString();string path = Serve ...
-
一起来Fit TDMA over WiFi(1)
1 概述 WiFI TDMA领域,2009年Sam Leffler在<TDMA for Long Distance Wireless Networks>首次系统提出了TDMA技术方案,并 ...
-
[strongswan] strongswan是如何实现与xfrm之间的trap机制的
目录 strongswan与xfrm之间的trap机制 0. 1. 前言 2. 描述 2.1 none 2.2 trap 3. 实验与过程 3.1 trap实验 3.2 none实验 4 背景知识 5 ...
-
【优秀的艺术文字和图标设计软件】Art Text 3.2.3 for Mac
[简介] Art Text 3.2.3 版本,这是一款Mac上简单易用的艺术文字和图标设计软件,今这款软件内置了大量的背景纹理和特效,能够让我们非常快速的制作出漂亮的图标,相比专业的PS,Art ...
-
suoi38 卖XY序列 (贪心+前缀和)
因为只能带一个,买卖价格又一样,所以只要右边的比左边的大,就从这买下来然后带到下一个卖掉就行了(我想到别处再卖的话大不了再重新买回来嘛) 所以给max(w[i]-w[i-1],0)维护一个前缀和就行了 ...
-
tailor multi fragment &;&; cutom-amd script demo 说明
tailor 官方demo 中提供了一个multi fragment 的demo,这个比较简单,就是使用不同的 后端server 做为fragment ,然后使用 html tag 进行加载就可以了. ...
-
Percona XtraDB Cluster 的一些使用限制(PXC 5.7)
Percona XtraDB Cluster有众多的优秀特性,使得mysql集群得以轻松实现.但是不要忽略了它的一些限制.如果你无法接受,或者你的应用程序或数据库(比如使用了memory引擎)对限制无 ...
-
JAVA面试中的陷阱
第一,谈谈final, finally, finalize的区别.最常被问到. 第二,Anonymous Inner Class (匿名内部类) 是否可以extends(继承)其它类,是否可以impl ...
-
Hibernate的10个常见面试问题及答案
在Java J2EE方面进行面试时,常被问起的Hibernate面试问题,大多都是针对基于Web的企业级应用开发者的角色的.Hibernate框架在Java界的成功和高度的可接受性使得它成为了Java ...