Java ExecutorService 重要BUG两例

声明
本文为Gleasy原创文章,转载请指明引自Gleasy团队博客

BUG1
请看下面的代码:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class TestExecutor {
	 public static void main(String[] args) throws InterruptedException {
 while(true) {
 	new TestThread().run();
 }
	}
	 
	public static class TestThread { 
		ExecutorService es = Executors.newFixedThreadPool(3);

		public void run() { 
			Future f = es.submit(new Runnable(){
				@Override
				public void run() {
					System.out.println("Name:" Thread.currentThread().getName() 
				 " doing something"); 
				}
			}); 
 try {
				f.get();
			} catch (Exception e) {
				e.printStackTrace();
			}
		} 
	} 
}

 

上面这段代码在JDK1.6,JRCOKIT28.1.x 以下版本运行,会得到以下结果:

[WARN ] Thread table can't grow past 16383 threads.

[ERROR][thread ] Could not start thread pool-16110-thread-1. errorcode -1
Exception in thread "Main Thread" java.lang.Error: errorcode -1
	at java.lang.Thread.start0(Native Method)
	at java.lang.Thread.start(Thread.java:640)
	at java.util.concurrent.ThreadPoolExecutor.addIfUnderCorePoolSize(ThreadPoolExecutor.java:703)
	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:652)
	at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:78)
	at com.gleasy.library.cloud.dfs.test.TestExecutor$TestThread.run(TestExecutor.java:18)
	at com.gleasy.library.cloud.dfs.test.TestExecutor.main(TestExecutor.java:10)

 

这个结果很怪异,因为TestThread对象不断创建和销毁,ExecutorService对象是TestThread对象的内部属性。当TestThread对象用完之后,es也应该被释放,而不应该对后面其它调用产生影响。但实际上,创建TestThread对象超过16000多次之后,再创建的时候就会报上面的错。这说明,通过 Executors.newFixedThreadPool(3)创建的线程池,会对后续的调用产生影响。这个BUG很阴险,如果稍不注意,在一个经常被创建的对象里面创建了线程池,那么整个JVM都将面临着不能再创建新的线程池的风险。而在研发过程中,这个BUG不会出错,而且是测不出来的。严重性在于它会影响整个JVM所有线程池的创建,令到整个JVM的其它应用都无法使用。

再过一个实验:

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class TestExecutor {
	 public static void main(String[] args) throws InterruptedException {
		 int i=0;
 while(true) {
 	new TestThread().run();
 	i ;
 	System.out.println("i:" i);
 }
	}
	 
	public static class TestThread { 
		ExecutorService es = Executors.newFixedThreadPool(100);

		public void run() { 
			final CountDownLatch c = new CountDownLatch(100);
			for(int i=0;i<100;i ){
				Future f = es.submit(new Runnable(){
					@Override
					public void run() {
						try {
							Thread.sleep(20);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
						c.countDown();
					}
				}); 
			}
			try {
				c.await();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
 //es.shutdown();
		} 
	} 
}

 

结果如下:

i:150
i:151
i:152
i:153
i:154
i:155
i:156
i:157
i:158
i:159
i:160
i:161
[WARN ] Thread table can't grow past 16383 threads.

[ERROR][thread ] Could not start thread pool-162-thread-10. errorcode -1
Exception in thread "Main Thread" java.lang.Error: errorcode -1

 

在上面这个实验中,通过不断创建TestThread对象,在对象里面创建线程池,线程数为100,然后等这些线程全部执行完毕,才开始创建下一下。
从结果看出,在创建第161个线程池的时候,就开始报这个错了。说明这个错误的本质是线程数量,而非池的数量。由始至终,线程数量没有减少,一直在增加,一直增至16382,开始抛上面的异常了。导致这个异常的本质是线程池不会自动关闭!

解决方法是调用ExecutorService的shutdown或者shutdownNow方法将创建的线程池显式关闭。

package com.gleasy.library.cloud.dfs.test;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class TestExecutor {
	 public static void main(String[] args) throws InterruptedException {
 while(true) {
 	new TestThread().run();
 }
	}
	 
	public static class TestThread { 
		ExecutorService es = Executors.newFixedThreadPool(3);

		public void run() { 
			Future f = es.submit(new Runnable(){
				@Override
				public void run() {
					System.out.println("Name:" Thread.currentThread().getName() 
				 " doing something"); 
				}
			}); 
 try {
				f.get();
			} catch (Exception e) {
				e.printStackTrace();
			}
 es.shutdown();
		} 
	} 
}

 

这个解决方法很不完善,因为很多时候,我们都不知道我们要做的事情是否已经做完,什么时候调用shutdown方法合适呢?最理想是在对象被析构时调用,但JAVA没有析构函数,所以,坑爹啊!!不完美的解决方法是,cachedThreadPool全局用同一个,fixsize的threadpool,用完要自己shutdown;或者说不确定什么时候shutdown的就全局大家共用,能确定用完的就自己shutdown.

BUG2
看下面的代码:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

public class TestThreadPool {
	private static final ExecutorService pool = Executors.newCachedThreadPool();

	public static void main(String[] args) online casino  throws InterruptedException {
		for (int i = 0; i < 10; i ) {
			pool.execute(new MyTask(i));
		}

		for (int i = 0; i < 40; i ) {
			System.out.println("poolSize:" ((ThreadPoolExecutor) pool).getPoolSize());
			Thread.sleep(1000L);
		}
		pool.shutdown();
	}

	static class MyTask implements Runnable {
		private int num;

		public MyTask(int index) {
			this.num = index;
		}

		public void run() {
			try {
				Thread.sleep(5000L);
				System.out.println("task " num " executed by " Thread.currentThread());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			if (num < 5) {
				throw new RuntimeException("task " num ",executed by " Thread.currentThread());
			}
		}
	}

}

 

运行结果如下:

poolSize:5
poolSize:5
poolSize:5
poolSize:5
poolSize:5
task 0 executed by Thread[pool-1-thread-1,5,main]
task 1 executed by Thread[pool-1-thread-2,5,main]
task 3 executed by Thread[pool-1-thread-4,5,main]
task 4 executed by Thread[pool-1-thread-5,5,main]
Exception in thread "pool-1-thread-1" task 2 executed by Thread[pool-1-thread-3,5,main]
poolSize:1
Exception in thread "pool-1-thread-4" Exception in thread "pool-1-thread-5" Exception in thread "pool-1-thread-3" java.lang.RuntimeException: task 0,executed by Thread[pool-1-thread-1,5,main]
	at com.gleasy.library.cloud.dfs.test.TestThreadPool$MyTask.run(TestThreadPool.java:37)
	at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:909)
	at java.lang.Thread.run(Thread.java:662)
Exception in thread "pool-1-thread-2" java.lang.RuntimeException: task 1,executed by Thread[pool-1-thread-2,5,main]
	at com.gleasy.library.cloud.dfs.test.TestThreadPool$MyTask.run(TestThreadPool.java:37)
	at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:909)
	at java.lang.Thread.run(Thread.java:662)
java.lang.RuntimeException: task 4,executed by Thread[pool-1-thread-5,5,main]
	at com.gleasy.library.cloud.dfs.test.TestThreadPool$MyTask.run(TestThreadPool.java:37)
	at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:909)
	at java.lang.Thread.run(Thread.java:662)
java.lang.RuntimeException: task 3,executed by Thread[pool-1-thread-4,5,main]
	at com.gleasy.library.cloud.dfs.test.TestThreadPool$MyTask.run(TestThreadPool.java:37)
	at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:909)
	at java.lang.Thread.run(Thread.java:662)
java.lang.RuntimeException: task 2,executed by Thread[pool-1-thread-3,5,main]
	at com.gleasy.library.cloud.dfs.test.TestThreadPool$MyTask.run(TestThreadPool.java:37)
	at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:909)
	at java.lang.Thread.run(Thread.java:662)
poolSize:1
poolSize:1
poolSize:1
poolSize:1
task 5 executed by Thread[pool-1-thread-6,5,main]
poolSize:1
poolSize:1
poolSize:1
poolSize:1
poolSize:1
task 6 executed by Thread[pool-1-thread-6,5,main]
poolSize:1

 

从上面的结果看到,当使用execute提交任务的时候,其中4个任务抛了异常,导致后续任务全部都在1个线程里面执行了。等于是串行执行。这对程序的性能有极大的影响。本来是多线程执行,结果因为执行任务的时候其中某些任务抛异常,可用线程数就大量减少了。这个BUG同样极为隐晦。

解决方案如下,使用submit代替execute提交任务。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

public class TestThreadPool {
	private static final ExecutorService pool = Executors.newFixedThreadPool(5);

	public static void main(String[] args) throws InterruptedException {
		for (int i = 0; i < 10; i ) {
			pool.submit(new MyTask(i));
		}

		for (int i = 0; i < 40; i ) {
			System.out.println("poolSize:" ((ThreadPoolExecutor) pool).getPoolSize());
			Thread.sleep(1000L);
		}
		pool.shutdown();
	}

	static class MyTask implements Runnable {
		private int num;

		public MyTask(int index) {
			this.num = index;
		}

		public void run() {
			try {
				Thread.sleep(5000L);
				System.out.println("task " num " executed by " Thread.currentThread());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			if (num < 5) {
				throw new RuntimeException("task " num ",executed by " Thread.currentThread());
			}
		}
	}

}

 

问题解决。

poolSize:5
poolSize:5
poolSize:5
poolSize:5
poolSize:5
task 0 executed by Thread[pool-1-thread-1,5,main]
task 3 executed by Thread[pool-1-thread-4,5,main]
task 2 executed by Thread[pool-1-thread-3,5,main]
task 4 executed by Thread[pool-1-thread-5,5,main]
task 1 executed by Thread[pool-1-thread-2,5,main]
poolSize:5
poolSize:5
poolSize:5
poolSize:5
poolSize:5
task 5 executed by Thread[pool-1-thread-1,5,main]
task 6 executed by Thread[pool-1-thread-2,5,main]
task 9 executed by Thread[pool-1-thread-4,5,main]
task 7 executed by Thread[pool-1-thread-3,5,main]
task 8 executed by Thread[pool-1-thread-5,5,main]
poolSize:5
poolSize:5
poolSize:5

 

发表在 Java技术 | 留下评论

Gleasy分布式架构体系介绍

声明
本文为Gleasy原创文章,转载请指明引自Gleasy团队博客

正文
Gleasy作为专业的云服务平台和一站式云办公平台,分布式技术为支撑平台正常运作关键性技术。从商业利润和运维成本角度出发,千方百计榨干服务器的每一分性能很大程度上影响着Gleasy的商业价值,因此对性能的追求,成为Gleasy分布式架构体系中极为重要的考量指标;从用户角度,特别是作为Gleasy主要收入来源的企业用户的角度出发,保证业务处理的正确性和服务不中断(高可用性)是支撑用户信心的重要来源。高性能,高可用,正确性成为Gleasy分布式架构体系的关键技术因素。

Gleasy一直奉行着“深耕细作”,“磨刀不误砍柴功”这种朴实的思路进行开拓发展。从公司成立之初就开始思考分布式架构体系的长远趋势,也参考了大量国内外比较有影响力的互联网公司公布的一些技术方案,比如Facebook,Twitter,新浪,淘宝,甚至有不少方案已经开源。当时Gleasy面临着艰难的选择,要么拥抱开源,要么自主研发;如果拥抱开源,则Gleasy的首个开发周期可能会缩短1/3左右,但付出的代价是要深入吃透这些开源方案为我所用,必要时还得修改源代码,这后面的代价很难估计,另外还存在一项风险就是万一选用的开源方案在将来才发现某一些特性不满足,就得推倒重来;如果自主研发,则项目的开发周期会相应延长,而且似乎有重复造轮子的嫌疑。Gleasy在艰难的选择下最终决定部分使用开源,部分自主研发,一个基本原则是凡是不对业务流程产生重大影响的工具类全部采用开源,决定业务流程及软件流程的中间件则以自主研发为主。一直走到今天,Gleasy选用的重要开源工具主要包括有Zookeeper,Solr(Gleasy在它基础上研发了分布式索引平台),Openfire(Gleasy在它基础上研发了分布式的扩展插件),Redis(在它基础上研发分布式NOSQL数据库集群),Nginx,Haproxy,Keepalived,MySQL(在它基础上研发的ShardDB);而自主研发的分布式中间件则包括分布式文件系统(Cloudfs),分布式即时通讯(CloudIm),分布式消息队列(CloudMQ),分布式任务调度(CloudJob),分布式检索平台(CloudIndex),分布式NOSQL数据库集群(CloudRedis)。本文接下来介绍一下Gleasy自主研发的中间件,它们共同构成了Gleasy的分布式架构体系。


分布式架构逻辑图.

分布式文件系统(Cloudfs):反复研究HDFS,TFS,Gridfs(Mongodb),FastDFS基础上研发出来的分布式文件系统。存储架构与FastDFS相似,包括数据结点(Node),数据组(Group),分区(Region)三级;数据结点(Node)为最终物理存储结点,相同数据组(Group)的不同结点会进行实时的数据同步,即同组的结点数据最终一致; 相同分区(Region)内的文件会自动去重,即相同内容的文件在同一个分区(Region)只会有一份。Cloudfs通过使用消息队列(CloudMQ)进行同组结点间的数据序列同步,从而保证最终一致性,这一点跟FastDFS的binlog双向同步最为相似。Cloudfs使用Zookeeper来监测存储结点状态维护和变更,使用CloudRedis存储文件索引信息,使用Nginx作为文件下载服务器。性能优秀,单台Server线上监测得到的数据,文件上传的IOPS可以达到1200,上传速率达到25MBps,复制文件TPS达到9000以上,创建和删除文件的TPS达到30000以上。支持存储结点的动态加入和退出,支持线上扩容,数据多备份,结点动态负载均衡,最大理论可支持文件数量达千亿以上,支持结点数3000以上。

分布式即时通讯(CloudIm):承载着Gleasy所有推送服务的平台级中间件。基于Openfire,但除了保留其基本的连接保持和结点间通讯能力外,几乎进行了全新的改造。CloudIm提供了方便的API给第三方应用,开发者可以使用CloudIm的API轻松地实现消息即时推送(从而实现例如即时通讯,协同办公,即时提醒等实时功能),而不用考虑长连接保持,线路故障,服务器负载,用户状态变更通知等繁复的要求。CloudIm性能优秀,其提供的全部API的TPS都介于15000-35000之间。单台服务器线上实测可保持连接数超过7万,消息延迟低于50MS,集群可支持上千结点。支持结点动态加入或退出,支持线上扩容。

分布式消息队列(CloudMQ):一种分布式的消息队列(MQ)实现方案,设计原理参考了Apache的开源项目Kafka及淘宝的开源项目MetaMorphosis,承继了分布式及高性能高吞吐量的特性。CloudMQ实现简单,无Broker设计,可保持消息顺序,采用纯PULL加通知机制几乎避免了消费延迟,采用多分区机制保证提高系统的吞吐量,最大消息数量为数十亿级别,另外还支持延迟消息(比如5分钟之内发生100次更改事件,只推送一次最终结果,大幅减少重复消息)。生产消息TPS介于35000-45000之间,消费消息的TPS为40000左右。

分布式任务调度(CloudJob):将任务按特定规则(负载均衡,地域原则等)分配到多个结点执行的调度框架。支持Crontab标准的任务重复和定时策略,支持海量定时任务(千万级),保证任务处理的实时性和顺序性,支持实时查询任务状态或中止任务。任务调度吞吐量可达2万每秒。

分布式检索平台(CloudIndex):海量实时检索系统,承载着Gleasy全部的分词检索任务。主要采用Solr和CloudMQ实现,使用CloudMQ保证更新性能以及保证集群结点获得相同的更新序列,使用Solr实现分词及实时检索。支持索引分片(分片规则包括HASH法及数字区间法),自定义分词,结点负载均衡。索引读写延迟小于200MS,单一索引的数据规模可以达到上亿级别。

分布式NOSQL数据库集群(CloudRedis):基于Redis研发的数据库集群,兼容Redis的全部数据结构及大部分的命令集合。由客户端使用一致性HASH算法将请求按照KEY的HASH请求到集群内不同结点执行,使用binlog操作序列同步方式来保证不同服务结点的数据最终一致性;当服务结点变更时,客户端主动发现结点变更,重新计算HASH,集群内其它服务结点获知结点变更,保证binlog已经消费完毕的情况下才继续提供更新服务,从而保证结点变更情况下的数据一致性。性能极为优秀,非批量操作读写命令可达到10万每秒以上处理速度,超越了原生Redis,可支持十亿级别或更高数据存储。

以上介绍的分布式中间件,共同构成了Gleasy分布式体系的基础,在这些基础之上,Gleasy研发出了各种各样分布式应用,支撑起在线办公平台各种复杂业务场景。

联系作者
对本文有兴趣,有想法的,不解不满等。。欢迎赐教交流。
邮箱:xueke@gleasy.net
qq:99103622

best online casino

发表在 Java技术, 分布式技术, 数据库技术, 运维 | 留下评论

Gleasy的分布式即时通讯中间件CloudIM

声明
本文为Gleasy原创文章,转载请指明引自Gleasy团队博客

一。CloudIM是什么
CloudIM,字面意思是即时消息中间件。通俗一点可以理解为一套推送服务,就像APNS(苹果推送通知服务)。它主要以下三件事情:建立和保持长连接(SOCKET),维护及实时通知用户(好友之间,房间内部)状态变更(离线,在线,忙碌,PC在线,手机在线,平板在线等等),充当消息中转站–应用服务器通过它把消息实时推送给客户端。

二。产生的背景
Gleasy有众多需要实时通讯的场景,比如一说(即时通讯工具),消息中心,表格(实时协同办公)。这些场景对实时要求非常高,定时拉取(PULL)的方式由于延迟问题往往不能很好地满足这些业务场景,很有必要维护SOCKET连接。
最先选取的方案是基于XMPP的Openfire,在它基础上轻松地扩展出了满足我们业务场景的上层协议。不过用一段时间就慢慢暴露出致命的问题,首先是单点问题,Openfire没有集群,结点一挂,服务立刻终止;其次是使用MYSQL作存储,性能真心一般,在线人数过万就基本上跑不动了;再次是业务级问题,GLEASY的在线状态和房间,跟XMPP协议标准有很大区别,最明显的区别是XMPP协议的CHAT ROOM的在线状态是进入了房间内的房间内在线,而我们需要的是统一的在线状态,不管进没进房间。
研究过OPENFIRE的集群插件,一是要收费,二是性能确实也不咋地,加上始终无法满足我们对于在线状态的要求,果断放弃,准备自主研发。
从零开始,似乎有些浪费,毕竟OPENFIRE维护连接和结点间通讯的能力还是很不错的。保留Openfire的这两点能力,其它的全部被替换掉。我们对这个中间件的要求有几点:
1. 无单点故障,任何一个结点挂掉,只要资源充足,都不会对用户体验造成任何影响。
2. 负载均衡,所有结点都可以直接对外提供服务(根据配置高低分配连接数量),不采用主备方案。
3. 可以随时增加新的结点进来,无须停服务
4. 总用户数量支持到亿级别而数据库依然坚挺
5. 同时在线数量可以支持到百万级别以上

三。架构设计
1. 软件架构
Openfire作为连接保持服务器
Java作为插件开发语言
Zookeeper用于维护结点状态
redis用于存储用户信息,房间信息,好友关系

2. 主体架构图

四。部分业务场景流程
1. 结点启动(新加入)
a. 启动
b. 向ZK注册本结点信息(IP,域名,端口,最大可容纳连接数)

2. 连接建立
a. 向集群内任意一结点发送查询命令
b. 结点收到命令,查找集群内所有存活结点,找到使用率最低的那个,返回结点信息给客户端
c. 客户端根据收到的结点信息,取出域名及端口,发起连接请求
e. 目标结点收到连接请求,建立连接,保存连接信息,并将连接数加1
f. 客户端连接断开或者建立不成功,重复a-e步骤

3. 连接断开
a. 己连上的结点监测到连接断开信息
b. 更改断开用户的在线状态
c. 将新的状态推送给所有在线好友
d. 将新的状态推送给所有TA已经进入的ROOM
e. 删除连接信息,连接数减1

4. 结点挂掉
a. 主结点监测到结点挂掉事件
b. 将已经连接到该结点的所有连接信息清除
c. 将挂掉结点的在线人数置0

5. 发送消息
a. 应用服务器调用cloudim的API,将发送消息命令发送给其中一结点(随机)
b. 结点收到命令,找出当前用户的连接信息,调用openfire接口,将消息以XMPP协议推送至客户端

五。性能测试
以下是服务器端调用CLOUDIM的API的性能报告
创建用户 tps best online casino 17000-19000
更新用户 tps 25000-27000
删除用户 tps 27000-35000
获取用户 tps 25000-27000
创建双向好友 tps 17000-19000
删除双向好友 tps 18000-19000
创建房间 tps 5500-6500
销毁房间 tps 5500-6500
加入房间 tps 5500-7500
批量加入房间 tps n*2500(n为一次加入的用户数,用户数越多,tps越高)
退出房间 tps 6000-8000
进入房间 tps 8000-9000
离开持久房间 tps 8000-10000
离开临时房间 tps 2000左右
获取房间成员 25000-27000
获取在线成员 25000-27000

结合线上实际情况和测试结果,CloudIM在单台服务器环境表现相当优秀,结合动态集群,完全可以承担得起大规模并发实时系统的需求.

发表在 Java技术, 分布式技术 | 一条评论

线上诡异故障处理一例

声明
本文为Gleasy原创文章,转载请指明引自Gleasy团队博客

一。故障描述
Gleasy手机版上线之后,时不时就收到用户投诉说他的手机版访问不了,也不能使用手机浏览器打开gleasy网站。后来经过分析总结,发现这些终端都有以下特点:
1. 手机使用WIFI环境异常
2. 手机使用2G/3G网络正常
3. 在相同的WIFI网络下的PC终端正常

二。故障分析
1. 使用ping命令
ping www.gleasy.com 正常

2. 使用telnet
telnet www.gleasy.com 80 online casino 建立连接超时

3. 抓包
发现 客户端发出了 syn 之后,服务器一直没有任何回应,一直没有发 syn ack。

三。故障处理
参考了这篇博客:http://blog.sina.com.cn/s/blog_781b0c850100znjd.html

调整了内核参数,设置

net.ipv4.tcp_tw_recycle = 0

问题解决。

发表在 运维 | 留下评论

使用Haproxy,Keepalived,Tproxy实现高可用透明反向代理

声明
本文为Gleasy原创文章,转载请指明引自Gleasy团队博客

一。需求场景

具体需求如下:
4台Server,2台为Proxy Server,2台为Web Server,均为双网卡;
1个公网IP(183.129.228.91);
要求如下:
1. 2台Proxy Server反向代理2台WEB Server,作负载均衡
2. 2台Proxy Server为主备模式,公网IP在两台Server之间自动切换
3. Proxy Server作透明代理(web server日志中要记录真实的访问IP)

二。配置步骤
1. 安装软件(proxy server)
Gleasy有自己的Yum库,全部软件已经打成RPM包供安装;
其中haproxy打包时使用了下面的参数:

make TARGET=linux26 CPU=x86_64 USE_STATIC_PCRE=1 USE_LINUX_TPROXY=1
make install target=linux26

安装:

yum install keepalived
yum install haproxy

 

2. 配置keepalived(proxy server)
配置主:

vrrp_instance VI_3{
 interface eth1 #这里的eth1是内网(192.168.1.X网段)的网卡!!!
 state MASTER 
 priority 100 #从为99
 virtual_router_id 101#路由ID,可通过#tcpdump vrrp查看。
 garp_master_delay 1 #主从切换时间,单位为秒。

 advert_int 1 #检查间隔,默认1秒
 authentication {
 auth_type PASS
 auth_pass KJj23576hYgu23IP
 }
 virtual_ipaddress {
 192.168.1.1/32 dev eth1
 183.129.228.91/27 brd 183.129.228.95 dev eth0
 }
 virtual_routes {
 via 183.129.228.65 dev eth0
 }
}

 

配置备:

vrrp_instance VI_3{
 interface eth1 #这里的eth1是内网(192.168.1.X网段)的网卡!!!
 state BACKUP #
 priority 99 # 主为100
 virtual_router_id 101#路由ID,可通过#tcpdump vrrp查看。
 garp_master_delay 1 #主从切换时间,单位为秒。

 advert_int 1 #检查间隔,默认1秒
 authentication {
 auth_type PASS
 auth_pass KJj23576hYgu23IP
 }
 virtual_ipaddress {
 192.168.1.1/32 dev eth1
 183.129.228.91/27 brd 183.129.228.95 dev eth0
 }
 virtual_routes {
 via 183.129.228.65 dev eth0
 } 
}

 

3. 配置tproxy(proxy server)

#!/bin/sh
/sbin/iptables -I FORWARD -i eth -j ACCEPT 
/sbin/iptables -t nat -A POSTROUTING -o eth1 -j MASQUERADE
/sbin/iptables -t mangle -N DIVERT
/sbin/iptables -t mangle -A PREROUTING -p tcp -m socket -j casino online  DIVERT
/sbin/iptables -t mangle -A DIVERT -j MARK --set-mark 1
/sbin/iptables -t mangle -A DIVERT -j ACCEPT
 
/sbin/ip rule add fwmark 1 lookup 100
/sbin/ip route add local 0.0.0.0/0 dev lo table 100

echo 1 > /proc/sys/net/ipv4/conf/all/forwarding 
echo 1 > /proc/sys/net/ipv4/conf/all/send_redirects 
echo 1 > /proc/sys/net/ipv4/conf/eth0/send_redirects

 

4. 配置haproxy(proxy server)

global
 log 127.0.0.1 local3 err
 chroot /usr/local/haproxy
 maxconn 13000
 daemon
 nbproc 4 #设置并发进程
 pidfile /usr/local/haproxy/logs/haproxy.pid

defaults
 log global
 option dontlognull # 不记录空连接
 option redispatch
 timeout connect 50000
 timeout client 120000
 timeout server 120000
 maxconn 10000
 balance roundrobin # 设置服务器分配算法
 retries 5 # 重连次数 
 errorfile 400 /usr/local/haproxy/html/400.http
 errorfile 403 /usr/local/haproxy/html/403.http
 errorfile 408 /usr/local/haproxy/html/408.http
 errorfile 500 /usr/local/haproxy/html/500.http
 errorfile 502 /usr/local/haproxy/html/502.http
 errorfile 503 /usr/local/haproxy/html/503.http
 errorfile 504 /usr/local/haproxy/html/504.http


frontend proxy-80
 bind *:80
 mode http
 option httplog
 option dontlognull
 option forwardfor # This sets X-Forwarded-For
 option http_proxy
 maxconn 13000
 timeout client 50s
 timeout http-keep-alive 1s
 timeout http-request 10s
 default_backend proxy-80
 

#backend配置 
backend proxy-80
 mode http
 timeout server 120s
 timeout connect 50s
 option nolinger
 option http_proxy
 option forwardfor # This sets X-Forwarded-For
 option httplog
 option http-server-close
 cookie JSESSIONID prefix
 stats enable
 balance roundrobin
 source 0.0.0.0 usesrc clientip #透明代理!!!!!
 option httpchk GET /check.html
 server S 192.168.1.27:80 weight 3 cookie JSESS1 check inter 1500 rise 3 fall 3 
 server S 192.168.1.28:80 weight 3 cookie JSESS2 check inter 1500 rise 3 fall 3 

 

5. 配置Web server网关(web server)

ip route default via 192.168.1.1 dev eth0

特别说明一下,这里的192.168.1.1是一个VIP(虚拟IP,它的配置在上面keepalived配置中可以找到)。它会在两台proxy server间自动切换。

6. 注意事项:
HAPROXY所在的服务器网关必须配置为外网(ip route default必须为外网网关);
web服务器网关必须配置为HAPROXY所在服务器的IP地址;

发表在 运维 | 留下评论

Gleasy的分布式消息队列CloudMQ

声明
本文为Gleasy原创文章,转载请指明引自Gleasy团队博客

一。诞生背景
Gleasy子系统众多,子系统间的沟通也非常紧密(比如联系人系统,就需要向一说等提供底层数据),为降低系统复杂度,解耦势在必行;Gleasy的若干场景对性能要求很高,但对实时性要求不高,比如索引平台,消息中心,日志系统,每秒可能产生上万条甚至十几万条记录,直接操作DB无疑等于自残;Gleasy的一些基础信息,比如用户名的修改,往往需要及时地通知到多个子系统,便要求有一种广播机制来传达这些信息。解耦,高速缓冲,消息广播,所有这些正是消息队列(MQ)最为擅长解决的问题,因此使用MQ势成必然。

最初选用的是MQ是一个基于全PUSH模型的轻量级MQ Postman(借鉴于某互联网公司内部未开源,Postman为我们起的名字),基于MYSQL开发,使用quartz轮循推送,HTTP协议(JSON方式)传参。使用一段时间,发现问题极其严重,首先是延迟太厉害,由于使用轮循机制和HTTP协议,性能不高,轮循间隔也不敢配太短(通常都是配置为1-5S),很多场景下这种延迟完全接受不了(比如我们的实时索引平台);其次是不支持分布式,同一个主题(topic)订阅者太多时性能很差,加上使用MYSQL库表方式实现悲观锁,经常造成死锁;还有就是数据量问题了,使用MYSQL集中存储,当未消费的消息数量达到千万级别,整个系统开始变得极不稳定,再稍加并发压力,延迟就急剧上升。最终不得不放弃掉。

然后在开源世界苦苦筛选。ActiveMQ?不行,不支持分布式,TPS也1000左右,远达不到要求;ZeroMQ?太底层了,只是一个通信框架,放弃。然后在Apache的网站上看到了Kafka,一个开源的分布式消息队列,基于Zookeeper实现,性能超凡(TPS可达几万),支持分布式消费,消息顺序,海量数据,果断拿来实验,实验结果确实非常理想,美中不足是采用Scala语言开发,维护起来有些吃力,将来想改一改源码也较为困难;后来淘宝开源Meta系统(Kafka的JAVA版),拿过来对比,发现各方面能力都和Kafka持平,而且加入了一些集群和事务的能力,着实做了不少工作,根据官方公布的性能测试数据,大概可以达到2-4W的TPS,在Gleasy服务器上测试,也证实了这个结果。

虽然Meta系统在大部分情况下都能满足我们的需求,Gleasy最终还是选择了自主研发CloudMQ。原因很简单,MQ为最基础的中间件之一,影响全局。在弄清楚了Kalfa和Meta的原理之上,朝着我们关注的指标上狠狠努力,深度定制,实现自己完全可控的核心中间件,值得一做; 加上我们还有不少需求是Kalfa和Meta都无法满足(比如延迟消息)。这些因素共同促成了CloudMQ的诞生。

二。架构和原理
1. 软件架构
采用JAVA实现
基于Zookeeper和Redis

2. 逻辑架构
某个topic的多个生产者和同一个消费者(同一个ID的消费者称为同一个消费者,它们可分布部署多结点)逻辑架构图如下:

3. 物理架构

4. 工作原理
CloudMQ的工作原理和Kafka/Meta大体相似,都是采用分区机制。

每个topic都可以有多个分区;
topic的消息队列分散于各个分区的redis中;
生产者生产的消息会根据分区映射规则进入到某个分区中;
一个消费者会消费N个分区的消息;
一个分区同时只会被一个消费者的实例消费;
举例如下:
topic名称: topic1
消费者名称: consumer1
生产者结点数量:1
映射规则:根据消息ID对分区数量取模
消费者结点数量:从1 逐渐增加到 3
工作过程如下:
a. 生产者生产10条消息,ID分别为1至10.
b. 根据映射规则,这10条消息分别进入5个分区,其中1,6消息进入1号分区;2,7进入2号分区;3,8进入3号分区;4,9进入4号分区;5,10进入5号分区。
c. 消费者结点A启动,由于只有它一个结点,因此它负责所有分区即1到10号分区都由A消费
d. 消费者结点B启动,消费者A和B都检测到当前有2个结点在消费,于是根据规则进行分配。A结点负责1-5号分区,B结点负责6-10号分区。
e. 消费结点C启动,消费者A,B,C检测到当前有3个结点在消费,重新进行分配。A结点负责1,2,3,10号,B负责4,5,6号,C负责7,8,9号

online casino wp-image-358″ />

三。关键特性
支持订阅pub-sub和消息队列queue两种工作模式.
支持消息顺序(选用功能,可保证消费顺序和生产顺序完全一致)
消息分区生成和处理,同一分区严格保证顺序,不同分区可并发处理
支持消费者集群,从而实现任务分布式处理,且每个分区只被唯一一个结点处理
无BROKER设计,避免单点故障
队列数据分区分片存储,最佳工作状态下,消息数量可达数十亿级别.
支持延迟消息队列(区别于现有MQ的重要特性)
纯pull模式 通知机制,实际消息延迟小于10MS

四。性能指标
生产者,非延迟(tps 45000左右)
target/benchtest/bin/TestMqProducer /data/config/util/config_0_11.properties normal 500 50000
线程总时间:782889;平均:15.65778
实际总时间:1831; 平均:0.02662

生产者,批量生产(批量5-100条每次) (tps = 50000 相当给力)
target/benchtest/bin/TestMqProducer /data/config/util/config_0_11.properties mnormal 100 5000
线程总时间:477496;平均:1.761016124035582
实际总时间:6102; 平均:0.022504314986649357

生产者,延迟(14285 - 27777 / s)
全部都是非首次生成(tps = 27777)
[root@DB1 bin]# ./TestMqProducer /data/config/util/config_0_11.properties delay 500 50000
线程总时间:699032;平均:13.98064
实际总时间:1823; 平均:0.03646

全部都是首次生成 500线程 (tps = 14285)
[root@DB1 bin]# ./TestMqProducer /data/config/util/config_0_11.properties delay 500 50000
线程总时间:893383;平均:17.86766
实际总时间:3706; 平均:0.07412

消费者
50线程 消费50000条数据(tps = 35000 )
49499:t=2387:avg=0.03022319642821067 ms

结论:CloudMQ生产消息TPS可达4W以上,消费消息TPS达3.5W以上;延迟特性TPS近于1.4W至2.7W之间。

五。实战小结
CloudMQ上线1年,目前广泛应用于一盘,一说,消息中心,索引平台等关键业务系统,线上运行情况理想,完美实现预期目标。在性能和稳定性方面的表现也令人满意,扛得住严酷的使用环境。而且部署简单,无额外服务器部署成本,是Gleasy理想的消息中间件。

发表在 Java技术, 分布式技术, 运维 | 2 条评论

Gleasy内核之在线应用进程间通信技术

声明
本文为Gleasy原创文章,转载请指明引自Gleasy团队博客

一。它有多重要
Gleasy在线应用进程间通信技术为Gleasy最核心技术之一,目前己申批国家发明专利和国际发明专利。它于Gleasy来讲到底有多重要呢,可以简单地说,Gleasy平台内一切看得到的功能都归它管。它就像PC机的主板,负责将各个子系统联系起来,沟通起来。

二。它解决什么问题
在线应用进程间通信技术涵盖两个主要概念,一个是在线应用进程,一个是通信技术。通过这两个概念的字面意思也反映出来,它解决的是在线应用的管理和沟通问题。Gleasy上面运行的一盘,一信,一说,项目管理,美图秀秀等等,全部都是一款款的独立应用。当用户使用Gleasy的时候,这些独立应用如何打开,资源如何加载,如何管理(比如关闭),如何监控,如何调用(比如用美图秀秀打开一盘文件)等等,都是该技术试图解决的问题。

如果你理解Android原理,考虑一个场景:美图秀秀调用了拍照功能得到一幅照片,然后自己编辑,编辑完之后使用新浪微博发出去。这个场景,正是Gleasy进程间通信技术解决的问题。

三。它如何工作

应用进程:应用在Gleasy运行时的存在形态,它可以是一个URL(比如美图秀秀),也可以是JS的类和DOM(比如一盘,一说)。当用户安装某个应用后,就会在注册表(reg.js)中生成注册项,注册项包含了应用的URL或者用于启动的JS类(MAIN函数)。点击图标启动时,GLEASY内核的进程管理器(ProcessManager)会读取注册项,动态加载应用运行时所需的资源(JS等),运行MAIN函数生成应用进程(Process)实例。

进程间通信:应用通过调用进程管理器提供的API来声明自己可以提供的服务(这里的服务是事件名称,比如选择文件服务的事件名称为gcd.file.selector)。调用者通过调用进程管理器的API来调用服务,调用的方法为sendMessageToProcess({pname:”一盘”,eventId:”gcd.file.selector”,messageBody:params});这个API的意思是,调用应用名一盘的应用,事件为”gcd.file.selector”,参数为params。进程管理器会寻找“一盘”这个应用进程,看是否存在,如果不存在,会尝试启动它,启动成功后把事件投递到该进程的消息队列中,一盘应用进程会定时消费在自己消息队列中的消息。从而实现应用间通信。

四。对应用的要求
应用可以完全不关心进程间通信的原理及过程,甚至可以不知道有应用进程的存在。应用开发者只需要按照Gleasy的规范实现若干简单的接口便可提供的能力供他人使用,而不关心调用来源及调用方式。
当需要调用别的应用提供的服务时,也只需要调用Gleasy提供的API将消息发送给目标进程,至于如何发送,如何启动目标应用,都是由进程管理器来完成,对调用方来说只关心一个sendMessageToProcess即可,非常简单方便。

五。一些技术依赖
由于Gleasy上面的应用可能是运行在其它网站域名之下,不可避免存在跨域问题。
解决跨域问题主要有以下几种途径:
1. 使用Access-Control-Allow-Origin头解决AJAX跨域调用(推荐)

add_header best online casino Access-Control-Allow-Origin *;
add_header Access-Control-Allow-Methods "GET,POST,OPTIONS";
add_header Access-Control-Allow-Headers Content-Type;
add_header Access-Control-Max-Age 86400;

这种方法最为简单方便,无须做任何额外处理便可跨域调用AJAX。
在IE8以上,Chrome,Firefox,Safari的大多数版本都适用。

2. 使用window.postMessage来解决Iframe和主窗口之间的跨域通信(推荐)
这个方法性能好,简单易用。在IE7以上,Chrome,Firefox,Safari的大多数版本都适用。

3. 使用Flash解决Ajax调用的跨域问题(不推荐)
该方法适用于所有安装了flash插件浏览器,缺点是不支持AJAX的一些特性,而且调试极不便利。不建议使用

4. 使用iframe的name和url的hash属性传递参数,来解决Iframe和主窗口之间的跨域通信(不推荐)
该方法性能很差,仅适用于IE6这些没办法使用postMessage的情况。

Gleasy会根据浏览器类型和版本自动决定使用1,2,3,4哪种技术。保证100%支持跨域。

六。技术之外
在线应用进程间通信技术带来的不单是一种在线平台方案,它还有其它方面的意义。这几年吵得最火的概念之一就是OPENAPI(开放平台),开放平台大多使用OAUTH(OR OAUTH2)进行,一个明显的特征是存在浏览器的跳转和用户的授权。除此之外,假如要A B C D四款应用要互相对接,就要进行4*3=12次研发才完成两两互相对接(因为每个应用每个接口的URL都不同),这就是所谓点到点的开放接口。另外还有一个严格的限制,如果A想同时使用B和C的服务,甚至有些依赖关系,那就变成几乎不可能了。

Gleasy平台上应用之间的通信就简单得多了,没有浏览器的跳转,无须用户反复授权,同时调用N个应用提供的服务,或者串行调用都极为方便,不存在任何限制。假如A B C D三款应用要互相对接,大家只需要跟Gleasy平台打交道就可以了,只须4次研发,另外大家只需要记住其它应用的名称和参数,完全不用理会对方的URL或者接口,即使对方的URL发生了根本变化也不会影响通信。没有浏览器的跳转,无须用户反复授权,由此便可大大节省沟通和研发成本,并且可以衍生出交互体验上更棒的产品,而不用受限于浏览器技术。

在Gleasy,使用拍照应用拍照,保存到一盘,使用美图秀秀打开一盘的图片编辑,保存后使用一信的邮件发出去。这就是在线应用进程间通信的魅力!@

发表在 前端技术 | 留下评论

Cloudfs的运维工具

一. 安装和配置
1.安装
yum clean all
yum install cloudfs
2.配置
>cd /usr/local/cloudfs
>vi etc/dfs.conf

#服务器监听端口
port=8123

#文件下载服务端口
httpPort=8124

#文件系统组名
group=group1

#本服务器结点的唯一标志(全局不能重复)
serverid=11

#对外服务的地址(可以为IP或者域名)
domain=192.168.0.11

#数据目录位置
dataDir=/cloudfs

#本服务器最大容量(字节)
#default:50G
capacity=53687091200

#性能优化参数
#默认为128K
bufferSize=131072
#默认为1M
maxBufferSize=1048576

#是否调试
debug=false

#Config服务器配置信息
config.server.ip=192.168.0.11
config.server.port.redis=6379

#log4j配置文件路径,默认和本文件在同一目录下,文件名为log4j.properties
#log4j=/data/gleasy/dfs/conf/log4j.properties

c. 启动
>cd /usr/local/cloudfs
>bin/dfs-server start

d. 关闭
>bin/dfs-server stop

e. 测试
>cd /usr/local/cloudfs
>bin/dfs-cli put /root/test.txt

二. 客户端集成
a. 目前仅提供了JAVA客户端,SVN地址为: http://svn.dev.gleasy.com/svn/gleasy.com/library/cloud-dfs/trunk
maven配置为:

 <dependency>
 <groupId>com.gleasy.library</groupId>
 <artifactId>cloud-dfs</artifactId>
 <version>1.0</version>
 </dependency>

 

b. API列表

i. DfsClient.createFile()
创建空文件

ii. DfsClient.uploadFile(String storageid, InputStream stream, String hash, long casino filesize)
上传文件内容
storageid: 文件ID
stream: 文件流
hash: 文件的HASH
filesize: 文件大小
hash和filesize必须同时存在,要么都有,要么都没有。

iii.DfsClient.opyFile(String storageid)
复制文件,返回新文件ID

iv. DfsClient.deleteFile(String storageid)
删除文件

v. DfsClient.getFileInputStream(String storageid)
获取文件流用于保存

vi.DfsClient.getFilesize(String storageid)
选用。获取文件已经上传的大小(用于断点续传)

三. 压力测试工具
>bin/dfs-benchmark

用法: dfs-benchmark options
-t 测试类型(create,upload,copy)
-p 持续时间(单位s)
-c 客户端数量
-s 文件大小

发表在 分布式技术, 运维 | 留下评论

Gleasy的分布式文件系统Cloudfs

声明
本文为Gleasy原创文章,转载请指明引自Gleasy团队博客

一。轮子,又见轮子
分布式文件系统确实很多,开源的也不少,像FASTDFS,TFS,MogileFS,HDFS,MongoDB(Gridfs),而且个个响当当,相当优秀。Gleasy在创业阶段坚持自己开发分布式文件系统,乍看上去确实有重复造轮子的嫌疑。可事实上,Gleasy选择此举属无奈。最初选用的是MongoDB(Gridfs),图的是它超级方便,用一段时间发现支持不了断点续传(由于分片保存的时候要对所有内容生成HASH),而且IOPS极低,压力一上下CPU就狂升100%,果断抛弃。后来换成FASTDFS,确实可以支持断点续传了,用一段时间为节省空间想引入文件去重能力(整个文件系统相同内容文件只留一份),结果令人失望,加上使用过程中还遇到1S只能建一个空文件的限制,实在没办法满足Gleasy这种协同读写环境,想改代码又觉得维护成本太高,忍痛弃之。开始研究TFS,HDFS,MogileFS,经过大量实验,对比,总结,Gleasy个个都是完美主义者,愣是发现没有一个极为适合我们。没办法了,关上门开始造车。。这就有了Gleasy的Cloudfs 1.0诞生。

二。没有最好,只有最适合
Cloudfs重点关注的指标是高性能,去重,容灾,线上扩容,最终一致,简单易用。经过长时间的优化,它已经开始具备这些能力。Cloudfs没有像大多数FS一样分块存储,虽然分块存储的好处非常明显,但高性能和加密的要求令我们对分块SAY NO。Cloudfs采用JAVA开发,也许在理论上,它永远超越不了C语言开发的TFS和FastDFS,但是从实际测试的对比结果来看,似乎又不一定。在Gleasy这个特定的环境,Cloudfs无疑是最优秀最合适的,虽然,换了别的地方,它可能就变得平庸。我们始终相信没有最好,只有最适合。

三。架构设计
1. 物理架构
数据结点(DataNode):真正用来存储数据的结点
组(Group): 包括若干数据结点,同组内的所有数据结点内容保持一致(实时互相同步)
分区(Region): 包括若干组,同一个分区内的文件会自动去重(相同内容的文件只会存在一份),每个分区最多可以容纳10亿个文件。
注意:一个分布式文件系统可以包括若干分区。

2. 软件架构
单个运行的Cloudfs只包含一个DataNode服务器(DataNode Server)和一台Nginx服务器。DataNode Server提供文件的上传,复制,删除等功能,Nginx服务器提供流式下载功能。
DataNode Server采用JAVA开发,采用MINA框架,使用java NIO FileChannel进行文件存取操作。
使用Zookeeper进行结点发现和管理
使用Redis存储文件信息和索引信息
使用CloudMQ进行同组内基于操作日志的数据同步

四。运行特性
a. 数据多备(一个组可以有多台服务器结点,组内每个结点的文件会自动同步,文件多结点互备)
b. 负载均衡(对某一个组的访问会平均地分配到各个结点,某个结点挂掉,不影响整个系统运作)
c. 动态扩容(通过增加新的组进行容量扩容,通过增加组内结点可以做到负载扩容,扩容过程不需要关闭服务)
d. 自动去重(在同一分区内相同内容的文件只会只存在一份,极大地节省空间和省去大量文件传输时间)
e. 高性能(IOPS达到1200以上,创建文件TPS达4W以上,复制文件TPS达9000以上,删除文件达2W以上,上传下载速率可达25MBPS)

五。性能测试报告

cloudfs-创建空文件
tps 46k
[root@DB1 cloudfs]# best online casino bin/dfs-benchmark -t create -p 60 -c 1000 -s 1024
成功:2816889
失败:0
命中缓存:0
总时间:60014662
最长用时:606
最短用时:0
平均用时:21.30529885984148
max tps:108389
avg tps:46178

cloudfs--创建并上传文件(全部不命中)
[root@gleasy cloudfs]# bin/dfs-benchmark -t upload -p 30 -c 50 -s 1024
成功:29782
失败:0
命中缓存:0
总时间:1728787
最长用时:5997
最短用时:0
平均用时:58.04804915720905
max tps:1700
avg tps:960

创建文件并上传---全部命中缓存
bin/dfs-benchmark -t cache -p 60 -c 200 -s 1024
成功:353676
失败:0
命中缓存:353676
总时间:12007269
最长用时:574
最短用时:0
平均用时:33.94991178366641
max tps:6861
avg tps:5797
清除测试数据(delete File) tps:19244.828832083822

复制文件
500 clients(fast redis)
bin/dfs-benchmark -t copy -p 60 -c 500 -s 1024
成功:507496
失败:0
命中缓存:0
总时间:30016590
最长用时:258
最短用时:0
平均用时:59.14645632674937
max tps:10004
avg tps:8319
清除测试数据(delete File) tps:18875.823589899835

删除文件
清除测试数据(delete File) tps:18875.823589899835

五。去重效果分析
采用Cloudfs(在10亿文件内只有一个分区)后,相同内容的文件只会存储一份。通过查看线上数据,线上运行6个月,计量容量600G,实际物理容量180G,节省420G,节省比例70%。可见Gleasy系统内,文件重复比例非常高,这与Gleasy企业客户为主的用户群体使用习惯很有关系。
可以预见,随着文件系统基数增大,节省的比例会继续增大,目标是可以做到1:10,即用1G的空间可以完美达到10G的使用量,使用该技术对用户体验没有任何损害,反而会带来性能质的提升。

发表在 Java技术, 分布式技术, 运维 | 2 条评论

Gleasy表格实现介绍(一)

1.表格需要具备功能

a.行列数据存储、处理(排序、筛选、单元格格式处理等)

b.公式计算

c.图表

d.其它辅助操作(复制粘贴、格式刷、undo/redo、行列拖动)

e.导入导出excel(深入人心的MS系列,不得不重视)

2.Gleasy表格功能组成

a.行列数据存储、处理(排序、筛选、单元格格式处理等)

b.公式计算

c.列属性(下拉菜单、多选框、优先级、星标等)

d.层级树

e.其它辅助操作(复制粘贴、格式刷、undo/redo、行列拖动隐藏)

f.导入导出Excel

3.使用技术

Java/mongo/redis/Javascript/JQuery/css/html/Jison, cloudim(Gleasy自主开发的即时通讯中间件)

公式计算引擎即基于Jison编写的表达式解析引擎

比如:部分四则运算、比较符解析

4.表格实现之单元格属性

任何前端的应用实现都基于HTML/CSS语义,同样地,单元格所需要的属性,比如字体、字形、背景色等,通过class和style的方式设置。如下图:粗体、下划线通过class来设置,背景色、字体颜色通过style设置。

best online casino alt=”图片5″ src=”http://rdc.gleasy.com/wp-content/uploads/2013/07/图片5.png” width=”826″ height=”93″ />

 

5.表格实现之undo/redo

一般单一文本类的undo/redo,都是通过不断保存整个文本的快照来实现,但表格应用需要涵盖所有的操作,而不仅仅是单元格内容的undo/redo, 以隐藏列操作为例,通过将当前命令、参数保存起来(数组中),隐藏列的undo是显示列(showColumn),隐藏列的redo也是隐藏列(hideColumn),参数都是列的index

在执行undo/redo时,通过设置数组的偏移来取得相应的操作指令,然后把操作指令传给doAction

doAction再将指令进行分发以完成操作。

发表在 Java技术, 分布式技术, 前端技术, 数据库技术 | 留下评论

一些压力测试结果(Mysql,Zookeeper,Redis,Mongodb)

声明
本文为Gleasy原创文章,转载请指明引自Gleasy团队博客

1. Redis(使用fastredisclient)

redis-shard 10连接
cpu 7-8% 9-12%
root@gleasy cloudredis]# bin/redis-benchmark -t get -h 192.168.0.11 -p 6680 -d 15 -l 60 -c 200 -b shard
成功:5740491
失败:0
总时间:11999368
最长用时:42
最短用时:0
平均用时:2.090303425264494
min tps:0
max tps:131439
avg tps:94106

cpu 9-12%
[root@gleasy cloudredis]# bin/redis-benchmark -t set -h 192.168.0.11 -p 6680 -d 15 -l 60 -c 200 -b shard
成功:5730516
失败:0
总时间:11999536
最长用时:205
最短用时:0
平均用时:2.093971293335539
min tps:0
max tps:109213
avg tps:93942

2. Mysql

Mysql 插入 1000线程 7000/s
target/benchtest/bin/TestMysql insert 1000 1000000 0
线程总时间:131320984;平均:131.320984
实际总时间:134504; 平均:0.134504

Mysql ibatis 插入 1000线程 5000/s
target/benchtest/bin/TestMysql insert 1000 1000000 0
线程总时间:131320984;平均:131.320984
实际总时间:134504; 平均:0.204504

Mysql 查询 1000线程 7000/s
target/benchtest/bin/TestMysql query 1000 200000 600000
线程总时间:27869248;平均:139.34624
实际总时间:29117; 平均:0.145585

Mysql ibatis 查询 1000线程 5000/s
target/benchtest/bin/TestMysql query 1000 200000 600000
线程总时间:27869248;平均:139.34624
实际总时间:29117; 平均:0.195585

Mysql 批量插入 500线程(50000/s)
target/benchtest/bin/TestMysql minsert 500 10000*100
线程总时间:10759531;平均:10.759531
实际总时间:22256; 平均:0.022256

3. Zookeeper

单结点:
set: tps 7500
get: tps 8700
del: tps 8400

4. Mongodb

写操作
200(线程数) 50000(记录数) add(操作) nbso online casino reviews writeConcern=normal
线程总时间:304881 最长用时:1928/250=7.712 最短用时:562/250=2.248 平均:6.09762
实际总时间:1933 平均: 1933/50000=0.03866 tps:25866.52871184687
200 50000 add writeConcern=safe (为保证数据正确,目前采用该方式)
线程总时间:1660848 最长用时:8580/250=34.32 最短用时:7401/250=29.604 平均:33.21696
实际总时间:8586 平均: 8586/50000=0.17172 tps:5823.433496389471

500 1000000 add writeConcern=normal
线程总时间:43969426 最长用时:108885/2000=54.4425 最短用时:61483/2000=30.7415 平均:43.969426
实际总时间:109016 平均: 109016/1000000=0.109016 tps:9172.965436266235
500 1000000 add writeConcern=safe
线程总时间:63972303 最长用时:129511/2000=64.7555 最短用时:122266/2000=61.133 平均:63.972303
实际总时间:129521 平均: 129521/1000000=0.129521 tps:7720.755707568657

1000 50000 add writeConcern=normal
线程总时间:2956438 最长用时:4276/50=85.52 最短用时:338/50=6.76 平均:59.12876
实际总时间:4303 平均: 4303/50000=0.08606 tps:11619.800139437602
1000 50000 add writeConcern=safe
线程总时间:7937768 最长用时:9196/50=183.92 最短用时:5995/50=119.9 平均:158.93826
实际总时间:9208 平均: 9208/50000=0.18416 tps:5430.060816681147

读操作
200 1000000 read
线程总时间:275106 最长用时:2116/5000=0.4232 最短用时:199/5000=0.0398 平均:0.275106
实际总时间:2137 平均: 2137/1000000=0.002137 tps:467945.7182966776

500 1000000 read
线程总时间:2205097 最长用时:5552/2000=2.776 最短用时:225/2000=0.1125 平均:2.205097
实际总时间:5576 平均: 5576/1000000=0.005576 tps:179340.0286944046

5.总结
在相同的硬件环境下,笔者测试的结果,redis在读和写性能都达到接近100K,性能表现最为优秀;MongoDB读写性能严重不均衡,读可以达到100K以上,写却只有5-6K,相差15倍之巨;Zookeeper在结点情况下,TPS接近8K,性能表现不俗,但不太适合用于关键性能场合;Mysql在读写性能大概介于5K-10K之间,批量操作性能优秀。
性能仅仅是衡量数据库优劣的其中一项指标,在具体的业务场景下,需要综合选取最优秀的存储方案或方案的组合,以达到最优设计。

发表在 数据库技术, 运维 | 留下评论

两种分布式锁实现方案(二)

声明
本文为Gleasy原创文章,转载请指明引自Gleasy团队博客
第一节:两种分布式锁实现方案(一)
第二节:两种分布式锁实现方案(二)

四。方案2,基于redis的分布式锁

/**
*分布式锁工厂类
*/
public class RedisLockUtil {
	private static final Logger logger = Logger.getLogger(RedisLockUtil.class);
	private static Object schemeLock = new Object();
	private static Map<String,RedisLockUtil> instances = new ConcurrentHashMap();
	public static RedisLockUtil getInstance(String schema){
		RedisLockUtil u = instances.get(schema);
		if(u==null){
			synchronized(schemeLock){
				u = instances.get(schema);
				if(u == null){
					LockObserver lo = new LockObserver(schema);		
					u = new RedisLockUtil(schema,lo);
					instances.put(schema, u);
				}
			}
		}
		return u;
	}

	private Object mutexLock = new Object();
	private Map<String,Object> mutexLockMap = new ConcurrentHashMap();
	private Map<String,RedisReentrantLock> cache = new ConcurrentHashMap<String,RedisReentrantLock>();
	private DelayQueue<RedisReentrantLock> dq = new DelayQueue<RedisReentrantLock>();
	private AbstractLockObserver lo;
	public RedisLockUtil(String schema, AbstractLockObserver lo){	
		Thread th = new Thread(lo);
		th.setDaemon(false);
		th.setName("Lock Observer:" schema);
		th.start();
		clearUselessLocks(schema);
		this.lo = lo;
	}

	public void clearUselessLocks(String schema){
		Thread th = new Thread(new Runnable(){
			@Override
			public void run() {
				while(!SystemExitListener.isOver()){
					try {
						RedisReentrantLock t = dq.take();						
						if(t.clear()){
							String key = t.getKey();
							synchronized(getMutex(key)){
								cache.remove(key);
							}
						}
						t.resetCleartime();
					} catch (InterruptedException e) {
					}
				}
			}

		});
		th.setDaemon(true);
		th.setName("Lock cleaner:" schema);
		th.start();
	}

	private Object getMutex(String key){
		Object mx = mutexLockMap.get(key);
		if(mx == null){
			synchronized(mutexLock){
				mx = mutexLockMap.get(key);
				if(mx==null){
					mx = new Object();
					mutexLockMap.put(key,mx);
				}
			}
		}
		return mx;
	}

	private RedisReentrantLock getLock(String key,boolean addref){
		RedisReentrantLock lock = cache.get(key);
		if(lock == null){
			synchronized(getMutex(key)){
				lock = cache.get(key);
				if(lock == null){
					lock = new RedisReentrantLock(key,lo);
					cache.put(key, lock);
				}
			}
		}
		if(addref){
			if(!lock.incRef()){
				synchronized(getMutex(key)){
					lock = cache.get(key);
					if(!lock.incRef()){
						lock = new RedisReentrantLock(key,lo);
						cache.put(key, lock);
					}
				}
			}
		}
		return lock;
	}

	public void reset(){
		for(String s : cache.keySet()){
			getLock(s,false).unlock();
		}
	}

	/**
	 * 尝试加锁
	 * 如果当前线程已经拥有该锁的话,直接返回,表示不用再次加锁,此时不应该再调用unlock进行解锁
	 * 
	 * @param key
	 * @return
	 * @throws Exception 
	 * @throws InterruptedException
	 * @throws KeeperException
	 */
	public LockStat lock(String key) {
		return lock(key,-1);
	}

	public LockStat lock(String key,int timeout) {
		RedisReentrantLock ll = getLock(key,true);
		ll.incRef();
		try{
			if(ll.isOwner(false)){
				ll.descrRef();
				return LockStat.NONEED;
			}
			if(ll.lock(timeout)){
				return LockStat.SUCCESS;
			}else{
				ll.descrRef();
				if(ll.setCleartime()){
					dq.put(ll);
				}
				return null;
			}
		}catch(LockNotExistsException e){
			ll.descrRef();
			return lock(key,timeout);
		}catch(RuntimeException e){
			ll.descrRef();
			throw e;
		}
	}

	public void unlock(String key,LockStat stat) {
		unlock(key,stat,false);
	}

	public void unlock(String key,LockStat stat,boolean keepalive){
		if(stat == null) return;
		if(LockStat.SUCCESS.equals(stat)){
			RedisReentrantLock lock = getLock(key,false);
			boolean candestroy = lock.unlock();
			if(candestroy && !keepalive){
				if(lock.setCleartime()){
					dq.put(lock);
				}
			}
		}
	}

	public static enum LockStat{
		NONEED,
		SUCCESS
	}
}

 

/**
*分布式锁本地代理类
*/
public class RedisReentrantLock implements Delayed{
	private static final Logger logger = Logger.getLogger(RedisReentrantLock.class);
 private ReentrantLock reentrantLock = new ReentrantLock();

 private RedisLock redisLock;
 private long timeout = 3*60;
 private CountDownLatch lockcount = new CountDownLatch(1);

 private String key;
 private AbstractLockObserver observer;

 private int ref = 0;
 private Object refLock = new Object();
 private boolean destroyed = false;

 private long cleartime = -1;

 public RedisReentrantLock(String key,AbstractLockObserver observer) {	
 	this.key = key;
 	this.observer = observer;
 	initWriteLock();
 }

	public boolean isDestroyed() {
		return destroyed;
	}

	private synchronized void initWriteLock(){
		redisLock = new RedisLock(key,new LockListener(){
			@Override
			public void lockAcquired() {
				lockcount.countDown();
			}
			@Override
			public long getExpire() {
				return 0;
			}

			@Override
			public void lockError() {
				/*synchronized(mutex){
					mutex.notify();
				}*/
				lockcount.countDown();
			}
 	},observer);
	}

	public boolean incRef(){
		synchronized(refLock){
			if(destroyed) return false;
 		ref ;
 	}
		return true;
	}

	public void descrRef(){
		synchronized(refLock){
 		ref --;
 	}
	}

	public boolean clear() {
		if(destroyed) return true;
		synchronized(refLock){
 		if(ref > 0){
 			return false;
 		}
 		destroyed = true;
 		redisLock.clear();
 		redisLock = null;
 		return true;
 	}
	}

 public boolean lock(long timeout) throws LockNotExistsException{
 	if(timeout <= 0) timeout = this.timeout;
 	//incRef();
 reentrantLock.lock();//多线程竞争时,先拿到第一层锁
 if(redisLock == null){
 	reentrantLock.unlock();
 	//descrRef();
 	throw new LockNotExistsException();
 }
 try{
 		lockcount = new CountDownLatch(1);
	 	boolean res = redisLock.trylock(timeout);
	 	if(!res){	 
	 		lockcount.await(timeout, TimeUnit.SECONDS);
					//mutex.wait(timeout*1000);
	 		if(!redisLock.doExpire()){
	 			reentrantLock.unlock();
						return false;
					}
	 	}
 	return true;
 }catch(InterruptedException e){
 	reentrantLock.unlock();
 	return false;
 }
 }

 public boolean lock() throws LockNotExistsException {
 	return lock(timeout);
 }

 public boolean unlock(){ 	 
 	if(!isOwner(true)) {
 		try{
 			throw new RuntimeException("big ================================================ error.key:" key);
 		}catch(Exception e){
 			logger.error("err:" e,e);
 		}
 		return false;
 	}
 try{
 	redisLock.unlock();
 	reentrantLock.unlock();//多线程竞争时,释放最外层锁
 }catch(RuntimeException e){
 	reentrantLock.unlock();//多线程竞争时,释放最外层锁
 	throw e;
 }finally{
 	descrRef();
 }
 return canDestroy();
 }

 public boolean canDestroy(){
 	synchronized(refLock){
 		return ref <= 0;
 	}
 }

 nbso online casino reviews  public String getKey() {
		return key;
	}

	public void setKey(String key) {
		this.key = key;
	}

	public boolean isOwner(boolean check) {
 	synchronized(refLock){
 		if(redisLock == null) {
 			logger.error("reidsLock is null:key=" key);
 			return false;
 		}
 		boolean a = reentrantLock.isHeldByCurrentThread();
 		boolean b = redisLock.isOwner();
 		if(check){
 			if(!a || !b){
 				logger.error(key ";a:" a ";b:" b);
 			}
 		}
 		return a && b;
 	}
 }

	public boolean setCleartime() {
		synchronized(this){
			if(cleartime>0) return false;
			this.cleartime = System.currentTimeMillis() 10*1000; 
			return true;
		}
	}

	public void resetCleartime(){
		synchronized(this){
			this.cleartime = -1;
		}
	}

	@Override
	public int compareTo(Delayed object) {
		if(object instanceof RedisReentrantLock){
			RedisReentrantLock t = (RedisReentrantLock)object;
	 long l = this.cleartime - t.cleartime;

			if(l > 0) return 1 ; //比当前的小则返回1,比当前的大则返回-1,否则为0
			else if(l < 0 ) return -1;
			else return 0;
		}
		return 0;
	}

	@Override
	public long getDelay(TimeUnit unit) {
		 long d = unit.convert(cleartime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
	 return d;
	}

}

 

/**
*使用Redis实现的分布式锁
*基本工作原理如下:
*1. 使用setnx(key,时间戮 超时),如果设置成功,则直接拿到锁
*2. 如果设置不成功,获取key的值v1(它的到期时间戮),跟当前时间对比,看是否已经超时
*3. 如果超时(说明拿到锁的结点已经挂掉),v2=getset(key,时间戮 超时 1),判断v2是否等于v1,如果相等,加锁成功,否则加锁失败,等过段时间再重试(200MS)
*/
public class RedisLock implements LockListener{
	private String key;
	private boolean owner = false;
	private AbstractLockObserver observer = null;
	private LockListener lockListener = null;
	private boolean waiting = false;
	private long expire;//锁超时时间,以秒为单位
	private boolean expired = false;

	public RedisLock(String key, LockListener lockListener, AbstractLockObserver observer) {
		this.key = key;
		this.lockListener = lockListener;
		this.observer = observer;
	}

	public boolean trylock(long expire) {
		synchronized(this){
			if(owner){
				return true;
			}
			this.expire = expire;
			this.expired = false;
			if(!waiting){
				owner = observer.tryLock(key,expire);
				if(!owner){
					waiting = true;
					observer.addLockListener(key, this);
				}
			}
			return owner;
		}
	}

	public boolean isOwner() {
		return owner;
	}

	public void unlock() {
		synchronized(this){
			observer.unLock(key);
			owner = false;
		}
	}

	public void clear() {
		synchronized(this){
			if(waiting) {
				observer.removeLockListener(key);
				waiting = false;
			}
		}
	}

	public boolean doExpire(){
		synchronized(this){
			if(owner) return true;
			if(expired) return false;
			expired = true;
			clear();
		}
		return false;
	}

	@Override
	public void lockAcquired() {
		synchronized(this){
			if(expired){
				unlock();
				return;
			}
			owner = true;
			waiting = false;
		}
		lockListener.lockAcquired();
	}

	@Override
	public long getExpire() {
		return this.expire;
	}

	@Override
	public void lockError() {
		synchronized(this){
			owner = false;
			waiting = false;
			lockListener.lockError();
		}
	}

}

public class LockObserver extends AbstractLockObserver implements Runnable{
	private CacheRedisClient client;
	private Object mutex = new Object();
	private Map<String,LockListener> lockMap = new ConcurrentHashMap();
	private boolean stoped = false;
	private long interval = 500;
	private boolean terminated = false;
	private CountDownLatch doneSignal = new CountDownLatch(1);
	
	public LockObserver(String schema){
		client = new CacheRedisClient(schema);
		
		SystemExitListener.addTerminateListener(new ExitHandler(){
			public void run() {
				stoped = true;
				try {
					doneSignal.await();
				} catch (InterruptedException e) {
				}
			}
		});
	}
	
	
	public void addLockListener(String key,LockListener listener){
		if(terminated){
			listener.lockError();
			return;
		}
		synchronized(mutex){
			lockMap.put(key, listener);
		}
	}
	
	public void removeLockListener(String key){
		synchronized(mutex){
			lockMap.remove(key);
		}
	}
	
	@Override
	public void run() {
		while(!terminated){
			long p1 = System.currentTimeMillis();
			Map<String,LockListener> clone = new HashMap();
			synchronized(mutex){
				clone.putAll(lockMap);
			}			
			Set<String> keyset = clone.keySet();
			if(keyset.size() > 0){
				ConnectionFactory.setSingleConnectionPerThread(keyset.size());
				for(String key : keyset){
					LockListener ll = clone.get(key);
					try{
					 if(tryLock(key,ll.getExpire())) {
					 	ll.lockAcquired();
					 	removeLockListener(key);
					 }
					}catch(Exception e){
						ll.lockError();
						removeLockListener(key);
					}
				}
				ConnectionFactory.releaseThreadConnection();
			}else{
				if(stoped){
					terminated = true;
					doneSignal.countDown();
					return;
				}
			}
			try {
				long p2 = System.currentTimeMillis();
				long cost = p2 - p1;
				if(cost <= interval){
					Thread.sleep(interval - cost);
				}else{
					Thread.sleep(interval*2);
				}
			} catch (InterruptedException e) {
			}
		}
		
	}
	
	
	/**
	 * 超时时间单位为s!!!
	 * @param key
	 * @param expire
	 * @return
	 */
	public boolean tryLock(final String key,final long expireInSecond){
		if(terminated) return false;
		final long tt = System.currentTimeMillis();
		final long expire = expireInSecond * 1000;
		final Long ne = tt expire;
		List<Object> mm = client.multi(key, new MultiBlock(){
			@Override
			public void execute() {
				transaction.setnxObject(key, ne);
				transaction.get(SafeEncoder.encode(key));
			}
		});
		Long res = (Long)mm.get(0);
	 if(new Long(1).equals(res)) {
	 	return true;
	 }else{
	 	byte[] bb = (byte[])mm.get(1);
			Long ex = client.deserialize(bb);
	 	if(ex == null || tt > ex){
	 		Long old = client.getSet(key, new Long(ne 1));
	 		if(old == null || (ex == null&&old==null) || (ex!=null&&ex.equals(old))){
	 			return true;
	 		}
	 	}
	 }
	 return false;
	}

	public void unLock(String key){
		client.del(key);
	}
}

 

 

使用本方案实现的分布式锁,可以完美地解决锁重入问题;通过引入超时也避免了死锁问题;性能方面,笔者自测试结果如下:

500线程 tps = 35000
[root@DB1 benchtest-util]# target/benchtest/bin/TestFastRedis /data/config/util/config_0_11.properties lock 500 500000
线程总时间:6553466;平均:13.106932
实际总时间:13609; 平均:0.027218

TPS达到35000,比方案1强了整整一个数量级;

五。总结
本文介绍了两种分布式锁的实现方案;方案1最大的优势在于避免结点挂掉后导致的死锁;方案2最大的优势在于性能超强;在实际生产过程中,结合自身情况来决定最适合的分布式锁方案是架构师的必修课。

发表在 Java技术, 分布式技术 | 一条评论

两种分布式锁实现方案(一)

声明
本文为Gleasy原创文章,转载请指明引自Gleasy团队博客
第一节:两种分布式锁实现方案(一)
第二节:两种分布式锁实现方案(二)

一。为何使用分布式锁?
当应用服务器数量超过1台,对相同数据的访问可能造成访问冲突(特别是写冲突)。单纯使用关系数据库比如MYSQL的应用可以借助于事务来实现锁,也可以使用版本号等实现乐观锁,最大的缺陷就是可用性降低(性能差)。对于GLEASY这种满足大规模并发访问请求的应用来说,使用数据库事务来实现数据库就有些捉襟见肘了。另外对于一些不依赖数据库的应用,比如分布式文件系统,为了保证同一文件在大量读写操作情况下的正确性,必须引入分布式锁来约束对同一文件的并发操作。

二。对分布式锁的要求
1.高性能(分布式锁不能成为系统的性能瓶颈)
2.避免死锁(拿到锁的结点挂掉不会导致其它结点永远无法继续)
3.支持锁重入

三。方案1,基于zookeeper的分布式锁

/**
* DistributedLockUtil.java
* 分布式锁工厂类,所有分布式请求都由该工厂类负责
**/
public class DistributedLockUtil {
	private static Object schemeLock = new Object();
	private static Object mutexLock = new Object();
	private static Map<String,Object> mutexLockMap = new ConcurrentHashMap();	
	private String schema;
	private Map<String,DistributedReentrantLock> cache = new ConcurrentHashMap<String,DistributedReentrantLock>();
	
	private static Map<String,DistributedLockUtil> instances = new ConcurrentHashMap();
	public static DistributedLockUtil getInstance(String schema){
		DistributedLockUtil u = instances.get(schema);
		if(u==null){
			synchronized(schemeLock){
				u = instances.get(schema);
				if(u == null){
					u = new DistributedLockUtil(schema);
					instances.put(schema, u);
				}
			}
		}
		return u;
	}
	
	private DistributedLockUtil(String schema){
		this.schema = schema;
	}
	
	private Object getMutex(String key){
		Object mx = mutexLockMap.get(key);
		if(mx == null){
			synchronized(mutexLock){
				mx = mutexLockMap.get(key);
				if(mx==null){
					mx = new Object();
					mutexLockMap.put(key,mx);
				}
			}
		}
		return mx;
	}
	
	private DistributedReentrantLock getLock(String key){
		DistributedReentrantLock lock = cache.get(key);
		if(lock == null){
			synchronized(getMutex(key)){
				lock = cache.get(key);
				if(lock == null){
					lock = new DistributedReentrantLock(key,schema);
					cache.put(key, lock);
				}
			}
		}
		return lock;
	}
	
	public void reset(){
		for(String s : cache.keySet()){
			getLock(s).unlock();
		}
	}
	
	/**
	 * 尝试加锁
	 * 如果当前线程已经拥有该锁的话,直接返回false,表示不用再次加锁,此时不应该再调用unlock进行解锁
	 * 
	 * @param key
	 * @return
	 * @throws InterruptedException
	 * @throws KeeperException
	 */
	public LockStat lock(String key) throws InterruptedException, KeeperException{
		if(getLock(key).isOwner()){
			return LockStat.NONEED;
		}
		getLock(key).lock();
		return LockStat.SUCCESS;
	}
	
	public void clearLock(String key) throws InterruptedException, KeeperException{
		synchronized(getMutex(key)){
			DistributedReentrantLock l = cache.get(key);
			l.clear();
			cache.remove(key);
		}
	}	
	
	public void unlock(String key,LockStat stat) throws InterruptedException, KeeperException{
		unlock(key,stat,false);
	}

	public void unlock(String key,LockStat stat,boolean keepalive) throws InterruptedException, KeeperException{
		if(stat == null) return;
		if(LockStat.SUCCESS.equals(stat)){
			DistributedReentrantLock lock = getLock(key);
			boolean hasWaiter = lock.unlock();
			if(!hasWaiter && !keepalive){
				synchronized(getMutex(key)){
					lock.clear();
					cache.remove(key);
				}
			}
		}
	}
	
	public static enum LockStat{
		NONEED,
		SUCCESS
	}
}

 

/**
*DistributedReentrantLock.java
*本地线程之间锁争用,先使用虚拟机内部锁机制,减少结点间通信开销
*/
public class DistributedReentrantLock {
	private static final Logger logger = Logger.getLogger(DistributedReentrantLock.class);
 private ReentrantLock reentrantLock = new ReentrantLock();

 private WriteLock writeLock;
 private long timeout = 3*60*1000;
 
 private final Object mutex = new Object();
 private String dir;
 private String schema;
 
 private final ExitListener exitListener = new ExitListener(){
		@Override
		public void execute() {
			initWriteLock();
		}
	};
	
	private synchronized void initWriteLock(){
		logger.debug("初始化writeLock");
		writeLock = new WriteLock(dir,new LockListener(){

			@Override
			public void lockAcquired() {
				synchronized(mutex){
					mutex.notify();
				}
			}
			@Override
			public void lockReleased() {
			}
 		
 	},schema);
		
		if(writeLock != null && writeLock.zk != null){
			writeLock.zk.addExitListener(exitListener);
		}
		
		synchronized(mutex){
			mutex.notify();
		}
	}
	
 public DistributedReentrantLock(String dir,String schema) {	
 	this.dir = dir;
 	this.schema = schema;
 	initWriteLock();
 }

 public void lock(long timeout) throws InterruptedException, KeeperException {
 reentrantLock.lock();//多线程竞争时,先拿到第一层锁
 try{
 	boolean res = writeLock.trylock();
 	if(!res){
	 	synchronized(mutex){
					mutex.wait(timeout);
				}
	 	if(writeLock == null || !writeLock.isOwner()){
	 		throw new InterruptedException("锁超时");
	 	}
 	}
 }catch(InterruptedException e){
 	reentrantLock.unlock();
 	throw e;
 }catch(KeeperException e){
 	reentrantLock.unlock();
 	throw e;
 }
 }
 
 public void lock() throws InterruptedException, KeeperException {
 	lock(timeout);
 }

 public void destroy() throws KeeperException {
 	writeLock.unlock();
 }
 

 public boolean unlock(){
 	if(!isOwner()) return false;
 try{
 	writeLock.unlock();
 	reentrantLock.unlock();//多线程竞争时,释放最外层锁
 }catch(RuntimeException e){
 	reentrantLock.unlock();//多线程竞争时,释放最外层锁
 	throw e;
 }
 
 return reentrantLock.hasQueuedThreads();
 }



 public boolean isOwner() {
 return reentrantLock.isHeldByCurrentThread() && writeLock.isOwner();
 }

	public void clear() {
		writeLock.clear();
	}

}

 

/**
*WriteLock.java
*基于zk的锁实现
*一个最简单的场景如下:
*1.结点A请求加锁,在特定路径下注册自己(会话自增结点),得到一个ID号1
*2.结点B请求加锁,在特定路径下注册自己(会话自增结点),得到一个ID号2
*3.结点A获取所有结点ID,判断出来自己是最小结点号,于是获得锁
*4.结点B获取所有结点ID,判断出来自己不是最小结点,于是监听小于自己的最大结点(结点A)变更事件
*5.结点A拿到锁,处理业务,处理完,释放锁(删除自己)
*6.结点B收到结点A变更事件,判断出来自己已经是最小结点号,于是获得锁。
*/
public class WriteLock extends ZkPrimative {
 private static final Logger LOG = Logger.getLogger(WriteLock.class);

 private final String dir;
 private String id;
 private LockNode idName;
 private String ownerId;
 private String lastChildId;
 private byte[] data = {0x12, 0x34};
 private LockListener callback;
 
 public WriteLock(String dir,String schema) {
 super(schema,true);
 this.dir = dir;
 }
 
 public WriteLock(String dir,LockListener callback,String schema) {
 	this(dir,schema);
 nbso online casino reviews  this.callback = callback;
 }

 public LockListener getLockListener() {
 return this.callback;
 }
 
 public void setLockListener(LockListener callback) {
 this.callback = callback;
 }

 public synchronized void unlock() throws RuntimeException {
 	if(zk == null || zk.isClosed()){
 		return;
 	}
 if (id != null) {
 try {
 	 zk.delete(id, -1); 
 } catch (InterruptedException e) {
 LOG.warn("Caught: " e, e);
 //set that we have been interrupted.
 Thread.currentThread().interrupt();
 } catch (KeeperException.NoNodeException e) {
 // do nothing
 } catch (KeeperException e) {
 LOG.warn("Caught: " e, e);
 throw (RuntimeException) new RuntimeException(e.getMessage()).
 initCause(e);
 }finally {
 if (callback != null) {
 callback.lockReleased();
 }
 id = null;
 }
 }
 }
 
 private class LockWatcher implements Watcher {
 public void process(WatchedEvent event) {
 LOG.debug("Watcher fired on path: " event.getPath() " state: " 
 event.getState() " type " event.getType());
 try {
 trylock();
 } catch (Exception e) {
 LOG.warn("Failed to acquire lock: " e, e);
 }
 }
 }
 
 private void findPrefixInChildren(String prefix, ZooKeeper zookeeper, String dir) 
 throws KeeperException, InterruptedException {
 List<String> names = zookeeper.getChildren(dir, false);
 for (String name : names) {
 if (name.startsWith(prefix)) {
 id = dir "/" name;
 if (LOG.isDebugEnabled()) {
 LOG.debug("Found id created last time: " id);
 }
 break;
 }
 }
 if (id == null) {
 id = zookeeper.create(dir "/" prefix, data, 
 acl, EPHEMERAL_SEQUENTIAL);

 if (LOG.isDebugEnabled()) {
 LOG.debug("Created id: " id);
 }
 }

 }

	public void clear() {
		if(zk == null || zk.isClosed()){
 		return;
 	}
		try {
			zk.delete(dir, -1);
		} catch (Exception e) {
			 LOG.error("clear error: " e,e);
		} 
	}
	
 public synchronized boolean trylock() throws KeeperException, InterruptedException {
 	if(zk == null){
 		LOG.info("zk 是空");
 		return false;
 	}
 if (zk.isClosed()) {
 	LOG.info("zk 已经关闭");
 return false;
 }
 ensurePathExists(dir);
 
 LOG.debug("id:" id);
 do {
 if (id == null) {
 long sessionId = zk.getSessionId();
 String prefix = "x-" sessionId "-";
 idName = new LockNode(id);
 LOG.debug("idName:" idName);
 }
 if (id != null) {
 List<String> names = zk.getChildren(dir, false);
 if (names.isEmpty()) {
 LOG.warn("No children in: " dir " when we've just " 
 "created one! Lets recreate it...");
 id = null;
 } else {
 SortedSet<LockNode> sortedNames = new TreeSet<LockNode>();
 for (String name : names) {
 sortedNames.add(new LockNode(dir "/" name));
 }
 ownerId = sortedNames.first().getName();
 LOG.debug("all:" sortedNames);
 SortedSet<LockNode> lessThanMe = sortedNames.headSet(idName);
 LOG.debug("less than me:" lessThanMe);
 if (!lessThanMe.isEmpty()) {
 	LockNode lastChildName = lessThanMe.last();
 lastChildId = lastChildName.getName();
 if (LOG.isDebugEnabled()) {
 LOG.debug("watching less than me node: " lastChildId);
 }
 Stat stat = zk.exists(lastChildId, new LockWatcher());
 if (stat != null) {
 return Boolean.FALSE;
 } else {
 LOG.warn("Could not find the" 
 		" stats for less than me: " lastChildName.getName());
 }
 } else {
 if (isOwner()) {
 if (callback != null) {
 callback.lockAcquired();
 }
 return Boolean.TRUE;
 }
 }
 }
 }
 }
 while (id == null);
 return Boolean.FALSE;
 }

 public String getDir() {
 return dir;
 }

 public boolean isOwner() {
 return id != null && ownerId != null && id.equals(ownerId);
 }

 public String getId() {
 return this.id;
 }
}

 

使用本方案实现的分布式锁,可以很好地解决锁重入的问题,而且使用会话结点来避免死锁;性能方面,根据笔者自测结果,加锁解锁各一次算是一个操作,本方案实现的分布式锁,TPS大概为2000-3000,性能比较一般;

发表在 Java技术, 分布式技术 | 一条评论

Gleasy的NOSQL数据库集群Cloudredis

声明
本文为Gleasy原创文章,转载请指明引自Gleasy团队博客

一。问题的提出
redis是目前最快的NO-SQL数据库之一,GLEASY自测的结果也充分证实了这一点。除了性能极强之外,还提供了丰富的数据结构,可以更简洁地解决许多关键性能问题。基于此,GLEASY选用REDIS作为缓存服务器以及存储服务器。并且在此基础上研发了分布式的消息队列cloud-mq,分布式的即时通讯中间件cloud-im,分布式文件系统cloudfs,分布式作业调度中间件cloudjob.这些中间件共同组成了gleasy高性能分布式系统的架构基础。redis做缓存服务器是没有任何问题的,由于缓存本身意识着可丢失。但作为存储服务器,除性能之外,还必须考虑高可用性(High availability)和一致性(Consistency)以及扩展性(Scalability)。而这些能力洽洽是redis所欠缺的。基于这些需求,GLEASY自主研发了分布式的redis集群cloudredis。

二。设计思想
1. 软件架构
a. powered by java
b. 使用mina2作为NIO服务器

2. 主体架构图

图1. 集群内部结构图


图2.客户端连接集群

3. 设计思想
3.1 基本构成
cloudredis集群由若干个cloudredis结点以及1-2个binglog server组成。
cloudredis: 可以称为redis代理服务器。唯一代理某个redis服务器,所有对该redis的访问都必须通过cloudredis进行代理。
binglog server: 独立的redis服务器,用于保存读取binlog和集群状态信息.

3.1 一致性和高可用问题
所有访问redis的操作都经过cloudredis,cloudredis会分析所有访问请求,抽取其中的写命令,并将命令以特定格式写入binlog server。同时,每台cloudredis会启动1个后端线程读取binglog server中其它结点的binlog,另外1个后端线程执行得到的binlog内容。这种机制非常类似于mysql的binglo机制。不同之处在于,这里的binlog不是以文件方式存在,而是直接放在独立的binglog server(redis)里面。为什么这么设计?答案是:性能。文件式的binlog同步已经满足不了redis这种10万TPS的应用场景了。当然,如果由于某种原因(binlog server全部挂掉?),无法连上binlog server,cloudredis会把binlog暂时存在本地文件系统中,等待binglog server恢复之后,会一次过将本地文件系统中的binlog重新写入binlog server从而最大限度上保证binlog不丢失.通过写入和消费binlog,集群内所有结点的数据会做到准实时(延迟在MS级)。

3.2 如何高可用?
通过使用cloud-redis的客户端连接上集群内任意一台结点cloudredis之后,客户端需要定时向cloudredis获取集群内存活结点的信息(ip和端口)。并且全部与它们建立连接,将访问请求使用一致性HASH机制,平均地分配到这些结点上面去。集群内任意一台结点挂掉,客户端会重新进行HASH计算,并且将请求指向存活的结点,由此做到高可用。

3.3 如何实现一致性?
通过使用一致性hash,对某个key的访问会去到唯一的一台服务器,从而避免读写分离方案中的数据延迟带来的不一致性问题。

3.4 如何做到扩展性?
cloudredis并不解决扩展性问题,只解决集群内的高可用和一致性问题。扩展性问题通过客户端的人工shard方式实现。

3.5 Cloudredis与CAP的关系
Cloudredis集群首先是允许分区P的存在,不单止允许,而且从一开始就显式地引入分区P:集群中的每个结点,在某一时刻的状态都可能不一致的;显式引入分区,并且通过一致性HASH算法将某一KEY的访问限制在特定的结点(此处可以理解为一个不变性约束规则),结点内部保证数据一致性,结点之间通过严格遵守顺序的BINLOG日志来同步状态,所有结点BINLOG同步完成,分区消失,结点间数据一致性得以保证。Cloudredis在分区存在情况下,不关心其它结点是否存在,继续提供服务,因此是偏向于PA的设计,甚至可以说属于BASE(可用为主,软状态【非实时同步】,最终一致)。

三。 性能怎么样?
我们在相同环境下跟原生redis对比,来判断cloudredis的性能如何。测试工具使用redis自带的压力测试工具 redis-benchmark.
1. 原生redis的性能测试

C语言版本benchmark 200连接
[root@gleasy redis]# bin/redis-benchmark -h 192.168.0.11 -p 6677 -t set -n 1000000 -c 200
====== SET ======
1000000 requests completed in 13.60 seconds
200 parallel clients
3 bytes payload
keep alive: 1
2.42% <= 1 milliseconds
19.71% <= 2 milliseconds
86.69% <= 3 milliseconds
97.18% <= 4 milliseconds
99.36% <= 5 milliseconds
73518.60 requests per second

[root@gleasy redis]# bin/redis-benchmark -h 192.168.0.11 -p 6677 -t get -n 1000000 -c 200
====== GET ======
1000000 requests completed in 13.12 seconds
200 parallel clients
3 bytes payload
keep alive: 1

1.96% <= 1 milliseconds
17.60% <= 2 milliseconds
90.74% <= 3 milliseconds
97.44% <= 4 milliseconds
99.60% <= 5 milliseconds
99.94% <= 6 milliseconds
99.99% <= 7 milliseconds
100.00% <= 7 milliseconds
76190.48 requests per second

总结: 读操作76K,写操作73K.

2. 单结点的cloudredis性能测试(不开binlog)

[root@gleasy redis]# bin/redis-benchmark -h 192.168.0.11 -p 16677 -t set -n 1000000 -c 200
====== SET ======
1000000 requests completed in 12.99 seconds
200 parallel clients
3 bytes payload
keep alive: 1
9.57% <= 1 milliseconds
72.71% <= 2 milliseconds
84.95% <= 3 milliseconds
97.21% <= 9 milliseconds
97.93% <= 10 milliseconds
76952.67 requests per second

[root@gleasy redis]# bin/redis-benchmark -h 192.168.0.11 -p 16677 -t get -n 1000000 -c 200
====== GET ======
1000000 requests completed in 12.03 seconds
200 parallel clients
3 bytes payload
keep alive: 1
11.86% <= 1 milliseconds
77.96% <= 2 milliseconds
86.32% <= 3 milliseconds
89.91% <= 4 milliseconds
83153.17 requests per second
小结: 读操作 83K, 写操作 76K。 (性能比原生的redis有10%的提升,很惊奇?后面讲解原理。。。)

3. 集群双结点,开binlog,cloudredis性能测试(使用cloud-redis客户端)

[root@gleasy cloudredis]# bin/redis-benchmark -t get -h 192.168.0.11 -p 6679 -d 15 -l 60 -c 800 online casino -b shard -m 10 -n 50
成功:5139656
失败:0
命中缓存:0
总时间:48002194
最长用时:117
最短用时:0
平均用时:9.339573309964713
min tps:0
max tps:195370
avg tps:84256

[root@gleasy cloudredis]# bin/redis-benchmark -t set -h 192.168.0.11 -p 6679 -d 15 -l 30 -c 800 -b shard -m 10 -n 50
成功:2282561
失败:0
命中缓存:0
总时间:24004343
最长用时:121
最短用时:0
平均用时:10.516408104756017
min tps:0
max tps:140498
avg tps:73631
小结: 读操作84K,写操作73K

通过以上对比,我们可以清晰地看到,cloudredis无论在单结点还是多结点情况下,都至少能够做到和原生redis相同的性能要求。甚至会有更好的性能表现。

四。 运维工具
除了作为redis的代理之外,cloudredis还提供了一系列的集群管理工具,方便管理集群
1. info

说明:会在原来redis info命令基础之上,额外返回本集群的信息
示例:
#集群信息
结点标识:11-6673
IP:192.168.0.11
端口: 6673
代理的redis:192.168.0.11:16673
当前连接数:34
内存(最大/己分配/己分配中剩余/最大可用):2048/512/170/1706(M)
CPU核数:4
集群开启:true
是否只读:false
日志开启:true
日志故障:false

#其它结点
结点:10-6673
IP:192.168.0.10
端口:6673
己同步binlog位置:1420434 1432391
最大binlog位置:1420434 1432391

2. cluster start|stop

说明: 开启/关闭本结点的集群功能
讲解:
cluster start: 将本结点挂上集群,可以被客户端访问到
cluster stop:将本结点从集群拿下,客户端将不再继续访问得到

3. cluster syn ip port

说明: 非常好用的同步工具
讲解:
cluster syn 192.168.0.11 6673 :与目标服务器完全同步数据(清除自己的原有数据,将将目标服务器数据全部拿过来),并且会将目标服务器的binlog位置等信息同步过来。是集群管理必用的工具。

4. cluster binlog start|stop|restart

说明: 开启,关闭,重新开启binlog
讲解:
cluster binlog start: 开启binlog
cluster binlog stop: 关闭binlog(不再写日志啦,这个很危险)
cluster binlog restart: 用于将本地文件系统中暂存的binlog一次性刷新进binlog server。用于binlog server故障恢复后使用。

5. cluster visible on|off

说明:对客户端可见,不可见
讲解:
cluster visible on :客户端可以访问得到
cluster visible off:客户端不可访问,等于集群中没有这个结点

6. cluster lockwrite|unlockwrite

说明:设置只读和可读写
讲解:
cluster lockwrite:只读(挂起所有写操作)
cluster unlockwrite:结束只读(执行所有被挂起的写操作)

五。 其它
1. 与redis官方研发中的cluster方案有何异同?为何要自己开发?
答: 通常来说,我们认为cluster主要的能力包括:扩展(Scalability),高可用,性能分担。附加要求为一致性,安全性等。从这层意思上说,我们的cloudredis应该只算是准集群方案,它并不解决扩展性问题。扩展性问题我们通过redis-shard使用人工shard方式来解决。我们使用binlog方式进行数据同步,一致性hash进行请求分发,而且把大量的功能放在客户端。之所以这么做,最根本的原因是我们要保持本方案的高性能。采用真正意义上的cluster,其性能从理论上必定有非常严重的损耗。这个可以从淘宝的tair系统得到证实。而我们的集群方案,可以保持甚至可以超越原生的单台redis。

2. 为什么代理的性能可以超越原生redis?
答: redis有一套机制叫做pipeline. 比如A,B,C命令同时到达cloudredis,如果不使用pipeline,我们可能要逐条命令送过去执行,假如每个需要1MS,我们共要3MS。但我们可以将A,B,C打包发送给redis执行,需时1.5MS。这样我们就节省了1.5MS。如果更多的命令同时到达,节省的辐度会更大。当到大一定程度便足可以抵销代理与REDIS之间的网络等开销。

六。 缺点和限制
1. cloudredis并不能完全支持全部的redis命令,不被支持命令包括: blpop,blpush,blpoprpush,subscribe,unsubscribe,zinterstore,zunionstore.
2. cloudredis使用多线程,会有更高的CPU占用,这也是压力测试中表现优秀的关键因素之一。

七。关于开放源码的说明
Gleasy现阶段暂无开放源码的打算,但为学习目的可提供源码供大家参考。如有需要可以在留下联系方式。

发表在 Java技术, 分布式技术, 数据库技术 | 3 条评论

Gleasy平台开发经验总结

目录
1. 前言
2. 数据库开发
2.1. 数据库
2.2. 数据表
2.3. 数据字段
3. 服务端开发
3.1. 实体层
3.2. 数据访问层
3.3. 服务层
3.4. 缓存层
3.5. 索引层
3.6. 控制层
3.7. 接口层
4. 前端开发
4.1. 数据访问层
4.2. 组件化开发
4.3. 组件服务化
4.4. 数据缓存层
4.5. 定时器
5. 结束语

1. 前言

1) 基础平台:Gleasy开发平台

2) 目标:经验 -> 规范 -> 软件工程化

2. 数据库开发

针对MySql

2.1. 数据库

2.1.1. 命名规范

1) 如:systemCircle-boot、systemCircle-global、systemCircle-local-0、systemCircle-local-1

2.2. 数据表

2.2.1. 命名规范

1) 如: card、card_extra_info

2.3. 数据字段

2.3.1. 命名规范

1) 如: driver_class_name

2) 唯一主键:id

2.3.2. 类型规范

1) 标识:bigint(20)

2) 字符串:varchar(32)、varchar(64) 、varchar(128)、 varchar(256) 、 varchar(512) 、 varchar(1024) 、 varchar(2048) 、 varchar(4096)

3) 布尔:tinyint(1)

4) 日期:datetime、date、time

5) 数值:tinyint(1)、smallint(5)

6) 其他:有待补充

2.3.3. 其他约束

1) 不允许使用外键(数据关联,是业务层干的活)。

2) 一般情况下,只有一个主键,而且就是字段名就是id。

3. 服务端开发

3.1 实体层(Domain)

3.1.1 实体类

与数据库表一一对应的类。类名称对应表名称,类属性对应表字段。

3.1.1.1 包规范

Source Folder:src/main/java

如:com.gleasy.person.domain、com.gleasy.gcd.domain

3.1.1.2 命名规范

如:Card.java、CardExtraInfo.java

(直接对应数据库表card、card_extra_info)

3.1.2 实体属性

3.1.2.1 命名规范

如:id、userId、birthdayType

3.1.2.2 类型规范

与数据库类型一一对应(虽然不是很严谨,但能解决大多数问题)

1)    标识:java.lang.Long

2)    字符串:java.lang.String

3)    布尔:java.lang.Byte(没错,不是boolean。0:false;1:true)

4)    日期:java.util.Date、java.sql.Date、java.sql.Time

5)    数值:java.lang.Byte、java.lang.Integer

来个映射表,直观些:

数据库 实体类
bigint(20) java.lang.Long
varchar java.lang.String
tinyint(1) java.lang.Byte
datetime java.util.Date
Date java.sql.Date
Time java.sql.Time
smallint(5) java.lang.Integer

3.1.3. 实体包装类

顾名思义,给实体类增加一些字段,方便在应用层使用。

3.1.3.1. 包规范

Source Folder:src/main/java

如:com.gleasy.person.bean

3.1.3.2. 命名规范

如:CardWrap.java、CardExtraInfoWrap.java

(当然拉,Card和CardWrap最好是继承关系)

3.1.4. IBatis映射文件

3.1.4.1. 包规范

Source Folder:src/main/resources

如:com.gleasy.person.domain、com.gleasy.gcd.domain

(与实体类包路径一致)

3.1.4.2. 命名规范

如:Card.map.xml、CardExtraInfo.map.xml

(前缀与类名一致)

3.1.4.3. 内容范例

3.1.4.4. 自动化

满足以上规范,实体类&Ibatis映射文件,甚至可通过自动化代码生成。^_^开心吧…

3.2. 数据访问层(Dao)

3.2.1. 包规范

接口如:com.gleasy.person.dao

实现如:com.gleasy.person.dao.ibatis

3.2.2. 类命名规范

接口如:CardDAO.java、CardExtraInfoDAO.java

实现如:CardDAOImpl.java、CardExtraInfoDAOImpl.java

(与实体类Card.java、CardExtraInfo.java一一对应)

3.2.3. 继承

接口需继承BaseLocalDAO,如:public interface CardDAO extends BaseLocalDAO<Card>

实现需继承BaseLocalDAOImpl,如:public class CardDAOImpl extends BaseLocalDAOImpl<Card> implements CardDAO

继承最直接的好处,就是不用自己实现get、save、delete、update方法。

3.2.4. 增单个对象

BaseLocalDAOImpl实现的save方法,但没有包含id的生成规则。所以一般情况下需要自己定义新的save方法。

3.2.4.1. 方法命名规范

如:saveCard,对应实体类Card。

3.2.4.2. 实例

3.2.5. 删单个对象

BaseLocalDAOImpl已经实现了。

3.2.6. 改单个对象

BaseLocalDAOImpl已经实现了。

3.2.7. 查单个对象

BaseLocalDAOImpl已经实现了。

3.2.8. 查多个对象

自定义方法,一般只允许根据多个id查询。其他复杂查询通过缓存、索引平台、索引库实现。

3.2.9. 其他操作

不允许

3.3. 服务层(Service)

3.3.1. 包规范

接口如:com.gleasy.person.service

实现如:com.gleasy.person.service.impl

(person代表项目标识)

3.3.2. 类命名规范

3.3.2.1. 按业务命名

如:ImportService表示导入业务功能,ExportService表示导出业务功能,CardService表示名片相关的功能。

ImportServiceImpl、ExportServiceImpl、CardServiceImpl是实现类。

3.3.3. 上下层关联

指Service调用DAO。

一般情况下,一个Service只能调用一个DAO,比如CardService只能调用CardDAO。

(反过来也就是,一般情况下,一个DAO都必将被一个Service所包)

3.3.4. 同层关联

指Service调用Servcie。

允许一个Service调用多个Service,这很常用。

3.3.5. 纯业务Service

也就是那些不直接调用DAO的Service。

3.3.6. 方法参数问题

3.3.6.1. 参数暴露

一般情况下,采用这种方式。直观,容易维护。

如:… 方法(参数1, 参数1, … , 参数N) throws BusinessLevelException;

3.3.6.2. 参数封装

如:… 方法(实体对象1, 实体对象2, … , 实体对象N) throws BusinessLevelException;

非迫不得已,禁止如下方式:

… 方法(Map1, Map2 , … , MapN) online casino throws BusinessLevelException;

3.4. 缓存层(Cache)

基本思想,将缓存的功能代码独立开来,并提供接口给服务层调用。

缓存一般非持久化,有失效时间。

3.4.1. 包命名规范

如:com.gleasy.person.cache、com.gleasy.person.cache.impl

3.4.2. 类命名规范

一般每个DAO对应一个Cache,如CardDAO对应CardCache、CardCacheImpl

3.4.3. 主要方法

3.4.3.1. put

将对象放入缓存

3.4.3.2. get

在缓存查询对象

3.4.3.3. remove

删除缓存对象

3.4.3.4. mget

批量查询缓存(按ID)

3.4.4. 服务层调用缓存

3.4.4.1. 增

写库 -> 写缓存

3.4.4.2. 删

删库 -> 删缓存

3.4.4.3. 改

先查 -> 改库 -> 写缓存

3.4.4.4. 查

查缓存 -> 不在缓存则查库 -> 不在缓存则写缓存

3.4.5. 持久化缓存

除非主动删除,否则永不失效。需根据具体业务进行设计,这里不做具体介绍。

3.5. 索引层(Index)

调用全文索引平台的能力,实现复杂查询。

需索引平台提供了相应的接口实现索引的创建和查询。对于应用来说,主要是实现接口的二次封装,以方便服务层调用。

3.5.1. 包命名规范

如:com.gleasy.person.index、com.gleasy.person.index.impl

3.5.2. 类命名规范

如:CardIndex、CardIndexImpl

3.5.3. 更新索引

如:public void update(Card card)

3.5.4. 删除索引

public void remove(Card card)

3.5.5. 清除索引

public void clear()

3.5.6. 查索引

public List<Map> search(参数1, 参数2, … , 参数N , Integer pageNum, Integer pageSize)

pageNum:第几页,从1开始

pageSize:每页大小

3.5.7. 查询计数

public Integer count(参数1, 参数2, … , 参数N)

3.6. 控制层(Action)

负责接收过滤用户参数,并调用恰当的服务解决问题,最后以Json格式返回结果。

3.6.1. 包命名规范

如:com.gleasy.person.card.openapi.action

person:项目标识

card:业务划分

openapi:标识公开(一般去掉表示内网调用)

3.6.2. 类命名规范

3.6.2.1. 增

如:SaveCardAction

3.6.2.2. 删

如:DeleteCardsByCardIdsAction

3.6.2.3. 改

如:UpdateAvatarAction

3.6.2.4. 查

如:GetCardsByCardIdsAction

3.6.3. 上下层关联

也就是Action层调用Servcie层。

一般一个Action只会调用一个Service。

3.6.4. 配置文件

按业务分离,一般一个包对应一个配置文件。

如:com.gleasy.person.card.openapi.action对应applicationContext-controller-card.xml

当然了,这个包下的Action,都需要加上前缀/card/

而且这个也需要对应前端的CardModel

3.6.5. 其他

不要把业务的东西放在Action层。难维护,难复用功能。

复杂的参数解析,特别前端的Json格式的内容解析,可以放在Service层处理。

3.7. 接口层(Api)

以Jar包的方式,对控制层进行封装,方便其他项目调用。如常用的GcdApi、UserInfoClient。

4. 前端开发

4.1. 数据访问层(Model)

4.1.1. 实例

4.1.2. 包规范

如:com.gleasy.attendance.model

(这里的attendance一般指项目标识,可能多级,如:com.gleasy.a.b.c.attendance.model)

4.1.3. 命名规范

类名:ManagerModel(对应applicationContext-controller-manager.xml)

变量名/方法名:createManagersAction/createManagers(对应CreateManagersAction.java)

4.1.4. 参数规范

一般情况下function(x, y, z, success, error, token)

x,y,z是业务级参数,success,error,token是框架级参数(独立开参数的考虑是增强可读性、可维护性)

特殊情况下function(object, success, error, token)

object包含大量(一般>10个)参数,如:card就有40个字段,object一般是对应实体类。

防止传”null”字符串到后台

这段代码是必须的。


4.1.5. 单例类

4.1.6. 调用方式

4.2. 组件化开发(components)

先按照业务功能、系统功能、页面区域划分组件。

4.2.1. 组件定义

最简单的组件由一个JS文件和一个HTML文件组成。如部门的下拉列表组件就由:DepartmentSelector.class.js和DepartmentSelector.html组成(名称需要一致,便于阅读)。
html文件中JS组件需要的所有html元素(不包含子组件需要的html元素)。

4.2.1.1. 实例



4.2.2. 组件交互
4.2.2.1. Notify机制

每个组件都有notify方法,用于监听事件。
组件通过container直接获取应用句柄,调用notify,要求传递消息。
顶级组件(container)自顶向下,逐层下发消息(下发过程中不允许改变消息名称)。
组件收到消息后,自行决定是否处理or向下传播。
通过parent直接获取直接父组件句柄,进而调用父组件notify方法(比较少用)。

4.2.2.2. 注册监听机制

高度解耦,但维护成本过高,不推荐。

4.2.2.3. 延迟回调机制

一般情况下,组件间的相互调用,结果都需要以回调的方式获取较为靠普。但如果调用的时候,对方组件还没有完成自身初始化工作,咋办?延迟回调即可,代码如下:



4.3. 组件服务化(Daemon)

如果您的组件需要被跨项目调用,组件服务化是最佳选择。
注册服务(略)。
调用服务:


4.4. 数据缓存层(Cache)

基本思想:不直接调用Model,而是调用Cache。Cache判断是否前端有缓存且未过期,满足条件使用前端缓存,否则调用Model,并更新缓存。

4.5. 定时器
4.5.1. 循环任务(setInterval)

4.5.1.1. setInterval陷阱

经典案例:一说频繁发状态变更消息,导致收发双方系统崩溃。
根本原因:多次并发调用了setInterval。
解决办法:setInterval之前要先clearInterval,要不然代码被异步调用多次,之前的就永远也清除不了,成死循环了。

4.5.2. 定时任务(setTimeout)

5. 结束语

以上经验,是个人在实际的研发过程中不断探索并汲取其他同事经验中的精华部分,最后总结出来的,但仍不免存在纰漏,期待您的批评意见(dingyong@gleasy.net)。

发表在 Java技术, 分布式技术, 前端技术, 数据库技术 | 留下评论