Semaphore 和操作系统的信号量很类似，但是这里它也可以作为锁使用。即可以作为锁与协调线程作用
1 | public class SemaphoreDemo { |
打印结果
2等待
1等待
3等待
2进入
2释放
1进入
1释放
3进入
3释放
可以看到首先3个线程同时到来临界区边界，但是同时只有一个线程能进入临界区。
1 | public class SemaohoreDemo2 { |
打印结果
pool-1-thread-3进入临界区
pool-1-thread-2进入临界区
pool-1-thread-1进入临界区
pool-1-thread-3释放信号
pool-1-thread-1释放信号
pool-1-thread-2释放信号
pool-1-thread-4进入临界区
pool-1-thread-5进入临界区
pool-1-thread-6进入临界区
pool-1-thread-4释放信号
pool-1-thread-5释放信号
pool-1-thread-6释放信号
从结果可以看到有3个线程同时进入了临界区。
LockSupport 相比较suspend会造成永久被挂起，它是安全的，即使unpark()方法再在前面执行也无所谓，因为它的底层使用的信号量，为每一个线程准备了许可，如果许可可用，则park惠消费它，如果不可以不可用则阻塞等待，而unpark是将一个许可变成可用，但是和信号量有些许区别，它是不可以累加的，每次最多有一个许可。
1 | public class LockSupportDemo { |
事务在平常开发会经常遇到，介于有些同学对事务的理解还不是很透彻，在此稍作总结下
1.什么是事务
把一堆事情绑在一起做，都成功了才算完成，否则就恢复之前的样子。
2.事务的特性ACID（核心为C，即一致性，其他三个均是为一致性服务的）
原子性：事务是数据库的逻辑工作单位，事务中包含的各操作要么都做，要么都不做 。
一致性：事务执行的结果必须是使数据库从一个一致性状态变到另一个一致性状态。比如，当数据库只包含成功事务提交的结果时，就说数据库处于一致性状态。如果数据库系统在运行中发生故障，有些事务尚未完成就被迫中断，这些未完成事务对数据库所做的修改有一部分已写入物理数 据库，这时数据库就处于一种不正确的状态，或者说是不一致的状态。
隔离性：一个事务的执行不能其它事务干扰。即一个事务内部的操作及使用的数据对其它并发事务是隔离的，并发执行的各个事务之间不能互相干扰。
持续性：指一个事务一旦提交，它对数据库中的数据的改变就应该是永久性的。接下来的其它操作或故障不应该对其执行结果有任何影响。
3.spring中事务
Spring事务管理高层抽象主要包括3个接口
PlatformTransactionManager：事务管理器—主要用于平台相关事务的管理
TransactionDefinition： 事务定义信息(隔离、传播、超时、只读)—通过配置如何进行事务管理
TransactionStatus：事务具体运行状态—事务管理过程中，每个时间点事务的状态信息
3.1Spring为不同的持久化框架提供了不同PlatformTransactionManager接口实现类
事务 说明
org.springframework.jdbc.datasource.DataSourceTransactionManager 使用Spring JDBC或iBatis 进行持久化数据时使用（咱项目是ORM是myBatis，所以用这）
org.springframework.orm.hibernate3.HibernateTransactionManager 使用Hibernate3.0版本进行持久化数据时使用
org.springframework.orm.jpa.JpaTransactionManager 使用JPA进行持久化时使用
org.springframework.jdo.JdoTransactionManager 当持久化机制是Jdo时使用
org.springframework.transaction.jta.JtaTransactionManager 使用一个JTA实现来管理事务，在一个事务跨越多个资源时必须使用
DataSourceTransactionManager针对JdbcTemplate、MyBatis 事务控制 ，使用Connection（连接）进行事务控制 ：
开启事务 connection.setAutoCommit(false);
提交事务 connection.commit();
回滚事务 connection.rollback();
HibernateTransactionManager针对Hibernate框架进行事务管理， 使用Session的Transaction相关操作进行事务控制 ：
开启事务 session.beginTransaction();
提交事务 session.getTransaction().commit();
回滚事务 session.getTransaction().rollback();
TransactionDefinition事务定义信息，该接口主要提供的方法：
lgetIsolationLevel：获取隔离级别
lgetPropagationBehavior：获取传播行为
lgetTimeout：获取超时时间（事务的有效期）
lisReadOnly 是否只读(保存、更新、删除—对数据进行操作-变成可读写的，查询-设置这个属性为true，只能读不能写)，事务管理器能够根据这个返回值进行优化。
这些事务的定义信息，都可以在配置文件中配置和定制。
4.事务的隔离级别
隔离级别 说明
DEFAULT 使用后端数据库默认的隔离级别(spring中的默认选择项，mysql为可重复度)
READ_UNCOMMITED 允许你读取还未提交的改变了的数据。可能导致脏、幻、不可重复读
READ_COMMITTED 允许在并发事务已经提交后读取。可防止脏读，但幻读和 不可重复读仍可发生
REPEATABLE_READ 对相同字段的多次读取是一致的，除非数据被事务本身改变。可防止脏、不可重复读，但幻读仍可能发生。
SERIALIZABLE 完全服从ACID的隔离级别，确保不发生脏、幻、不可重复读。这在所有的隔离级别中是最慢的，它是典型的通过完全锁定在事务中涉及的数据表来完成的。
5.事务不回滚原因
默认spring事务只在发生未被捕获的 runtimeexcetpion时才回滚；
spring aop 异常捕获原理：被拦截的方法需显式抛出异常，并不能经任何处理，这样aop代理才能捕获到方法的异常，才能进行回滚，默认情况下aop只捕获runtimeexception的异常，但可以通过配置来捕获特定的异常并回滚 ！
解决方案
如果是service层处理事务，那么service中的方法中不做异常捕获，或者在catch语句中最后增加throw new RuntimeException()语句，以便让aop捕获异常再去回滚，并且在service上层（webservice客户端，view层action）要继续捕获这个异常并处理
在service层方法的catch语句中增加：TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();语句，手动回滚，这样上层就无需去处理异常
前言:目前大部分操作系统都是以线程为CPU调度和分派基本单位。多线程在日常程序中运用的还是比较多的，潜在的我们web容器帮我们在http层面可以同时处理多个请求，这些可能多我们是无感的。平常的开发中，我们需要结合业务场景来合理运用多线程，例如大文件的IO操作，大量消息发送，都可以运用多线程来处理，充分利用多核CPU性能。
继承Thread类创建线程
实现Runnable接口创建线程
使用Callable和Future创建线程
使用线程池
a.继承Thread类，重写run()
public class Test extends Thread {
@Override
public void run() {
System.out.println(“hello,word!”);
}
public static void main(String[] args) {
Test test = new Test();
test.start();
}
}
b.实现Runnable接口创建线程
public class Test implements Runnable {
@Override
public void run() {
System.out.println(“hello,word!”);
}
public static void main(String[] args) {
Thread thread = new Thread(new Test());
thread.start();
}
}
//Java 8 lambda写法
public class Test {
public static void main(String[] args) {
Thread thread = new Thread(
() -> System.out.println(“hello,word!”));
thread.start();
}
}
c.使用Callable和Future创建线程
public class Test implements Callable
@Override
public String call() throws Exception {
return “hello,word!”;
}
public static void main(String[] args) {
Callable<String> test = new Test();
try {
FutureTask<String> futureTask = new FutureTask<>(test);
Thread thread = new Thread(futureTask);
thread.run();
System.out.println(futureTask.get());
} catch (Exception e) {
e.printStackTrace();
}
}
}
//java8 lambda匿名类写法
public class Test {
public static void main(String[] args) {
FutureTask
Thread thread = new Thread(futureTask);
thread.start();
try {
System.out.println(futureTask.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
d.使用线程池
public class Test {
public static void main(String[] args) {
FutureTask
ExecutorService service=Executors.newFixedThreadPool(2);
service.submit(futureTask);
try {
System.out.println( futureTask.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
## 二、Callable
Callable和Runnbale一样代表着任务，区别在于Callable有返回值并且可以抛出异常！Callable是一种可以返回结果的任务，这是它与Runnable的区别，但是通过适配器模式可以使Runnable与Callable类似。Future代表了一个异步的计算，可以从中得到计算结果、查看计算状态，其实现FutureTask可以被提交给Executor执行，多个线程可以从中得到计算结果。Callable和Future是配合使用的，当从Future中get结果时，如果结果还没被计算出来，那么线程将会被挂起，FutureTak内部使用一个单链表维持等待的线程；当计算结果出来后，将会对等待线程解除挂起，等待线程就都可以得到计算结果了。
# 三、Future接口、FutureTask类
Future是一个接口，代表了一个异步计算的结果。接口中的方法用来检查计算是否完成、等待完成和得到计算的结果。当计算完成后，只能通过get()方法得到结果，get方法会阻塞直到结果准备好了。如果想取消，那么调用cancel()方法。其他方法用于确定任务是正常完成还是取消了。一旦计算完成了，那么这个计算就不能被取消。
FutureTask类实现了RunnableFuture接口，而RunnnableFuture接口继承了Runnable和Future接口，所以说FutureTask是一个提供异步计算的结果的任务。 FutureTask可以用来包装Callable或者Runnbale对象。因为FutureTask实现了Runnable接口，所以FutureTask也可以被提交给Executor。
//FutureTask 2构造方法
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
# 四、线程池工厂
>Executors面对开发人员的创建线程池的工厂类，它和线程执行器ExecutorService关系很紧密。
可以看到，提供了很多静态的方法，我们主要关注以下方法：
* newFixedThreadPool(int nThreads) 固定线程数量的线程池
* newScheduledThreadPool(int corePoolSize) 拥有至少corePoolSize数量的线程池，可以延迟或者固定频率执行
* newCachedThreadPool() 线程不固定线程池，常用于多且小的异步任务使用
* newSingleThreadExecutor() 串行按照提交顺序执行任务，的单线程化线程池
* newWorkStealingPool() 任务窃取线程池，JDK1.8以后加的，充分利用现有的cpu资源创建线程池,一个线程维护一个任务队列，常用在任务执行时间差别较大的业务
```java
public class ExecutorsDemo {
private static ExecutorService service = Executors.newCachedThreadPool();
private static ExecutorService service2 = Executors.newFixedThreadPool(3);
private static ExecutorService service3 = Executors.newScheduledThreadPool(3);
private static ExecutorService service4 = Executors.newSingleThreadExecutor();
private static ExecutorService service5 = Executors.newWorkStealingPool();
public static void main(String[] args) throws Exception {
addTask("newCachedThreadPool", service);
addTask("newFixedThreadPool", service2);
addTask("newScheduledThreadPool", service3);
addTask("newSingleThreadExecutor", service4);
addTask("newWorkStealingPool", service5);
}
private static void addTask(String type, ExecutorService service) throws Exception {
for (int i = 0; i < 10; i++) {
service.execute(() -> System.out.println(type + "," + Thread.currentThread().getName()));
}
service.shutdown();
service.awaitTermination(5, TimeUnit.MILLISECONDS);//等待任务完成再关闭主main线程
}
}
注意：虽然service.shutdown()是关闭线程，此时已经提交的线程会被执行完毕，但是打印需要时间，所以再最后加了一句 service.awaitTermination(5, TimeUnit.MILLISECONDS)，目的是为了看到打印消息。
ScheduledExecutorService接口里面的4个方法
1 | public interface ScheduledExecutorService extends ExecutorService { |
打印结果
00000
00000
i:2071039910,j:1508352969
i:-1148719051,j:-743674023
00000
00000
线程中断来替代stop
上面代码读取的不一致，核心原因是一以为线程突然停止，那么有没有一种优雅的停止呢，JDK为我们提供了更优雅的方式，中断。我们将上述代码稍做改变，数据的一致性立马可以得到保证。
1 | public class ThreadTest { |
结果
00000
00000
···
注意点：若是被设置了中断的线程使用了sleep方法，则运行时可能会抛出中断异常且中断标志会被清空，这里贴出异常，代码就不写出来了。
1 | java.lang.InterruptedException: sleep interrupted |
wait和notify时Object里面定义的方法。
1 | public class ThreadTest2 { |
打印结果
wait方法被执行了
notify方法被执行了
notify休眠1秒后
wait方法释放后被执行了,等待了notify方法释放锁1秒
suspend 和resume (Deprecated）
suspend线程会被挂起，它是不释放监视器（锁）的，必须等到resume 才会释放锁。这2个方法也被废弃了，因为若resume在suspend前执行，会导致线程永远别挂起，自己和其他线程由于获取不到监视器根本没办法正常工作，下面演示个被一直被挂起例子
1 | public class ThreadTest3 { |
1 | "Thread-1" #11 prio=5 os_prio=31 tid=0x00007fdbd703f800 nid=0x5903 runnable [0x000070000aa79000] |
if (millis == 0) {
while (isAlive()) {
wait(0);
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
* yield（谦让） 个人觉得这个方法使用需要勇气,有点鸡肋，因为它执行了，表明是让出当前CPU时间片，让和它同等优先级的线程优先执行，但是它持有的锁又不释放，共有资源还会再接下来时间竞争，且时间还不固定。
```java
public class ThreadTest4 {
static int j = 0;
public static void main(String[] args) throws Exception {
Thread thread = new Thread(() -> {
for (int i = 0; i < 10; i++) {
j = i;
}
});
thread.start();
thread.join();//main的输出，等待线程执行完再执行,不然输出可能是0或者很小的值
System.out.println(j);
}
}
每个eureka客户端都需要添加@EnableDiscoverClient注解，作用是开启一个DiscoveryClient客户端，具体实现是在eureka相关配置类加载的时候调用了DiscoveryClient构造方法来初始化的
1
2
3
4
5
6
7
8
9
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Import({EnableDiscoveryClientImportSelector.class})
public @interface EnableDiscoveryClient {
boolean autoRegister() default true;
}
看看DiscoveryClient它的注解
构造函数1
2
3
4
5
6
7 @Inject
DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
Provider<BackupRegistry> backupRegistryProvider) {
...
initScheduledTasks();//初始化定时任务（注册、续约、取消续约、拉取服务）
...
}
1 | private void initScheduledTasks() { |
1 | if (clientConfig.shouldRegisterWithEureka()) { |
我们来看看InstanceInfoReplicator 这个类的start和run方法
1 | public void start(int initialDelayMs) { |
DiscoveryClient里面的register注册方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14boolean register() throws Throwable {
logger.info(PREFIX + appPathIdentifier + ": registering service...");
EurekaHttpResponse<Void> httpResponse;
try {
httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
} catch (Exception e) {
logger.warn("{} - registration failed {}", PREFIX + appPathIdentifier, e.getMessage(), e);
throw e;
}
if (logger.isInfoEnabled()) {
logger.info("{} - registration status: {}", PREFIX + appPathIdentifier, httpResponse.getStatusCode());
}
return httpResponse.getStatusCode() == 204;
}
实际上它是调用了 RetryableEurekaHttpClient 的execute 方法1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47@Override
protected <R> EurekaHttpResponse<R> execute(RequestExecutor<R> requestExecutor) {
List<EurekaEndpoint> candidateHosts = null;
int endpointIdx = 0;
// 循环次数，默认3次，成功就reutrn
for (int retry = 0; retry < numberOfRetries; retry++) {
EurekaHttpClient currentHttpClient = delegate.get();
EurekaEndpoint currentEndpoint = null;
if (currentHttpClient == null) {
if (candidateHosts == null) {
// 获取注册中心的节点（全部和无效取交集）
candidateHosts = getHostCandidates();
if (candidateHosts.isEmpty()) {
throw new TransportException("There is no known eureka server; cluster server list is empty");
}
}
if (endpointIdx >= candidateHosts.size()) {
throw new TransportException("Cannot execute request on any known server");
}
// 只获取了一个节点
currentEndpoint = candidateHosts.get(endpointIdx++);
currentHttpClient = clientFactory.newClient(currentEndpoint);
}
try {
// 发送请求注册
EurekaHttpResponse<R> response = requestExecutor.execute(currentHttpClient);
if (serverStatusEvaluator.accept(response.getStatusCode(), requestExecutor.getRequestType())) {
delegate.set(currentHttpClient);
if (retry > 0) {
logger.info("Request execution succeeded on retry #{}", retry);
}
// 直接reutrn 了
return response;
}
logger.warn("Request execution failure with status code {}; retrying on another server if available", response.getStatusCode());
} catch (Exception e) {
logger.warn("Request execution failed with message: {}", e.getMessage()); // just log message as the underlying client should log the stacktrace
}
// Connection error or 5xx from the server that must be retried on another server
delegate.compareAndSet(currentHttpClient, null);
if (currentEndpoint != null) {
// 无效的节点
quarantineSet.add(currentEndpoint);
}
}
throw new TransportException("Retry limit reached; giving up on completing the request");
}
代码太多就不继续跟进了，实质是向注册中心的一个节点发送POST方法向注册中心注册，那么注册中心是怎么处理这个请求的呢，后面有分析。
同理我们看看DiscoveryClient里面的renew注册方法，不再具体分析
1 | boolean renew() { |
这个比较简单，也不再赘述，按照注册思路可以很容易看明白1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24if (clientConfig.shouldFetchRegistry()) {
// registry cache refresh timer
int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
scheduler.schedule(
new TimedSupervisorTask(
"cacheRefresh",
scheduler,
cacheRefreshExecutor,
registryFetchIntervalSeconds,
TimeUnit.SECONDS,
expBackOffBound,
// 这个线程为实现拉取服务主要工作
new CacheRefreshThread()
),
registryFetchIntervalSeconds, TimeUnit.SECONDS);
}
// 内部类，一个线程
class CacheRefreshThread implements Runnable {
public void run() {
// 真正实现功能方法
refreshRegistry();
}
}
1 | @Override |
父类的regiser方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
try {
read.lock();
//首先根据appName获取一些列的服务实例对象，如果为Null,则新创建一个map并把当前的注册应用程序信息添加到此Map当中
Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
REGISTER.increment(isReplication);
if (gMap == null) {
final ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String, Lease<InstanceInfo>>();
gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);
if (gMap == null) {
gMap = gNewMap;
}
}
... 注册过程
} finally {
read.unlock();
}
}
同步到其他节点
1 | private void replicateToPeers(Action action, String appName, String id, |
其他源码不讲了，大致差不多的思路。
1 | @SpringBootApplication |
1 | public void registerFeignClients(AnnotationMetadata metadata, BeanDefinitionRegistry registry) { |
工厂Bean我们都知道通过getObject方法获取的是具体的Bean实例，不是工厂本身
1 | @Override |
再一步步跟进去会发现是ReflectiveFeign 类的newInstance方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26 @Override
public <T> T newInstance(Target<T> target) {
Map<String, MethodHandler> nameToHandler = targetToHandlersByName.apply(target);
Map<Method, MethodHandler> methodToHandler = new LinkedHashMap<Method, MethodHandler>();
List<DefaultMethodHandler> defaultMethodHandlers = new LinkedList<DefaultMethodHandler>();
for (Method method : target.type().getMethods()) {
if (method.getDeclaringClass() == Object.class) {
continue;
} else if(Util.isDefault(method)) {
DefaultMethodHandler handler = new DefaultMethodHandler(method);
defaultMethodHandlers.add(handler);
methodToHandler.put(method, handler);
} else {
methodToHandler.put(method, nameToHandler.get(Feign.configKey(target.type(), method)));
}
}
// jdk动态代理
InvocationHandler handler = factory.create(target, methodToHandler);
T proxy = (T) Proxy.newProxyInstance(target.type().getClassLoader(), new Class<?>[]{target.type()}, handler);
// 注入handler，拦截器
for(DefaultMethodHandler defaultMethodHandler : defaultMethodHandlers) {
defaultMethodHandler.bindTo(proxy);
}
return proxy;
}
SynchronousMethodHandler的invoke方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 @Override
public Object invoke(Object[] argv) throws Throwable {
// 请求模板
RequestTemplate template = buildTemplateFromArgs.create(argv);
// 重试配置类，默认5次
Retryer retryer = this.retryer.clone();
while (true) {
try {
// 发送请求
return executeAndDecode(template);
} catch (RetryableException e) {
retryer.continueOrPropagate(e);
if (logLevel != Logger.Level.NONE) {
logger.logRetry(metadata.configKey(), logLevel);
}
continue;
}
}
}
1 | Object executeAndDecode(RequestTemplate template) throws Throwable { |
整个负载还是比较复杂的，继续跟踪下去太多，大家可以自行看源码。
不知道具体业务，只能做个demo。
vim 对于程序员的重要性，不言语。一个合格的程序员熟练使用命令编辑器，是必要条件。
1 | while 判断条件： |
和Java 差不多，就是每个判断表达式后面需要带上冒号，另外结尾不需要封号。
1 | if condition_1: |
1 | for <variable> in <sequence>: |
1 | sites = ["Baidu", "Google","Runoob","Taobao"] |
1 | while True: |
1 | import sys # 引入 sys 模块 |
1 | def 函数名（参数列表）: |
1 | if True: |
spring-context
构建与core和bean基础之上，主要是提供了供程序员使用的ApplicationContext。
spring-expression
强大的表达式语言，供Spring对于各种文件中变量的访问和操控。
spring-test 模块支持带有JUnit或TestNG的Spring组件的 unit testing 和 integration testing 。它提供了一致的 ApplicationContext 的Spring ApplicationContext 和 caching 这些上下文。它还提供 mock objects ，您可以使用它来单独测试代码。
映射器可以理解为我们平常说的mapper接口，一般映射器可以由接口+xml/注解方式组成。生命周期：一个会话内。
1 |
|
1 | (value="select *id,username,age from user where id=#{id}") |
1 | // MapperProxy工厂 |
四大金刚中的核心，使用数据库的PreparedStatement执行数据库操作。
1 | public interface StatementHandler { |