任务一：分布式集群架构场景化解决方案

顺利完成第二阶段模块一的学习，正式进入第二阶段模块二。
分布式和集群
分布式和集群是不⼀样的，分布式⼀定是集群，但是集群不⼀定是分布式（因为集群就是多个实例⼀起⼯作，分布式将⼀个系统拆分之后那就是多个实例；集群并不⼀定是分布式，因为复制型的集群不是拆分⽽是复制）

1. 一致性Hash算法

Hash算法，⽐如说在安全加密领域MD5、SHA等加密算法，在数据存储和查找⽅⾯有Hash表等, 以上都应⽤到了Hash算法。
为什么需要使⽤Hash?
Hash算法较多的应⽤在数据存储和查找领域，最经典的就是Hash表，它的查询效率⾮常之⾼，其中的哈希算法如果设计的⽐较好的话，那么Hash表的数据查询时间复杂度可以接近于O(1)。

算法缺点	解决方案
浪费空间	除留余数法（开放寻址法、拉链法）
宕机（求模都需要重新计算）	一致性哈希算法

一致性哈希算法
⾸先有⼀条直线，直线开头和结尾分别定为为1和2的32次⽅减1，这相当于⼀个地址，对于这样⼀条线，弯过来构成⼀个圆环形成闭环，这样的⼀个圆环称为hash环。我们把服务器的ip或者主机名求hash值然后对应到hash环上，那么针对客户端⽤户，也根据它的ip进⾏hash求值，对应到环上某个位置，然后如何确定⼀个客户端路由到哪个服务器处理呢？按照顺时针⽅向找最近的服务器节点。

2. ⼿写实现⼀致性Hash算法

普通哈希算法

/**
* 普通Hash算法实现
*/
public class GeneralHash {
public static void main(String[] args) {
// 定义客户端IP
String[] clients = new String[]
{"10.78.12.3","113.25.63.1","126.12.3.8"};
// 定义服务器数量
int serverCount = 5;// (编号对应0，1，2)
// hash(ip)%node_counts=index
//根据index锁定应该路由到的tomcat服务器
for(String client: clients) {
int hash = Math.abs(client.hashCode());
int index = hash%serverCount;
System.out.println("客户端：" + client + " 被路由到服务器编号为："
+ index);
}
}
}

⼀致性Hash算法实现（不含虚拟节点）

import java.util.SortedMap;
import java.util.TreeMap;
public class ConsistentHashNoVirtual {
public static void main(String[] args) {
//step1 初始化：把服务器节点IP的哈希值对应到哈希环上
// 定义服务器ip
String[] tomcatServers = new String[]
{"123.111.0.0","123.101.3.1","111.20.35.2","123.98.26.3"};
SortedMap<Integer,String> hashServerMap = new TreeMap<>();
for(String tomcatServer: tomcatServers) {
// 求出每⼀个ip的hash值，对应到hash环上，存储hash值与ip的对应关系
int serverHash = Math.abs(tomcatServer.hashCode());
// 存储hash值与ip的对应关系
hashServerMap.put(serverHash,tomcatServer);
}
//step2 针对客户端IP求出hash值
// 定义客户端IP
String[] clients = new String[]
{"10.78.12.3","113.25.63.1","126.12.3.8"};
for(String client : clients) {
int clientHash = Math.abs(client.hashCode());
//step3 针对客户端,找到能够处理当前客户端请求的服务器（哈希环上顺时针最
近）
// 根据客户端ip的哈希值去找出哪⼀个服务器节点能够处理（）
SortedMap<Integer, String> integerStringSortedMap =
hashServerMap.tailMap(clientHash);
if(integerStringSortedMap.isEmpty()) {
// 取哈希环上的顺时针第⼀台服务器
Integer firstKey = hashServerMap.firstKey();
System.out.println("==========>>>>客户端：" + client + " 被
路由到服务器：" + hashServerMap.get(firstKey));
}else{
Integer firstKey = integerStringSortedMap.firstKey();
System.out.println("==========>>>>客户端：" + client + " 被
路由到服务器：" + hashServerMap.get(firstKey));
}
}
}
}

⼀致性Hash算法实现（含虚拟节点）

import java.util.SortedMap;
import java.util.TreeMap;
public class ConsistentHashWithVirtual {
public static void main(String[] args) {
//step1 初始化：把服务器节点IP的哈希值对应到哈希环上
// 定义服务器ip
String[] tomcatServers = new String[]
{"123.111.0.0","123.101.3.1","111.20.35.2","123.98.26.3"};
SortedMap<Integer,String> hashServerMap = new TreeMap<>();
// 定义针对每个真实服务器虚拟出来⼏个节点
int virtaulCount = 3;
for(String tomcatServer: tomcatServers) {
// 求出每⼀个ip的hash值，对应到hash环上，存储hash值与ip的对应关系
int serverHash = Math.abs(tomcatServer.hashCode());
// 存储hash值与ip的对应关系
hashServerMap.put(serverHash,tomcatServer);
// 处理虚拟节点
for(int i = 0; i < virtaulCount; i++) {
int virtualHash = Math.abs((tomcatServer + "#" +
i).hashCode());
hashServerMap.put(virtualHash,"----由虚拟节点"+ i + "映射过
来的请求："+ tomcatServer);
}
}
//step2 针对客户端IP求出hash值
// 定义客户端IP
String[] clients = new String[]
{"10.78.12.3","113.25.63.1","126.12.3.8"};
for(String client : clients) {
int clientHash = Math.abs(client.hashCode());
//step3 针对客户端,找到能够处理当前客户端请求的服务器（哈希环上顺时针最
近）
// 根据客户端ip的哈希值去找出哪⼀个服务器节点能够处理（）
SortedMap<Integer, String> integerStringSortedMap =
hashServerMap.tailMap(clientHash);
if(integerStringSortedMap.isEmpty()) {
// 取哈希环上的顺时针第⼀台服务器
Integer firstKey = hashServerMap.firstKey();
System.out.println("==========>>>>客户端：" + client + " 被
路由到服务器：" + hashServerMap.get(firstKey));
}else{
Integer firstKey = integerStringSortedMap.firstKey();
System.out.println("==========>>>>客户端：" + client + " 被
路由到服务器：" + hashServerMap.get(firstKey));
}
}
}
}

任务二：分布式ID解决方案

1. UUID(可以⽤)
   UUID 是指Universally Unique Identifier，翻译为中⽂是通⽤唯⼀识别码，产⽣重复 UUID 并造成错误的情况⾮常低，是故⼤可不必考虑此问题。，Java中得到⼀个UUID，可以使⽤java.util包提供的⽅法。

public class MyTest {
public static void main(String[] args) {
System.out.println(java.util.UUID.randomUUID().toString());
}
}

2. SnowFlake 雪花算法（可以⽤，推荐）
   雪花算法是Twitter推出的⼀个⽤于⽣成分布式ID的策略。
   雪花算法是⼀个算法，基于这个算法可以⽣成ID，⽣成的ID是⼀个long型，那么在Java中⼀个long型是8个字节，算下来是64bit，如下是使⽤雪花算法⽣成的⼀个ID的⼆进制形式示意：
   
   注意：图中序列号为12
3. 借助Redis的Incr命令获取全局唯⼀ID（推荐）
   Redis Incr 命令将 key 中储存的数字值增⼀。如果 key 不存在，那么 key 的值会先被初始化为 0，然后再执⾏ INCR 操作。

任务三：分布式调度问题

调度—>定时任务，分布式调度—>在分布式集群环境下定时任务这件事
什么是分布式任务调度？有两层含义
1）运⾏在分布式集群环境下的调度任务（同⼀个定时任务程序部署多份，只应该有⼀个定时任务在执行）
2）分布式调度—>定时任务的分布式—>定时任务的拆分（即为把⼀个⼤的作业任务拆分为多个小的作业任务，同时执行）
定时任务与消息队列的区别：
定时任务作业是时间驱动，⽽MQ是事件驱动； 时间驱动是不可代替的，⽐如⾦融系统每⽇的利息结算，不是说利息来⼀条（利息到来事件）就算⼀下，⽽往往是通过定时任务批量计算；所以，定时任务作业更倾向于批处理，MQ倾向于逐条处理；

任务四：分布式调度框架Elastic-Job

Elastic-Job是当当⽹开源的⼀个分布式调度解决⽅案，基于Quartz⼆次开发的，由两个相互独⽴的⼦项⽬Elastic-Job-Lite和Elastic-Job-Cloud组成。我们要学习的是 Elastic-Job-Lite，它定位为轻量级⽆中⼼化解决⽅案，使⽤Jar包的形式提供分布式任务的协调服务，⽽Elastic-Job-Cloud⼦项⽬需要结合Mesos以及Docker在云环境下使⽤。
引入jar包

<!-- https://mvnrepository.com/artifact/com.dangdang/elastic-job-lite-core
-->
<dependency>
<groupId>com.dangdang</groupId>
<artifactId>elastic-job-lite-core</artifactId>
<version>2.1.5</version>
</dependency>

定时任务类

package elasticjob;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import util.JdbcUtil;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.List;
import java.util.Map;
public class BackupJob implements SimpleJob {
// 定时任务每执⾏⼀次都会执⾏如下的逻辑
@Override
public void execute(ShardingContext shardingContext) {
/*
从resume数据表查找1条未归档的数据，将其归档到resume_bak
表，并更新状态为已归档（不删除原数据）
*/
// 查询出⼀条数据
String selectSql = "select * from resume where
state='未归档' limit 1";
List<Map<String, Object>> list =
JdbcUtil.executeQuery(selectSql);
if(list == null || list.size() == 0) {
return;
}
Map<String, Object> stringObjectMap = list.get(0);
long id = (long) stringObjectMap.get("id");
String name = (String) stringObjectMap.get("name");
String education = (String)
stringObjectMap.get("education");
// 打印出这条记录
System.out.println("======>>>id：" + id + " name：" +
name + " education：" + education);
// 更改状态
String updateSql = "update resume set state='已归档'
where id=?";
JdbcUtil.executeUpdate(updateSql,id);
// 归档这条记录
String insertSql = "insert into resume_bak select *
from resume where id=?";
JdbcUtil.executeUpdate(insertSql,id);
}
}

主类

package elasticjob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import
com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.lite.api.JobScheduler;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import
com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter;
import
com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
import
com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
public class ElasticJobMain {
public static void main(String[] args) {
// 配置注册中⼼zookeeper，zookeeper协调调度，不能让任务重复执⾏，
通过命名空间分类管理任务，对应到zookeeper的⽬录
ZookeeperConfiguration zookeeperConfiguration = new
ZookeeperConfiguration("localhost:2181","data-archive-job");
CoordinatorRegistryCenter coordinatorRegistryCenter = new
ZookeeperRegistryCenter(zookeeperConfiguration);
coordinatorRegistryCenter.init();
// 配置任务
JobCoreConfiguration jobCoreConfiguration =
JobCoreConfiguration.newBuilder("archive-job","*/2 * * * *
?",1).build();
SimpleJobConfiguration simpleJobConfiguration = new
SimpleJobConfiguration(jobCoreConfiguration,BackupJob.class.getNam
e());
// 启动任务
new JobScheduler(coordinatorRegistryCenter,
LiteJobConfiguration.newBuilder(simpleJobConfiguration).build()).i
nit();
}
}

JdbcUtil⼯具类

package util;
import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class JdbcUtil {
//url
private static String url = "jdbc:mysql://localhost:3306/job?
characterEncoding=utf8&useSSL=false";
//user
private static String user = "root";
//password
private static String password = "123456";
//驱动程序类
private static String driver = "com.mysql.jdbc.Driver";
static {
try {
Class.forName(driver);
} catch (ClassNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
public static Connection getConnection() {
try {
return DriverManager.getConnection(url, user,
password);} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return null;
}
public static void close(ResultSet rs, PreparedStatement ps,
Connection con) {
if (rs != null) {
try {
rs.close();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
if (ps != null) {
try {
ps.close();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
if (con != null) {
try {
con.close();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
}
}
}
public static void executeUpdate(String sql,Object...obj) {
Connection con = getConnection();
PreparedStatement ps = null;
try {
ps = con.prepareStatement(sql);
for (int i = 0; i < obj.length; i++) {
ps.setObject(i + 1, obj[i]);
}
ps.executeUpdate();} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
close(null, ps, con);
}
}
public static List<Map<String,Object>> executeQuery(String
sql, Object...obj) {
Connection con = getConnection();
ResultSet rs = null;
PreparedStatement ps = null;
try {
ps = con.prepareStatement(sql);
for (int i = 0; i < obj.length; i++) {
ps.setObject(i + 1, obj[i]);
}
rs = ps.executeQuery();
List<Map<String, Object>> list = new ArrayList<>();
int count = rs.getMetaData().getColumnCount();
while (rs.next()) {
Map<String, Object> map = new HashMap<String,
Object>();
for (int i = 0; i < count; i++) {
Object ob = rs.getObject(i + 1);
String key = rs.getMetaData().getColumnName(i
+ 1);
map.put(key, ob);
}
list.add(map);
}
return list;
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
close(rs, ps, con);
}
return null;
}
}

任务五：Session共享问题

Session共享及Session保持或者叫做Session⼀致性

出现这个问题的原因，从根本上来说是因为Http协议是⽆状态的协议。客户端和服务端在某次会话中产⽣的数据不会被保留下来，所以第⼆次请求服务端⽆法认识到你曾经来过， Http为什么要设计为⽆状态协议？早期都是静态⻚⾯⽆所谓有⽆状态，后来有动态的内容更丰富，就需要有状态，出现了两种⽤于保持Http状态的技术，那就是Cookie和Session。
解决方案

1. Nginx的 IP_Hash 策略（可以使⽤）
   同⼀个客户端IP的请求都会被路由到同⼀个⽬标服务器，也叫做会话粘滞
   优点：
   配置简单，不⼊侵应⽤，不需要额外修改代码
   缺点：
   服务器重启Session丢失、存在单点负载⾼的⻛险、单点故障问题
2. Session共享，Session集中存储（推荐）
   Session的本质就是缓存，把Session数据交给专业的缓存中间件，⽐如Redis。
   优点:
   能适应各种负载均衡策略、服务器重启或者宕机不会造成Session丢失、扩展能⼒强
   适合⼤集群数量使⽤
   缺点：
   对应⽤有⼊侵，引⼊了和Redis的交互代码。
   Spring Session使得基于Redis的Session共享应⽤起来⾮常之简单。

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.session</groupId>
<artifactId>spring-session-data-redis</artifactId>
</dependency>

配置redis

spring.redis.database=0
spring.redis.host=127.0.0.1
spring.redis.port=6379

添加注解