文章

上一篇文章简单写了JVM的常见垃圾回收算法，今天就让我们看看根据这些算法有哪些常见的垃圾收集器，他们有什么特点，然后根据自己的应用特点和要求组合出各个年代所使用的收集器。 上图展示了JDK1.7Update14之后的HotSpot虚拟机的7种作用于不同分代的收集器，如果两个收集器之间存在连线，就说明它们可以搭配使用。虚拟机所处的区域，则表示它是属于新生代收集器还是老年代收集器 Serial收集器 Serial收集器是最基本、发展历史最悠久的收集器，在JDK 1.3.1之前是虚拟机新生代收集的唯一选择。见名知意它是一个单线程的收集器，“单线程”的意义并不仅仅说明它只会使用一个CPU或一条收集线程去完成垃圾收集工作，更重要的是在它进行垃圾收集时，必须暂停其他所有的工作线程，直到它收集结束，也就是传说中的Stop The World，简称STW。那么它是不是已经被淘汰的一个垃圾收集器呢，事实上并不是，由于与其他收集器的单线程比简单而高效，对于限定单个CPU的环境来说，Serial收集器由于没有线程交互的开销，专心做垃圾收集自然可以获得最高的单线程收集效率。所以Serial收集器是虚拟机运行在Client模式下的默认新生代收集器。它的运行示意图如下： ParNew收集器 ParNew收集器其实就是Serial收集器的多线程版本，是一个并行垃圾收集器，并行的含义是：多条垃圾收集线程并行工作，但此时用户线程仍然处于等待状态。除了使用多条线程进行垃圾收集之外，其余行为包括Serial收集器可用的所有控制参数、收集算法、Stop The World、对象分配规则、回收策略等都与Serial收集器完全一样，在实现上，这两种收集器也共用了相当多的代码。ParNew收集器是许多运行在Server模式下的虚拟机中首选的新生代收集器，还有一个和性能无关的但是很重要的原因：除了Serial收集器外，目前只有它能与CMS收集器配合工作。和Serial收集器相比，ParNew收集器在单CPU的环境中绝对不会有比Serial收集器更好的效果，甚至由于存在线程交互的开销，该收集器在通过超线程技术实现的两个CPU的环境中都不能百分之百地保证可以超越Serial收集器。然而，随着可以使用的CPU的数量的增加，它对于GC时系统资源的有效利用还是很有好处的。其运行示意图如下： Parallel Scavenge收集器 Parallel Scavenge收集器是一个新生代收集器，它也是使用复制算法的收集器，又是并行的多线程收集器。那么他有什么应用场景呢？事实上Parallel Scavenge收集器的特点是它的关注点与其他收集器不同，其他收集器的关注点是尽可能地缩短垃圾收集时用户线程的停顿时间，而Parallel Scavenge收集器的目标则是达到一个可控制的吞吐量（Throughput），所以由于与吞吐量关系密切，Parallel Scavenge收集器也经常称为“吞吐量优先”收集器。所以它的应用场景体现在：停顿时间越短就越适合需要与用户交互的程序，良好的响应速度能提升用户体验，而高吞吐量则可以高效率地利用CPU时间，尽快完成程序的运算任务，主要适合在后台运算而不需要太多交互的任务。除此之外和ParNew收集器相比，它具有自适应调节策略。Parallel Scavenge收集器有一个参数-XX:+UseAdaptiveSizePolicy。当这个参数打开之后，就不需要手工指定新生代的大小、Eden与Survivor区的比例、晋升老年代对象年龄等细节参数了，虚拟机会根据当前系统的运行情况收集性能监控信息，动态调整这些参数以提供最合适的停顿时间或者最大的吞吐量，这种调节方式称为GC自适应的调节策略（GC Ergonomics），其运行示意图如下： Serial Old收集器 Serial Old是Serial收集器的老年代版本，它同样是一个单线程收集器，使用标记－整理算法。它的主要应用场景：1. 给Client模式下的虚拟机使用（还记得Client模式下新生代的默认垃圾收集器是啥吗？）；2. 在Server模式下，那么它主要还有两大用途：一种用途是在JDK 1.5以及之前的版本中与Parallel Scavenge收集器搭配使用，另一种用途就是作为CMS收集器的后备预案，在并发收集发生Concurrent Mode Failure时使用。运行示意图见Serial垃圾收集器 Parallel Old收集器 Parallel Old是Parallel Scavenge收集器的老年代版本，使用多线程和“标记－整理”算法。在注重吞吐量以及CPU资源敏感的场合，都可以优先考虑Parallel Scavenge加Parallel Old收集器。这个收集器是在JDK 1.6中才开始提供的，在此之前，新生代的Parallel Scavenge收集器一直处于比较尴尬的状态。原因是，如果新生代选择了Parallel Scavenge收集器，老年代除了Serial Old收集器外别无选择（Parallel Scavenge收集器无法与CMS收集器配合工作）。由于老年代Serial Old收集器在服务端应用性能上的“拖累”，使用了Parallel Scavenge收集器也未必能在整体应用上获得吞吐量最大化的效果，由于单线程的老年代收集中无法充分利用服务器多CPU的处理能力，在老年代很大而且硬件比较高级的环境中，这种组合的吞吐量甚至还不一定有ParNew加CMS的组合“给力”。直到Parallel Old收集器出现后，“吞吐量优先”收集器终于有了比较名副其实的应用组合。运行示意图见Parallel Scavenge垃圾收集器 CMS收集器 在JDK 1.5时期，HotSpot推出了一款在强交互应用中几乎可认为有划时代意义的垃圾收集器——CMS收集器，这款收集器是HotSpot虚拟机中第一款真正意义上的并发收集器，它第一次实现了让垃圾收集线程与用户线程同时工作。并发的含义就是：指用户线程与垃圾收集线程同时执行（但不一定是并行的，可能会交替执行），用户程序在继续运行，而垃圾收集程序运行于另一个CPU上，他可不用同于并行，并行会有STW，并发几乎没有STW。CMS（Concurrent Mark Sweep）收集器是一种以获取最短回收停顿时间为目标的收集器。目前很大一部分的Java应用集中在互联网站或者B/S系统的服务端上，这类应用尤其重视服务的响应速度，希望系统停顿时间最短，以给用户带来较好的体验。CMS收集器就非常符合这类应用的需求。它的运行示意图如下： 从图上可以看出，CMS收集器是基于“标记—清除”算法实现的，它的运作过程相对于前面几种收集器来说更复杂一些，整个过程分为4个步骤： ①. 初始标记（CMS initial mark） 初始标记仅仅只是标记一下GC Roots能直接关联到的对象，速度很快，需要“Stop The World”。 ...

上一篇文章简单写了一下JVM如何判断一个对象是否已经死了，当判断出一个对象已经死了之后，接下来就要进行垃圾回收了，所以在进行垃圾回收之前，先让我们看看垃圾回收的算法有哪些。 标记-清除算法 标记清除见名知意该算法分为“标记”和“清除”两个阶段：首先标记出所有需要回收的对象，在标记完成之后统一回收所有被标记的对象，至于如何标记就是上一篇文章中所讲的方法了。这种算法比较简单，容易理解，同时也是一个最基础的垃圾回收算法，后面所讲的算法都是对他的改进，至于为什么需要改进，因为他主要存在两个不足： ①. 效率问题，标记和清除两个阶段的效率都不高； ②. 空间问题，标记清除之后会留下大量的不连续的内存碎片，内存碎片会导致当后面在程序的运行过程中可能需要给较大的对象分配空间时，无法找到足够的内存而不得不提前触发另一次垃圾回收。这种算法的执行过程如下图： 复制算法 复制算法为了解决标记清除算法的效率问题：它将内存分为容量大小相同的两块，每次只使用其中一块，当这一块的内存空间用完了，他就讲里面存活着的对象复制到另一块上面，然后再把已使用过的这一块内存空间一次性清理掉，这样使得每次都是对半个内存区域进行会回收，内存分配时也不用考虑空间碎片的问题，只需要移动指针，按顺序分配即可，实现简单，运行高效。但是他也有缺点：每次使用的内存空间只有整个空间的一半，这“浪费”有点高啊。复制算法的执行过程如下图： 不过现代的商业JVM都采用了这种算法来回收新生代，究其原因不仅仅他解决了效率问题，更是经研究表明：新生代中的对象高达98%都是“朝生夕死”的，所以这样一来就不需要1:1来划分空间了。直接将内存空间分为一块较大的Eden区和两块较小的Survivor区，每次使用Eden和其中一块Survivor区，当需要回收时，直接将Eden和Survivor中还存活的对象一次性的复制到另一块Survivor区，最后在清理Eden和刚使用过的Survivor。HotSpot虚拟机Eden和Survivor的大小比例默认为8:1，也就是每次使用的空间大小是90%，只“浪费”了10%。但是98%的对象回收也不能保证，每次存活的对象所使用的空间小于10%，所以当Survivor空间不够时，就需要其他空间（一般是老年代）进行分配担保，如果另一块Survivor空间没有足够的空间存放上一次新生代垃圾回收存活的对象时，这些对象将直接通过分配担保机制进入老年代。 标记-整理算法 复制算法适用于新生代的对象“朝生夕死”，如果一个区域内的对象老是不死，不仅内次都需要复制大量的对象，效率很低，而且还需要额外的空间进行担保（分配担保是一个很复杂的东西，今后有机会会说到），所以对于老年代的对象，这种算法是不适合的，于是就提出了标记-整理算法。 标记-整理算法和标记-清除算法一样，也是分两个阶段，而且第一个阶段也一样，都是标记，所不同的是第二个阶段，不是对可回收的对象进行直接清理，而是让还存活着的对象向一端移动，然后直接清理掉端边界以为的内存空间，该算法的执行过程如下图： 分代收集算法 当前的商业JVM都是采用的这种算法，其实这种算法并没有什么新思想，而是根据对象存活周期的不同将内存划分几块，一般是把Java堆分为两块：新生代和老年代，这样就可以根据各个代的不同特点采取最适当的垃圾收集算法。新生代的对象大多都是“朝生夕死”的，而且还可以有老年代进行担保，那就采用复制算法，只需要付出少量的存活对象的复制成本就可以完成收集，而且一般也不需要启用担保策略，而老年代的对象存活率一般比较高、没有空间进行担保，就只有采用“标记-清理”或者“标记-整理”算法来进行回收了。 参考资料：周志明《深入理解Java虚拟机》第二版第三章

差不多两年以前曾经写过一篇文章：JAVA 性能调优，其实在那篇文章中只是简单的说了，对象的分布。这篇文章继续对分布于堆中的对象的生命周期进行说明，也就是确定堆中的这些对象哪些还是“活着”的，哪些是已经“死去”（即不可能再被任何途径使用的对象）的。 引用计数算法 有很多人认为判断对象是否活着的算法是这样的：给对象添加一个引用计数器，每当有一个地方引用他的时候，计数器就加1，引用失效的时候，计数器减1，当计数器的数值为0时就是不可能在被引用的对象，此时就就可以认为是已死的对象。引用计数器算法实现简单，效率也很高，是一个不错的算法，但是主流的Java虚拟机并没有采用这种算法来管理内存，其中最主要的原因就是：它很难解决对象之间循环引用的问题。 举一个简单的例子：对象objA和objB都有字段instance，赋值令，除此之外，这两个对象再无任何引用，实际上他们已经不可能在被访问到，但是他们因为相互引用对方，计数器都不可能为0，计数器算法是无法通知GC收集器收集他们的。 package demo; /** * testGC()方法执行后，objA和objB会不会被GC呢？ * * @author BridgeLi * */ public class ReferenceCountingGC { public Object instance = null; private static final int _1MB = 1024 * 1024; // 这个成员的唯一意义就是占用内存，以便能在GC日志中看清楚是否被回收过 private byte[] bigSize = new byte[2 * _1MB]; public static void testGC() { ReferenceCountingGC objA = new ReferenceCountingGC(); ReferenceCountingGC objB = new ReferenceCountingGC(); objA.instance = objB; objB.instance = objA; objA = null; objB = null; // 假设发生了GC，看objA和objB是否能被回收 System.gc(); } public static void main(String[] args) { ReferenceCountingGC.testGC(); } } 从这个例子的运行结果来看，虚拟机并没有这两个对象存在相互引用就不收集他们，从而证明了Java虚拟机不是通过引用计数算法来判断对象是否已死的。 ...

大家都知道Redis是NoSQL的一种，目前在互联网公司中在作为缓存广泛的使用者，其实利用Redis的setnx还可以快速实现一个分布式锁，公司的业务就需要使用分布式锁保证数据的唯一性，经检索在网上发现已经有活雷锋分享了一套，本着不在重新发明轮子的想法，测试了一下好像没有问题，几乎不用对原代码进行修改，就能使用，下面就分享在这里，供需要的朋友参考。原文里面还有对实现的原理进行解释，所以本文就不再赘述，详情请通过参考资料进行访问。 package cn.bridgeli.distributedlock; import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPool; import redis.clients.jedis.JedisPoolConfig; import redis.clients.jedis.Transaction; public class RedisDistributedLock { private static final Logger LOG = LoggerFactory.getLogger(RedisDistributedLock.class); private static final String redisHost = "127.0.0.1"; private static final int port = 6381; private static JedisPoolConfig config; private static JedisPool pool; private static ExecutorService service; private static int ThLeng = 10; private static CountDownLatch latch; private static AtomicInteger Countor = new AtomicInteger(0); private static int count = 0; private static String LockName = "mylock_test10"; static { // 利用Redis连接池，保证多个线程利用多个连接，充分模拟并发性 config = new JedisPoolConfig(); config.setMaxIdle(10); config.setMaxWaitMillis(1000); config.setMaxTotal(30); pool = new JedisPool(config, redisHost, port); // 利用ExecutorService 管理线程 service = Executors.newFixedThreadPool(ThLeng); // CountDownLatch保证主线程在全部线程结束之后退出 latch = new CountDownLatch(ThLeng); } /** * 获取锁 tips：生成一个UUID，作为Key的标识，不断轮询lockName，直到set成功，表示成功获取锁。 * 其他的线程在set此lockName时被阻塞直到超时。 * * @param pool * @param lockName * @param timeouts * @return 锁标志 */ public static String getLock(JedisPool pool, String lockName, long timeouts) { Jedis client = pool.getResource(); try { String value = UUID.randomUUID().toString(); long timeWait = System.currentTimeMillis() + timeouts * 1000; while (System.currentTimeMillis() < timeWait) { if (client.setnx(lockName, value) == 1) { // 防止key卡死，一直不释放锁，所以1800s过期 client.expire(lockName, 1800); LOG.info("lock geted"); return value; } try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } LOG.info("get lock timeouts"); } finally { // pool.returnBrokenResource(client); pool.returnResource(client); } return null; } /** * 释放锁 tips：对lockName做watch，开启一个事务，删除以LockName为key的锁，删除后，此锁对于其他线程为可争抢的。 * * @param pool * @param lockName * @param value */ public static void relaseLock(JedisPool pool, String lockName, String value) { Jedis client = pool.getResource(); try { while (true) { client.watch(lockName); if (client.get(lockName).equals(value)) { Transaction tx = client.multi(); tx.del(lockName); tx.exec(); return; } client.unwatch(); } } finally { // pool.returnBrokenResource(client); pool.returnResource(client); } } public static void main(String args[]) { for (int i = 0; i < ThLeng; i++) { String tName = "thread-" + i; Thread t = new Thread(new SubAddThread(pool, tName)); LOG.info(tName + "inited&#8230;"); service.submit(t); } service.shutdown(); try { latch.await(); } catch (InterruptedException e) { e.printStackTrace(); } LOG.info(Countor.get() + ""); LOG.info(count + ""); } public static class SubAddThread implements Runnable { private String name; private JedisPool pool; public SubAddThread(JedisPool pool, String uname) { this.pool = pool; this.name = uname; } @Override public void run() { for (int i = 0; i < 100; i++) { LOG.info(name + " starting&#8230;"); String valuse = getLock(pool, LockName, 50); LOG.info(name + " get Lock " + valuse); count++; relaseLock(pool, LockName, valuse); Countor.incrementAndGet(); LOG.info(name + " " + count); } latch.countDown(); LOG.info(name + " complated"); } } } 参考资料：http://www.jianshu.com/p/c1f5d26cb1c9

去年五月份的时候曾经写过一篇：Spring加Mybatis实现MySQL数据库主从读写分离，实现的原理是配置了多套数据源，相应的sqlsessionfactory，transactionmanager和事务代理各配置了一套，如果从库或数据库有多个的时候，需要配置的信息会越来越多，远远不够优雅，在我们编程界有一个规范：约定优于配置。所以就用Sping的aop实现了一个简单的数据库分离方案，具体实现代码放在了Github上，地址如下： https://github.com/bridgeli/practical-util/tree/master/src/main/java/cn/bridgeli/datasource 读者如果想使用再简单的方法就是把这个代码download下来，放到自己的项目里面，当然更优雅的方式是：打成jar包，放到项目里面了，具体打jar的方法，老夫就不在这里多说了，相信看这篇文章的读者肯定都会了。当然仅仅有这份代码，他们是不会自动生效的，既然是使用Spring的Aop实现数据库读写分离，所以肯定会有牵涉到Aop的配置了，所以在spring-mybatis.xml中有如下配置： <?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd"> <!&#8211; 配置写数据源 &#8211;> <bean id="masterDataSource" class="com.alibaba.druid.pool.DruidDataSource" destroy-method="close"> <property name="driverClassName" value="${bridgeli.jdbc.driver}" /> <property name="url" value="${bridgeli.jdbc.url}" /> <property name="username" value="${bridgeli.jdbc.username}" /> <property name="password" value="${bridgeli.jdbc.password}" /> <property name="validationQuery" value="${bridgeli.jdbc.validationQuery}" /> <property name="initialSize" value="1" /> <property name="maxActive" value="20" /> <property name="minIdle" value="0" /> <property name="maxWait" value="60000" /> <property name="testOnBorrow" value="false" /> <property name="testOnReturn" value="false" /> <property name="testWhileIdle" value="true" /> <property name="timeBetweenEvictionRunsMillis" value="60000" /> <property name="minEvictableIdleTimeMillis" value="25200000" /> <property name="removeAbandoned" value="true" /> <property name="removeAbandonedTimeout" value="1800" /> <property name="logAbandoned" value="true" /> <property name="filters" value="stat" /> </bean> <!&#8211; 配置读数据源 &#8211;> <bean id="parentSlaveDataSource" class="com.alibaba.druid.pool.DruidDataSource" destroy-method="close"> <property name="driverClassName" value="${bridgeli.jdbc.driver}" /> <property name="validationQuery" value="${bridgeli.jdbc.validationQuery}" /> <property name="initialSize" value="1" /> <property name="maxActive" value="20" /> <property name="minIdle" value="0" /> <property name="maxWait" value="60000" /> <property name="testOnBorrow" value="false" /> <property name="testOnReturn" value="false" /> <property name="testWhileIdle" value="true" /> <property name="timeBetweenEvictionRunsMillis" value="60000" /> <property name="minEvictableIdleTimeMillis" value="25200000" /> <property name="removeAbandoned" value="true" /> <property name="removeAbandonedTimeout" value="1800" /> <property name="logAbandoned" value="true" /> <property name="filters" value="stat" /> </bean> <bean id="slaveDataSource1" class="com.alibaba.druid.pool.DruidDataSource" destroy-method="close" parent="parentSlaveDataSource"> <property name="url" value="${bridgeli_slave1.jdbc.url}" /> <property name="username" value="${bridgeli_slave1.jdbc.username}" /> <property name="password" value="${bridgeli_slave1.jdbc.password}" /> </bean> <bean id="dataSource" class="cn.bridgeli.datasource.MasterSlaveDataSource"> <property name="targetDataSources"> <map> <entry key-ref="masterDataSource" value-ref="masterDataSource"/> <entry key-ref="slaveDataSource1" value-ref="slaveDataSource1"/> </map> </property> <property name="defaultTargetDataSource" ref="masterDataSource"/> <property name="masterSlaveSelector" ref="dataSelector"/> </bean> <bean id="dataSelector" class="cn.bridgeli.datasource.MasterSlaveSelectorByPoll"> <property name="masters"> <list> <ref bean="masterDataSource"/> </list> </property> <property name="slaves"> <list> <ref bean="slaveDataSource1"/> </list> </property> <property name="defaultDataSource" ref="masterDataSource"/> </bean> <aop:aspectj-autoproxy/> <!&#8211; mybaits 数据工厂 &#8211;> <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean"> <property name="dataSource" ref="dataSource" /> </bean> <!&#8211; 自动扫描所有注解的路径 &#8211;> <bean class="org.mybatis.spring.mapper.MapperScannerConfigurer"> <property name="basePackage" value="cn.bridgeli.mapper" /> <!&#8211; <property name="sqlSessionFactory" ref="sqlSessionFactory" /> &#8211;> <property name="sqlSessionFactoryBeanName" value="sqlSessionFactory"></property> </bean> <!&#8211; 数据库切面 &#8211;> <bean id="masterSlaveAspect" class="cn.bridgeli.datasource.MasterSlaveAspect"> <property name="prefixMasters"> <list> <value>save</value> <value>update</value> <value>delete</value> </list> </property> </bean> <aop:config> <aop:aspect id="c" ref="masterSlaveAspect"> <aop:pointcut id="tx" expression="execution(\* cn.bridgeli.service..\*.*(..))"/> <aop:before pointcut-ref="tx" method="before"/> </aop:aspect> </aop:config> <context:annotation-config /> <context:component-scan base-package="cn.bridgeli" /> </beans> 这样我们就很优雅的利用Spring的Aop实现了对数据库的读写分离，读的时候走slaveDataSource1这个数据源，写的时候走masterDataSource这个数据源。哎，等等，这里哪里体现了约定优于配置这一规范，他们怎么知道哪些方法走读库哪些走写库？同学你别急，仔细读读这个配置文件，你就会发现在第98行，配置了一个MasterSlaveAspect，也就是说代码里面service层（为什么是service层？）的方法以这里面配置的这些关键字打头，都将会走写库，所以当我们想让一个方法走主库的时候，必须在这个地方添加该方法的前缀或者用这里面已有的前缀，这就要求我们必须约定好走主库的方法的打头，即约定优于配置。

前几天网上突然出现流言：某东发生数据泄露12G，最终某东在一篇声明中没有否认，还算是勉强承认了吧，这件事对于一般人有什么影响、应该怎么做已经有一堆人说了，所以就不凑热闹了，咱来点对程序猿来说实际点的，说一个个人认为目前比较安全的加密算法：Blowfish。 上代码之前，先说几点Blowfish加密算法的特点： 对称加密，即加密的密钥和解密的密钥是相同的； 每次加密之后的结果是不同的（这也是老夫比较欣赏的一点）； 可逆的，和老夫之前的文章介绍的md5等摘要算法不一样，他是可逆的； 速度快，加密和解密的过程基本上由ADD和XOR指令运算组成； 免费，任何人都可以免费使用不需要缴纳版权费； BlowFish 每次只能加密和解密8字节数据； 接下来就是最重要的部分，Blowfish加密算法的实现： package cn.bridgeli.encrypt; public enum BlowfishManager { BRIDGELI_CN("bridgeli_cn!@#$abc123_"); private BlowfishManager(String secret) { this.blowfish = new Blowfish(secret); } private Blowfish blowfish; public Blowfish getBlowfish() { return blowfish; } /** * 解密 * @param sCipherText * @return */ public String decryptString(String sCipherText){ return this.getBlowfish().decryptString(sCipherText); } /** * 加密 * @param sPlainText * @return */ public String encryptString(String sPlainText){ return this.getBlowfish().encryptString(sPlainText); } public static void main(String[] args) { String encryptString = BlowfishManager.BRIDGELI_CN.encryptString(10 + ""); System.out.println(encryptString); String decryptString = BlowfishManager.BRIDGELI_CN.decryptString(encryptString); System.out.println(decryptString); } } 这是对外的接口，使用起来非常简单，对用户很友好，下面是算法的具体实现： ...

公司的RPC的服务使用的是阿里巴巴的dubbo，老夫之前曾经写过一篇如何在测试环境远程调试dubbo服务，详情请参考这篇，但一直对如何调试线上dubbo服务不得法，不得已每次都需要写一个web服务调一下看数据，前一段时间经新来的一个同事提示可以使用Telnet调试，网上搜了一下资料，发现真的很爽，以下是学习笔记。 需要说明的是：Dubbo2.0.5以上版本服务提供端口支持telnet命令，不过应该没有公司使用2.0.5以下版本吧。 进入调试模式 telnet localhost 20880 即：telnet + ip + 端口，这个不用解释，使用dubbo的肯定都知道 ls 使用上一个命令之后，敲一下回车，就进入dubbo的telnet调试服务了，然后就可以使用ls命令了，这个命令有几个用法： ①. 显示服务列表 ls ②. 显示服务详细信息列表 ls -l ③. 显示服务的方法列表 ls XxxService ④. 显示服务的方法详细信息列表 ls -l XxxService ps 这个命令主要是看连接信息，也有如下几个用法： ①. 显示服务端口列表 ps ②. 显示服务地址列表 ps -l ③. 显示端口上的连接信息 ps 20880 ④, 显示端口上的连接详细信息 ps -l 20880 cd 这个缺省服务，主要有以下两个用法： ①. 改变缺省服务，当设置了缺省服务，凡是需要输入服务名作为参数的命令，都可以省略服务参数 cd XxxService ②. 取消缺省服务 cd / pwd 显示当前缺省服务 trace 这个命令顾名思义：跟踪，具体有以下用法： ①. 跟踪1次服务任意方法的调用情况 trace XxxService ②. 跟踪10次服务任意方法的调用情况 ...

上个周苹果公司悍然发布了新版mac，消灭了功能键，包括ESC，终于使下面这个段子成为了事实：问，如何生成一段随机数？答：让一个非开发人员退出vim。哈哈，现在开发人员是不是也可以产生随机字符串了？发现自己作为一个vim党，竟然对很多vim常用的命令都不知道，今天就记一下笔记，让自己这个vim党称呼实至名归。 首先要说明的是，基本的vim命令像A、I、O进入编辑模式，ESC进入命令模式，“:”进入末行模式，以及常用的什么dd，yy，p等都认为大家已经熟练掌握，就不说了。 替换字符 :%s/oldchar/newchar/g 这个命令同样可以解决： 注1. 在windows记事本下的文件放到Linux下时，行末多出来一个^M，这个问题，直接把oldchar换成\r，newchar传承空就可以了。 注2. 在windows记事本下的文件放到Linux下时，行末多出来一个^@，这个问题，直接把oldchar换成先摁ctrl+v，然后摁ctrl+2，newchar传承空就可以了。 注3. 在windows记事本下的文件放到Linux下时，行末多出来一个^A，这个问题，直接把oldchar换成先摁ctrl+v，然后摁ctrl+A，newchar传承空就可以了。 注4. oldchar也可以用正则表达式，之前一直不知道怎么在每一行的行末加东西，其实如此简单而已。 加密文件 进入末行模式，然后输入大写的X，然后输入密码，保存退出即可，这样的话今后每次打开都需要输入密码才行，否则就是一堆乱码。 undo和redo 这个比较简单，undo直接摁u，redo是ctrl+r 简单的移动光标 0 数字零，到行头 ^ 到本行第一个不是blank字符的位置 $ 到本行行尾 g_ 到本行最后一个不是blank字符的位置 /pattern 搜索 pattern 的字符串,如果搜索出多个匹配，可按n键到下一个 . (小数点) 可以重复上一次的命令 N<command> 重复某个命令N次 :N 到第N行 gg 到第一行。（陈皓注：相当于1G，或 :1） G 到最后一行 块操作: ctrl-v 块操作，典型的操作： 0、ctrl-v、ctrl-d、I、ESC ^ 到行头 ctrl-v 开始块操作 ctrl-d 向下移动 (你也可以使用hjkl来移动光标，或是使用%，或是别的) I 插入，然后输入 ESC 来为每一行生效。 自动提示 在输入模式下，你可以输入一个词的开头，然后按 ctrl-p或是ctrl-n，自动补齐功能就出现了 可视化选择： v,V,ctrl-v ctrl-v，我们可以使用 v 和 V。一但被选好了，你可以做下面的事： J 把所有的行连接起来（变成一行） < 或 > 左右缩进 = 自动给缩进 窗口分屏浏览 :He 在下边分屏浏览目录 :He! 在上分屏浏览目录 :Ve 在左边分屏间浏览目录 :Ve! 要在右边则是 多页签（tab page） 在末行模式下，输入： ...

今年年中的时候由公司平台部转组到Alpha项目中心负责公司一拍项目组的技术研发工作，到现在已经快有将近半年的时间了，随着对系统的越来越熟悉，对原有系统的架构也越来越感到有些不合理的地方，随着自己水平的提升感觉对架构也有了一点自己的理解，所以今天就借这个机会说说自己的不成熟的建议。 一. 原有的架构 俗话说，一图胜千言，直接上图： 解释一下这几个系统分别的作用： 后台管理系统不用说了，管理C和B可见的内容； C端用户系统，是对C可见的一个系统，一拍是一个招聘系统，所以就是对候选人操作的后台； B端用户系统，是对B可见的一个系统，通俗点讲就是HR操作的后台； Dubbo系统，是对兄弟部门和B提供服务的一个系统； Recommend系统，其实也是一个Dubbo系统，区别在于Dubbo系统是对外提供各种服务的，而Recommend系统是发现全站系统用户的行为，然后对用户的行为进行分析，甄选出一部分C端用户作为一拍的现在用户； msgpush系统，是用netty做的一个实时消息推送的IM服务，目前主要是给后台管理用户和C端候选人聊天的一个系统； 其中：1. 在我们接手之前B和C是同一个系统，也就是说B是我们这半年新加的一个系统；2. Recommend和msgpush系统也同样是我们这半年新增的一个系统 二. 系统架构存在的问题 目前后台管理系统、C端用户系统、Dubbo系统各自独立，这样存在的问题： 最主要的是各自分别操作数据库，这样只要底层数据库发送变动，那么三个系统操作数据库的地方都要同步修改三次； 操作数据库的地方代码冗余，很多地方都一样，这样一个地方出bug，三个地方要同步修改，然后都要上线； 后台管理系统采用分层的模式分模块而不是根据业务分模块，这样每次上线service和dao都要先deploy jar到maven私服； 当时为了快速迭代，Recommend系统也是单独操作操作数据库，不过还好用了后台管理系统的dao这个jar包，但是首先根本不需要这么重的一个jar包，其次jar出bug了，有时候他也需要重新上线啊，不然这个jar包就会一直很旧，当然只要不涉及到他操作数据库的地方出bug，你不改也是可以的； 目前C和B虽然已经分开，但如果用户激增，横向扩展依然不合理，只能整体加机器，而不能针对性对某些模块单独加机器； 代码中存在的问题：不知道什么原因大量的逻辑被写在了controller层，导致代码可复用性差； 由于之前后台管理系统和其他系统不是同一个团队开发的，命名各有各的风格，代码不仅冗余还同样一个类名字不一样； 很多系统日志配置的也有问题，错误日志和最基本的业务日志没有区分开，目前在将就用； 系统中的jar依赖不仅存在循环依赖，而且加入了大量的自己不需要的依赖，导致各种jar冲突出问题； 综合以上问题，我个人认为这是一个：可维护性、可扩展性不高的系统。 三. 我的个人思考 同样先来一张图，来总体说明一下我的想法： 整体来说只有相对独立的实时消息推送系统不懂，然后把其余的各个业务层抽象成微服务，采用公司目前使用比较成熟的dubbo作为rpc框架，controller层只负责业务转发不负责逻辑的一个简单系统，这样带来的好处： Dubbo系统作为核心的业务系统，分别对兄弟部门、后台管理系统、C端用户系统、B端用户系统、Recommend系统提供服务，如果用户量增加，不仅可以整体增加Dubbo系统的机器，也可以把调用量大的接口单独拆出来，部署到另外的机器上，实现隔离，不会因某个接口调用量大，导致整个系统不可用，而且把数据库的底层操作也放到了这个Dubbo系统中，这样就可以避免数据库修改，要修改多处的问题； 抽象出来的这个dubbo系统，不仅可以解决后台系统和其他系统命名不一样的问题，而且和可以解决不同团队造成的coding style不一的问题，一举多得； 由于后台管理系统、C端用户系统、B端用户系统、Recommend系统都调用Dubbo系统，所以他们的controller层讲极其简单，很多业务逻辑类似的东西全部放到了Dubbo系统里面，代码的可复用性提高了不少； 抽象出来的逻辑都统一放到了dubbo中，这样系统如果有bug，这样就做到了一个地方修改，这样多个地方就可以同时生效； 由于业务都在Dubbo系统里面这样同时也避免了曾经出现了，后台管理系统和BC系统使用的缓存不一致，导致缓存出问题的这种低级bug； 由于controller层简单没有逻辑，这样就可以避免目前由于后台管理系统单机，修改一个业务逻辑bug重启系统，导致后台不可用的问题，因为只需要重启dubbo就行了； 这样controller层变得很轻，只需要一个简单的servlet容器，对机器的要求会降低不少； 四. 备注 由于我工作时间不长，见过的系统更有限，所以对系统架构几乎没有什么经验，这些只是我个人的一点很粗浅的理解，例如把dubbo做的那么重，虽然可扩展性提高了不少，但其实也不知道算不算合理，因为调用rpc服务，肯定会增加网络IO延时，所以这些算是我个人的抛砖引玉吧，一方面希望对同样和我一样没有经验的小伙伴能有所帮助，另一方面希望有经验的小伙伴能留言交流 五. 总结 以上便是我个人对拉勾一拍所有的核心系统进行了审视后的一番分析，如果这些核心系统架构的重构真的达到自己的理想状况这将是一番浩大的工程，对于高速发展的互联网公司来说，这就是一边驾驶者一辆高速前进的汽车，一边对这辆汽车进行换轮胎换发动机，先不说工作量的问题，难度程度也可见一斑。

上一篇文章讲了redis的主从搭建，主从一般只能解决我们读写分离的问题，可以增加我们的系统的负载能力，但是并不能解决单点问题，大家应该知道在互联网公司各个服务肯定不能出现单点问题，所以这一节就记录一下如果让我们的系统更加高可用。 一、集群搭建 需要先说明的是，集群搭建需要至少6个节点：3主3从(因为没有那么多机器，所以就在一台上搞了) 创建文件夹redis-cluster，然后在其下面分别创建6个文件夹，存放6个实例 mkdir -p /usr/local/redis-cluster mkdir 7001;mkdir 7002;mkdir 7003;mkdir 7004;mkdir 7005;mkdir 7006 把之前redis.conf配置文件分别copy到700*下，修改各个实例的配置文件的内容，如下： ①. daemonize yes ②. port 700*（一台机器端口号肯定不能相同，就和文件夹一样吧） ③. bind ip（和当前机器的ip地址绑定） ④. dir /usr/local/redis-cluster/700*/（文件存储位置应该不一样吧，原因都知道） ⑤. cluster-enabled yes ⑥. cluster-config-flie nodes-700\*.conf（700\*就和端口一样吧） ⑦. cluster-node-timeout 5000 ⑧. appendonly yes 安装ruby yum install ruby yum install rubygems gem install redis 启动各个实例 /usr/local/redis/bin/redis-server /usr/local/redis-cluster/700*/redis.conf 创建集群 cd /usr/local/redis-3.0.0-rc2/src ./redis-trib.rb create &#8211;replicas 1 ip:7001 ip:7002 ip:7003 ip:7004 ip:7005 ip:7006 然后看输出日志，会有一步需要输入yes，然后集群就创建完成了 ...