基于spring-jdbc中JdbcTemplate实现查询高可用

在传统系统或网站架构中,一般都是使用单点,当然单点在应对并发访问量不是很大的场景下是能够应对自如的,并且单机不存在服务延迟,分布式事务等问题,部署也比较简单。

但是随着互联网的发展,传统的架构已经远远满足不了庞大的并发访问量,并且单机所带来的问题也日益显著,比如:1)系统挂掉所有用户都无法访问 2)当瞬时访问量比较大的

时候单机部署的站点,单纯的通过硬件的扩充已经解决不了性能瓶颈问题;现在比较流行的就是SOA或者微服务,归根结底其目的就是解决单机情况下的系统可用性,可靠性以及性能问题,这里简单描述一下从请求接受层到数据库层的通用架构方案:

1.反向代理层:比较常用的就是nginx做反向代理和负载均衡

2.站点层:tomcat做集群

3.服务层:各种rpc框架都提供了负载均衡(dubbo)

4.数据库层:大多使用DB cache方式,DB的话又分很多种主从复制方式(一主多从,双主以及其他数据库中间件实现的读写分离方案)

本篇文章对其他问题不做赘述,简单对读写分离中的多个从库实现负载均衡访问和提高查询高可用(基于JdbcTemplate实现),实现步骤如下

一、创建maven项目并添加最简依赖

1. 新建maven项目,项目结构如下图:

2. 添加依赖

<groupId>junit</groupId>

<artifactId>junit</artifactId>

</dependency>

<groupId>mysql</groupId>

<artifactId>mysql-connector-java</artifactId>

</dependency>

<groupId>org.springframework</groupId>

<artifactId>spring-jdbc</artifactId>

<version>4.3.7.RELEASE</version>

</dependency>

<groupId>org.springframework</groupId>

<artifactId>spring-context</artifactId>

<version>4.3.7.RELEASE</version>

</dependency>

<groupId>com.alibaba</groupId>

<artifactId>druid</artifactId>

</dependency>

<groupId>com.alibaba</groupId>

<artifactId>fastjson</artifactId>

</dependency>

</dependency>

<groupId>org.apache.commons</groupId>

<artifactId>commons-lang3</artifactId>

</dependency>

二、添加配置文件

添加jdbc属性配置(jdbc.properties)

jdbc.trace=true

# 写库

jdbc.crmupdate.database=10.11.9.243:3306/test

jdbc.crmupdate.user=crm

jdbc.crmupdate.password=USszJ497whda

# 读库1

jdbc.crmquery1.database=10.11.9.243:3306/test

jdbc.crmquery1.user=crm

jdbc.crmquery1.password=USszJ497whda

# 读库2

jdbc.crmquery2.database=10.1.8.76:3306/test

jdbc.crmquery2.user=root

jdbc.crmquery2.password=root

2. 添加日志配置(log4j.properties)

### 设置###

log4j.rootLogger = info,stdout,D,E

### 输出信息到控制抬 ###

log4j.appender.stdout = org.apache.log4j.ConsoleAppender

log4j.appender.stdout.Target = System.out

log4j.appender.stdout.layout = org.apache.log4j.PatternLayout

log4j.appender.stdout.layout.ConversionPattern = [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n

### 输出DEBUG 级别以上的日志文件设置 ###

log4j.appender.D = org.apache.log4j.DailyRollingFileAppender

log4j.appender.D.File = logs/jdbcTemplateHA_debug.log

log4j.appender.D.Append = true

log4j.appender.D.Threshold = DEBUG

log4j.appender.D.layout = org.apache.log4j.PatternLayout

log4j.appender.D.layout.ConversionPattern = %-d{yyyy-MM-dd HH:mm:ss} [ %t:%r ] - [ %p ] %m%n

### 输出ERROR 级别以上的日志文件设置 ###

log4j.appender.E = org.apache.log4j.DailyRollingFileAppender

log4j.appender.E.File = logs/jdbcTemplateHA_error.log

log4j.appender.E.Append = true

log4j.appender.E.Threshold = ERROR

log4j.appender.E.layout = org.apache.log4j.PatternLayout

log4j.appender.E.layout.ConversionPattern = %-d{yyyy-MM-dd HH:mm:ss} [ %t:%r ] - [ %p ] %m%n

3. 添加spring数据源配置(spring-jdbc.xml)

</bean>

<list>

</list>

</property>

</bean>

<value>jdbc:mysql://${jdbc.crmupdate.database}?useUnicode=true&characterEncoding=gbk&zeroDateTimeBehavior=convertToNull</value>

</property>

</bean>

<value>jdbc:mysql://${jdbc.crmquery1.database}?useUnicode=true&characterEncoding=gbk&zeroDateTimeBehavior=convertToNull</value>

</property>

</bean>

<value>jdbc:mysql://${jdbc.crmquery2.database}?useUnicode=true&characterEncoding=gbk&zeroDateTimeBehavior=convertToNull</value>

</property>

</bean>

</bean>

</bean>

</bean>

三、编写实现查询HA的相关类

从图中可以看到,实现HA主要有上边标注的几个类,下边将逐个描述各个类的作用

1. BalanceNodeBean

该类是实现负载均衡的bean

2. NodeHealthValueBean

实现健康值检查的bean,

3. MyLoadBalance

根据可用JdbcTemplate列表实现负载均衡访问的工具类,代码如下:

/**

* 负载均衡器

* (1)支持各节点间的负载均衡

* (2)支持每个节点的权重设定

* (3)支持单个节点的动态故障隔离和故障恢复

* (4)支持每个节点的动态权重修正系数设定

* @author Typhoon

* @date 2017-09-17 18:39 Sunday

* @since V2.0

public class MyLoadBalance {

/**

* 取得将使用的 BalanceNode index

* @param serverList serverList

* @return 将使用的 BalanceNode index

public static int getForwardServerIndex(

ArrayList<BalanceNodeBean> serverList) {

// check para

if(CollectionUtils.isEmpty(serverList)) {

return -1;

}

int len = serverList.size();

int result = -1;

if (len == 1) {

// (2)仅有一台server时

if (serverList.get(0).getDownFlag() == 1) {

// 无可用的server

result = -1;

} else {

// 仅这一台

result = 0;

}

return result;

}

// (3)有多台后端服务器

BalanceNodeBean bestBean = null;

BalanceNodeBean tempBean = null;

float currentWeight = 0;

// 所有服务器总权重

float totalEffectiveWeight = 0;

// 遍历所有服务器, 按照各台服务器的当前权值进行选择

for (int i = 0; i < len; i ) {

tempBean = serverList.get(i);

if (tempBean.getDownFlag() == 1) {

// Down掉的server,将被skip.

continue;

}

// 计算当前权值，包含权重修正系数

currentWeight = tempBean.getCurrentWeight() tempBean.getWeight()

* (1 tempBean.getCorrectionWeight());

tempBean.setCurrentWeight(currentWeight);

// 总的权重计算，包含权重修正系数

totalEffectiveWeight = tempBean.getWeight()

* (1 tempBean.getCorrectionWeight());

// 选择

if (bestBean == null

|| bestBean.getCurrentWeight() < tempBean

.getCurrentWeight()) {

// 选择当前权重最大的serverNode

bestBean = tempBean;

result = i;

}

if (bestBean == null) {

// 无可用的server

return -1;

}

if (bestBean != null) {

// 该服务器这次被选中，因此要降低权值,以便下次计算。

currentWeight = bestBean.getCurrentWeight() - totalEffectiveWeight;

bestBean.setCurrentWeight(currentWeight);

}

return result;

}

4. HABalanceSimpleJdbcOperations

和调用方直接打交道的类,实现了JdbcOperations接口,提供了和JdbcTemplate一样的功能(具体操作还是委托给具体的jdbcTemplate去完成),同时实现了PropertyChangeListener接口,对JdbcTemplate列表对应的健康值监听并完成健康值矫正操作.代码如下:

/**

* 高可用jdbcOperations

* @author Typhoon

* @date 2017-09-17 13:54 Sunday

* @since V2.0

public class HABalanceSimpleJdbcOperations implements PropertyChangeListener, JdbcOperations {

/**

* LOG

private static Logger log = Logger.getLogger(HABalanceSimpleJdbcOperations.class);

// private static org.slf4j.Logger log = LoggerFactory.getLogger(HABalanceSimpleJdbcOperations.class);

/**

* DB从库server列表

private ArrayList<BalanceNodeBean> serverList = new ArrayList<BalanceNodeBean>();

/**

* DB从库server列表用的readWriteLock

private ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();

/**

* 备机列表(查询库)

protected List<JdbcOperations> crmB_N_JdbcList;

/**

* 主机ip(写库)

protected JdbcOperations crmJdbc;

/**

* 构造器

* @param crmB_N_JdbcList

public HABalanceSimpleJdbcOperations(

List<JdbcOperations> crmB_N_JdbcList,

JdbcOperations crmJdbc) {

this.crmB_N_JdbcList = crmB_N_JdbcList;

this.crmJdbc = crmJdbc;

// 生成相应的serverList

int len = this.crmB_N_JdbcList.size();

BalanceNodeBean nodeBean = null;

for (int i = 0; i < len; i ) {

nodeBean = new BalanceNodeBean();

// 初始状态为正常

nodeBean.setDownFlag(0);

nodeBean.setCorrectionWeight(0);

nodeBean.setCurrentWeight(0);

// add

this.serverList.add(nodeBean);

}

// Thread数

int threadNum = 1;

// 新建一个Threadpool

ExecutorService threadPool = Executors.newFixedThreadPool(threadNum);

HABalanceJdbcHealthCheckTask task = null;

for (int i = 0; i < threadNum; i ) {//使用threadNum个线程去监听jdbcTemplate列表的健康状态,如果出现健康值变动,触发监听器修改节点的健康状态

// 新作成一个Task

task = new HABalanceJdbcHealthCheckTask(this.crmB_N_JdbcList);

// add PropertyChangeListener

task.addPropertyChangeListener(this);

// 提交task到Threadpool

threadPool.execute(task);

}

/**

* healthyValue属性改变通知处理。根据healthyValue修改节点的DownFlag和权重修正系数。

public void propertyChange(PropertyChangeEvent event) {

if (event.getPropertyName().equalsIgnoreCase("healthyValue")) {

Object newBean = event.getNewValue();

if (newBean instanceof NodeHealthValueBean) {

NodeHealthValueBean valueBean = (NodeHealthValueBean) newBean;

int nodeIndex = valueBean.getNodeIndex();

int healthValue = valueBean.getHealthValue();

if (healthValue < 0) {

// 故障发生

this.setNodeDownFlag(nodeIndex, 1);

} else {

// 故障恢复

this.setNodeDownFlag(nodeIndex, 0);

}

// 权重修正系数

float correctionWeight = 0;

if (healthValue == 0) {

// 设置节点的权重修正系数为0

correctionWeight = 0;

} else if (healthValue == 1) {

// 设置节点的权重修正系数为 -0.1

correctionWeight = -0.1f;

} else if (healthValue == 2) {

// 设置节点的权重修正系数为 -0.3

correctionWeight = -0.3f;

} else if (healthValue == 3) {

// 设置节点的权重修正系数为 -0.5

correctionWeight = -0.5f;

} else if (healthValue == 4) {

// 设置节点的权重修正系数为 -1.0

correctionWeight = -1.0f;

}

// 设置节点的权重修正系数

this.setNodeCorrectionWeight(nodeIndex, correctionWeight);

// log

log.warn("nodeIndex=" nodeIndex ",healthValue="

healthValue ",correctionWeight=" correctionWeight);

}

/**

* 设置节点的downFlag

* @param nodeIndex

* @param downFlag

public void setNodeDownFlag(int nodeIndex, int downFlag) {

// check para

if (nodeIndex < 0 || nodeIndex >= this.serverList.size()) {

return;

}

Lock writeLock = this.readWriteLock.writeLock();

writeLock.lock();

try {

BalanceNodeBean nodeBean = this.serverList.get(nodeIndex);

// set downFlag

nodeBean.setDownFlag(downFlag);

} finally {

writeLock.unlock();

}

/**

* 设置节点的权重修正系数

* @param nodeIndex

* @param downFlag

public void setNodeCorrectionWeight(int nodeIndex, float correctionWeight) {

// check para

if (nodeIndex < 0 || nodeIndex >= this.serverList.size()) {

return;

}

Lock writeLock = this.readWriteLock.writeLock();

writeLock.lock();

try {

BalanceNodeBean nodeBean = this.serverList.get(nodeIndex);

// set correctionWeight

nodeBean.setCorrectionWeight(correctionWeight);

} finally {

writeLock.unlock();

}

/**

* 使用负载均衡器取得一个可用JdbcOperations。

* @return

protected JdbcOperations getJdbcOperations() {

JdbcOperations result = null;

int resultIndex = -1;

Lock readLock = this.readWriteLock.readLock();

// lock

readLock.lock();

try {

// 取得一个server index

resultIndex = MyLoadBalance.getForwardServerIndex(this.serverList);

//log.warn("resultIndex=" resultIndex);

} finally {

// unlock

readLock.unlock();

}

log.info("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<获取的负载均衡jdbcOperations节点resultIndex:" resultIndex);

if (resultIndex >= 0) {

result = this.crmB_N_JdbcList.get(resultIndex);

}

return result;

}

@Override

public <T> List<T> query(String sql, RowMapper<T> rowMapper)

throws DataAccessException {

JdbcOperations jdbcOperation = getJdbcOperations();

//log.debug("当前jdbc信息:{}",ToStringBuilder.reflectionToString(jdbcOperation));

return jdbcOperation.query(sql, rowMapper);

}

@Override

public int update(String sql) throws DataAccessException {

return crmJdbc.update(sql);

}

由于实现了JdbcOperations接口,其定义的所有方法都要实现,此处忽略;另外可以观察到所有的读操作都是操作从库,写操作直接指定主库,从而实现了读写分离的操作.

5. HABalanceJdbcHealthCheckTask

该类实现了Runnable接口,可以独立运行,其作用是监听从库相关jdbcTemplate列表的健康值,如果有变化或者某个从库挂掉会通知其监听器修改jdbcTemplate节点的健康值,代码如下:

/**

* 高可用DB从库HealthCheck Task

* @author Typhoon

* @date 2017-09-17 14:04 Sunday

* @since V2.0

public class HABalanceJdbcHealthCheckTask implements Runnable {

/**

* LOG

private static Logger log = Logger.getLogger(HABalanceJdbcHealthCheckTask.class);

/**

* DB从库Jdbc列表

private List<JdbcOperations> crmB_N_JdbcList;

/**

* Node健康值Bean列表

private ArrayList<NodeHealthValueBean> healthyValueList = new ArrayList<NodeHealthValueBean>();

/**

* 运行中标志

private volatile boolean runFlag = true;

/**

* 健康值 PropertyChangeSupport

protected PropertyChangeSupport listeners = new PropertyChangeSupport(this);

public HABalanceJdbcHealthCheckTask(List<JdbcOperations> crmB_N_JdbcList) {

// DB从库Jdbc列表

this.crmB_N_JdbcList = crmB_N_JdbcList;

int len = this.crmB_N_JdbcList.size();

NodeHealthValueBean nodeBean = null;

for (int i = 0; i < len; i ) {

nodeBean = new NodeHealthValueBean();

nodeBean.setNodeIndex(i);

// 初始状态为正常

nodeBean.setHealthValue(0);

// add

this.healthyValueList.add(nodeBean);

}

/**

* 添加健康值PropertyChangeListener

* @param listener

public void addPropertyChangeListener(PropertyChangeListener listener) {

this.listeners.addPropertyChangeListener(listener);

}

/**

* 移除健康值PropertyChangeListener

* @param listener

public void removePropertyChangeListener(PropertyChangeListener listener) {

this.listeners.removePropertyChangeListener(listener);

}

/**

* 取得健康值

* @param jdbcOpers

* @return -1: 表示已经Down; 0-4级;

private int getHealthyValue(JdbcOperations jdbcOpers) {

int result = 0;

try {

long beforeMS = System.currentTimeMillis();

// query 主要用来做DB存活性以及性能探测

jdbcOpers.queryForObject("select 1 from dual", Integer.class);

long afterMS = System.currentTimeMillis();

// diff

long diff = afterMS - beforeMS;

// log

// log.warn("diff=" diff);

if (diff >= 0 && diff <= 500) {// 小于500毫秒为0级

// 执行时长0-500ms内为正常

result = 0;

} else if (diff > 500 && diff <= 1000) {// 500毫秒到1秒之间是1级

// 执行时长500-1000ms内为降1级

result = 1;

} else if (diff > 1000 && diff <= 3000) {// 1秒到3秒之间是2级

result = 2;

} else if (diff > 3000 && diff <= 5000) {// 3秒到5秒之间是3级

result = 3;

} else if (diff > 5000) {// 大于5秒是4级

result = 4;

}

} catch (Exception e) {// 如果出现异常,认为DB挂掉

result = -1;

}

return result;

}

/**

* 检测所有Node的健康值(5秒钟检查一下 )

public void run() {

while (this.runFlag) {

try {

int len = this.crmB_N_JdbcList.size();

JdbcOperations jdbcOpers = null;

NodeHealthValueBean nodeBean = null;

int healthyValue = -1;

for (int i = 0; i < len; i ) {

jdbcOpers = this.crmB_N_JdbcList.get(i);

healthyValue = getHealthyValue(jdbcOpers);// 获得jdbc节点健康值

nodeBean = this.healthyValueList.get(i);// 健康节点

if (nodeBean.getHealthValue() == healthyValue) {// 上一次节点健康值和当前健康值一致,不需要变动

continue;

}

//节点健康值有变化

NodeHealthValueBean oldValue = new NodeHealthValueBean();//节点旧健康值

oldValue.setNodeIndex(i);

oldValue.setHealthValue(nodeBean.getHealthValue());

NodeHealthValueBean newValue = new NodeHealthValueBean();//节点新健康值

newValue.setNodeIndex(i);

newValue.setHealthValue(healthyValue);

PropertyChangeEvent event = new PropertyChangeEvent(this, "healthyValue", oldValue, newValue);

// fire 健康值变化event

this.listeners.firePropertyChange(event);//触发监听器操作

// set new healthvalue

nodeBean.setHealthValue(healthyValue);//设置节点的新健康值

// set lastUpdate

nodeBean.setLastUpdate(new Date());

}

} catch (Exception e) {

log.warn(e.getMessage());

}

try {

// sleep 5 秒

Thread.sleep(5000);

} catch (Exception e) {

log.error("exception",e);

}

可以看到该类定义了监听管理器,再run方法中5秒执行一次jdbcTemplate节点健康性检查,如果发现当前健康值和节点之前健康值不一致,会触发监听器通知其修改节点健康值属性

四、添加HA相关配置

在spring-jdbc.xml中添加如下配置

<constructor-arg index="0">

<list>

</list>

</constructor-arg>

<constructor-arg index="1" ref="crmUpdate"/>

</bean>

该案例中我们忽略更新操作;在HABalanceSimpleJdbcOperations构造器中根据jdbcTemplate列表构造出对应的用于实现负载均衡的BalanceNodeBean列表(默认节点不是挂掉,当前权重0,矫正权重0),然后由一个线程池创建指定数量的线程是监视检查jdbcTemplate节点的健康情况,并且为每一个task添加监视器(实现任何一个监视线程监测到jdbcTemplate健康值有变化,就会通知监听器修改健康值),为了不增加过多类和逻辑,HABalanceSimpleJdbcOperations也是task的监听类(触发逻辑propertyChange)

五、测试

HA功能和配置都写好了,接着我们需要些代码来验证

两个从库中test库添加user表并插入两条记录

从库1:

从库2:

两台机器上的test库中都有user表,有两条相同的记录

2. 编写业务类

接口:

public interface UserService {

List<User> selectAll();

}

接口默认实现:

@Service("userService")

public class UserServiceImpl implements UserService {

@Resource

UserDao userDao;

@Override

public List<User> selectAll() {

return this.userDao.selectAll();

}

数据交互dao操作类:

/**

* 用户信息操作dao

* @author Typhoon

@Repository("userDao")

public class UserDao {

@Resource

JdbcOperations crmBalanceJdbc;

public List<User> selectAll() {

String sql = " select * from user ";

return this.crmBalanceJdbc.query(sql, BeanPropertyRowMapper.newInstance(User.class));

}

可以看到UserDao中的JdbcTemplate已经不是之前用到的了,而是我们自己定义的类型HABalanceSimpleJdbcOperations,然后注入后就可以像其他JdbcTemplate或者NamedJdbcTemplate一样操作数据库了.

3. 编写单元测试类

使用main方法借助SpringContextUtils来实现单元测试,代码如下:

public class HAJdbcConsumer {

public static void main(String[] args) {

new ClassPathXmlApplicationContext("spring-root.xml").start();

UserService userService = SpringContextUtil.getBean("userService", UserService.class);

List<User> list = null;

for (int i = 0; i < 10; i ) {

list = userService.selectAll();

System.out.println(JSON.toJSONString(list));

}

如上述代码所示,我们获取userService后,执行了十次查询,然后我们运行程序得到如下结果:

细心的人发现了,我们确实查询到了我们想要的结果,但是完全看不出来十次查询分别是从哪个jdbcTemplate节点查询的,很好办,刚开始我们定义的日志就派上用场了,

我们查看获取jdbcTemplate节点的代码:

在根据权重规则获取到权重最高的jdbcTemplate节点的时候会计算处其所在列表的索引,我们在计算出所有后使用日志打印索引的值,也就是jdbcTemplate对应的节点索引值,如红色标注部分,接着我们再次运行单元测试,可以看到如下结果:

可以看到我们每次查询操作都打印出了获取到的jdbcTempalte索引值,这样我们就基于spring-jdbc中的jdbcTemplate简单实现了HA

总结

当然此例只是简单的实现了HA和负载均衡,具体的业务场景中可能需要更加强大完善和性能比较好的实现方式,目前市面上比较流行的有mycat,sharding-jdbc,atlas等一系列,各有千秋,不同公司要根据不同的业务场景进行技术选型或者自己团队实现.

PS:希望我的总结能够给你带来一些改变和提升,荣幸至极!

node.js javascript jquery jdbc

0 人点赞