| |

事故描述

事故现象是部分服务http请求无响应。事故从发生到恢复，接近3个小时，事故过程中重启应用服务，只能坚持几分钟到十几分钟，在真正发现问题前通过不断重启服务实例来支撑，庆幸的是核心服务没有出现无响应的事故。

最终分析为AMQ出现故障，现象是MQ客户端sendMessage后等待响应，但一直在等待，AMQ监控端口ok，控制台也可以打开，由于紧急没有具体分析，直接重启AMQ服务，切换master，通过验证服务全部恢复。

这次故障大部分服务都使用了AMQ，但除了一个核心服务没受到明显影响外，其他使用AMQ的服务都不同程度的收到了影响，服务不可用。


事后通过分析这个核心服务正式使用了本地文件队列避免了事故放大，逃过一劫，当时如果这个核心服务也受到影响就可想而知了，事故间期正直商户业务高峰期，客户估计要炸了，公司也会受到很大的损失。

此次事故比较严重，就是因为使用了本地文件队列有效隔离故障，使得影响面不大。假设（当然不希望发生了）核心业务没有使用本地文件队列来隔离故障，整个下单、收银服务将不可用，商户无法营业，损失应该在数量级。

此次事故也证明了我当时的这个架构思路的正确性，主要体现在隔离和降级。

说说这个核心服务使用“本地文件队列”

我开发出来这个组件在这个团队使用一直很稳定效果也很好。

实际上这个团队使用“本地文件队列”的姿势并不是我期望的，本身使用方法并没有明显不妥，只是会延迟消息的消费，但这中方法可以很好的且有效的隔离故障。

就是在发送AMQ消息的方法上添加了@AsyncExecutable，所以在入AMQ前先入队“本地文件队列”，然后“本地文件队列”消费者再把消息生产到AMQ。

这里正是利用率“本地文件队列”的优点，比较可靠，只依赖于本地文件系统，不会有网络故障的特性，近似不会被阻塞。

在事故分析中，实际上AMQ对该核心服务也受到影响，但由于采用了“本地文件队列”作为一级队列，有效的隔离了对AMQ的网络依赖，所以没有放大事故。事故中“本地文件队列”中消息被积累，没有被消费，重启服务后才被消费，原因后面再分析。

再说说此次事故中服务不可用的原因

  1. AMQ出现故障，现象是MQ客户端sendMessage后等待响应，但一直在等待。
  2. AMQ Client消息的生产和Tomcat共用worker线程
  3. AMQ Client消息的生产没有超时机制
  4. AMQ Client消息的生产采用同步发送，异步发送有一些问题场景不太合适。

所以基于以上信息，AMQ消息的生产阻塞了Tomcat worker线程，最终导致worker线程被耗光而服务不可用。

在事故中通过线程堆栈信息和Tomcat线程使用数统计也确定了线程很快被耗光而请求被阻塞。

另一个服务中采用了异步线程池来生产AMQ消息，但拒绝策略采用了CallerRunsPolicy, 也是线程池线程很快被耗光而再耗光Tomcat worker线程，最终导致服务不可用。

事故产生的原因就是代码中没有有效做网路调用的隔离和降级。

上面提到的核心服务也受到影响，参考《本地文件队列-异步隔离》，也是因为“本地文件队列”拒绝策略采用了CallerRunsPolicy，最终导致线程池线程很快被耗光，而使用“本地文件队列”消费调度主线程，消费调度主线程被阻塞而无法消费“本地文件队列”消息并生产到AMQ。但这里和其他服务不同的是，主业务和AMQ的消息生产是隔离的，主业务生产消息到“本地文件队列”就返回，并不直接依赖AMQ。

本文中提到了使用本地文件队列的隔离姿势。

后面再介绍使用本地文件队列的降级姿势。

| |

常见的异步方式：

创建异步线程

每个新创建一个线程来执行异步任务，任务结束线程也终止。
线程的创建成本比较大，不建议使用。

使用Queue，producer/consumer方式

在内部创建一个Queue，worker线程直接将异步处理的任务放入queue，一个或多个异步线程从queue中消费并执行任务。

线程池

用线程池来替换每次创建线程，减少线程创建的成本，线程被复用，一次创建多处使用。

和使用Queue类似，也是通过BlockingQueue实现,但策略上更复杂，向线程池提交Callable&Runnable任务，由线程池调度执行。

参考：java.util.concurrent.ThreadPoolExecutor#execute

spring @Async注解

通过注解来来简化了异步编程，只需要在需要异步的方法上使用@Async注解即可。
其本质也是在线程池功能上扩展的，将异步执行方法封装为一个Callable，然后提交给线程池。

org.springframework.aop.interceptor.AsyncExecutionInterceptor:

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
public Object invoke(final MethodInvocation invocation) throws Throwable {
		Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
		Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);
		final Method userDeclaredMethod = BridgeMethodResolver.findBridgedMethod(specificMethod);
		AsyncTaskExecutor executor = determineAsyncExecutor(userDeclaredMethod);
		if (executor == null) {
			throw new IllegalStateException(
					"No executor specified and no default executor set on AsyncExecutionInterceptor either");
		}
		Callable<Object> task = new Callable<Object>() {
			@Override
			public Object call() throws Exception {
				try {
					Object result = invocation.proceed();
					if (result instanceof Future) {
						return ((Future<?>) result).get();
					}
				}
				catch (ExecutionException ex) {
					handleError(ex.getCause(), userDeclaredMethod, invocation.getArguments());
				}
				catch (Throwable ex) {
					handleError(ex, userDeclaredMethod, invocation.getArguments());
				}
				return null;
			}
		};
		return doSubmit(task, executor, invocation.getMethod().getReturnType());
	}

详细参考：http://docs.spring.io/spring/docs/current/spring-framework-reference/htmlsingle/#scheduling-annotation-support-async

背景和场景

产生的背景

在项目中使用了@Async来执行异步任务，但在线上运行时出现了一次OOM的故障。通过分析发现是，线程池队列设置的比较大，当时的JVM内存给的也比较少（2048M），异步任务方法参数中传了大量的数据，任务执行被后端数据库阻塞（后端数据库变慢），最后导致缓存了大量的数据被放到线程池队列。其实JVM内存配置合适，线程池队列数合适，并配置合适的RejectedExecutionHandler策略。

产生这个组件，1) 旨在替换内存队列的异步方式 2) 用来方便扩展集成分布式MQ

异步隔离

除了上面背景和场景，开发这个组件的另一个初衷就是有效异步隔离和作为一个降级备份方案。
也是主要实现了文件队列方式的一个原因。

当我们使用分布式MQ时，难免分布式MQ宕机或者其他网络等原因导致不能生产消息，或者阻塞影响到本身的业务，出现这种情况时可以降级到本地文件队列。

本地文件队列的优点是速度快，只要文件系统不出问题可以认为不会被阻塞。缺点是本地文件队列生产的消息必须自己来消费，出现故障时消息消费会延迟，文件系统的损坏也会导致消息丢失。主要看使用的姿势，更看重哪一方面了。

基本架构设计思路

采用producer/consumer生产消费设计模式。

参考了@Async思路，定义一个注解@AsyncExecutable, 使用Spring拦截器拦截注解了@AsyncExecutable的方法，可以使用AOP或者BeanPostProcessor来应用拦截器。

producer

拦截器拦截到@AsyncExecutable方法后，将该方法所有的参数和方法信息作为Message，并序列化Message，序列化采用Kryo或者Json，将序列化后的信息放入队列。

1
2
3
4
5
6
7
8
9
10
class Message {
     String beanName;
     String klassName;
     String methodName;
     Class<?>[] argTypes;
     Object[] args;
     boolean hasTransactional = true;
     
}

consumer

有1个调度主线程和worker线程组成，主线程负责从队列中拉取消息，并分发到worker线程，worker线程采用线程池，使用了spring提供的TaskExecutor。

worker线程反序列化消息为Message对象，并根据Message中的方法信息在spring ApplicationContext中查找到spring 管理的bean，并通过反射来调用。

队列

队列抽象了一个BlockableQueue, 通过BlockableQueue具体实现来扩展，可以是内存，文件，或分布式MQ。

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
 public interface BlockableQueue<T> {
    String DefaultQueueName = "fileQueue";
    /**
     * push一个消息到队列
     *
     * @param t
     * @return
     */
    boolean offer(T t);
    /**
     * 从队列pop一个消息，如果队列中无可用消息，则阻塞
     *
     * @return
     * @throws InterruptedException
     */
    T take() throws InterruptedException;
    /**
     * 从队列pop一个消息，如果队列中无可用消息，则返回null
     *
     * @return
     */
    T poll();
    /**
     * 队列中消息数量
     *
     * @return
     */
    int size();
}

通用的默认实现：

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
 
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class DefaultBlockableQueue implements BlockableQueue<byte[]> {
    final Lock lock = new ReentrantLock();
    final Condition notEmpty = lock.newCondition();
    private Queue<byte[]> queue = null;
    public FileBlockableQueue(Queue<byte[]> queue) {
        this.queue = queue;
    }
    @Override
    public boolean offer(byte[] bytes) {
        lock.lock();
        try {
            boolean v = queue.offer(bytes);
            notEmpty.signal();
            return v;
        } finally {
            lock.unlock();
        }
    }
    @Override
    public byte[] take() throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == 0) {
                notEmpty.await();
            }
            byte[] bytes = queue.poll();
            return bytes;
        } finally {
            lock.unlock();
        }
    }
    @Override
    public byte[] poll() {
        return queue.poll();
    }
    @Override
    public int size() {
        return queue.size();
    }
}

本文实现了一个文件队列，采用去哪儿文件队列实现，这是一个fork：https://github.com/tietang/fqueue。

对编程模型来说不用关心异步细节，只需要在需要异步的方法上注解@AsyncExecutable即可。

| |

面试了很多人，每每问到“JDBC如何开启一个事务？”，大部分人的回答是：“通过openTransaction方法”，有的说是通过Connection，有的是说通过Statement，更有的说通过Connection拿到一个Transaction实例，再通过openTransaction方法来开启，那么同样关闭事务就有close方法“closeTransaction”。

想必这些人都是被“hibernate”害了还是“不注重java基础”，一上来就被框架误导呢？

先看看一个标准的JDBC例子伪代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
Connection conn = DriverManager.getConnection(...);
try{
  con.setAutoCommit(false);
  Statement stmt = con.createStatement();
   //1 or more queries or updates
   con.commit();
}catch(Exception e){
   con.rollback();
}finally{
   con.close();
}

所以，看到上面的例子，开启手动事务的关键是con.setAutoCommit(false)，JDBC事务默认是开启的，并且是自动提交：

  1. 关闭自动提交：java.sql.Connection.setAutoCommit(false)
    • setAutoCommit(true)：每次操作都会被认为是一个事务并且自动提交
  2. 手动提交事务：con.commit();
  3. 出现异常时回滚，不一定在catch语句中，只要在con.commit()前需要回滚时执行都可：con.rollback();
  4. 关闭连接：con.close();
  5. 设置事务隔离级别: java.sql.Connection#setTransactionIsolation

参考：https://docs.oracle.com/javase/tutorial/jdbc/basics/transactions.html

| |

Eureka的相关知识，在之前的《微服务之Eureka服务发现》中已经讲了很多，这里不再重复，本文主要通过Eureka Server源码和配置来阐述Eureka Server的工作原理。

Eureka提供了一系列REST的API，供Eureka Client来调用，实现服务注册，注销，心跳，状态更新等等操作，参考官网EurekaREst操作。

1
2
3
4
5
6
REST API <Jersey>
Response Cache <com.google.common.cache.LoadingCache>
InstanceRegistry <ConcurrentHashMap>
EvictionTimer<java.util.Timer>
CacheUpdateTask<java.util.Timer>

REST API

基于Jersey实现，主要以appId[appname]和instanceId为操作维度，内容可以是xml或者json。相关的实现可以在com.netflix.eureka.resources包中找到。基于appId和instanceId和各种操作组合的实现，可以认为是InstanceRegistry的操作入口。

InstanceRegistry

Registry是Eureka Server的核心，服务发现就是围绕Registry来实现。以下提到的类都可以在com.netflix.eureka.registry包中找到。

整个Registry由4个接口组成：

  • LeaseManager: register,cacel,renew,evict等基本操作
  • LookupService: 这个抽象是要是client和server端共用，EurekaClient也继承了该接口，主要用来查找服务和服务实例。
  • InstanceRegistry: 提供了实例相关的丰富的操作。
  • PeerAwareInstanceRegistry: eureka server之间的注册信息复制

如下是类关系图：

几个重要的定时任务

心跳补偿任务EvictionTask

1
[com.netflix.eureka.registry.AbstractInstanceRegistry.EvictionTask]

主要是用来做心跳补偿，目的是用来取消或清理过期的注册信息，通常是eureka client在停止前未成功发送cacel请求。例如eureka client停止时网络不通了、eureka client进程奔溃了等等。

这个定时任务是通过eureka.server.eviction-interval-timer-in-ms参数来配置处理间隔，默认是60s。

补偿时间=当前时间-该任务最后执行时间-执行间隔
其中该任务最后执行时间-执行间隔主要是计算出实际执行时间的细微差别，也为后面的补偿时间>0埋下伏笔。

补偿时间>0或者最后更新时间 + leaseDuration[默认是90s，客户端durationInSecs]+补偿时间<当前时间 时会执行evict清理过期实例。

补偿时间>0, 即就是当前任务执行和上次执行的时间间隔大于配置的时间间隔，正常情况补偿时间应该很小。

最后更新时间+leaseDuration+补偿时间<当前时间，即就是最后一次成功心跳到当前时间的间隔比eureka client配置的间隔大。

其中leaseDuration: 在eureka client中通过eureka.instance.lease-expiration-duration-in-seconds[leaseExpirationDurationInSeconds]参数来配置，默认是90s。

下面是源代码：

计算补偿时间：

1
2
3
4
5
6
7
8
9
10
11
long getCompensationTimeMs() {
            long currNanos = getCurrentTimeNano();
            long lastNanos = lastExecutionNanosRef.getAndSet(currNanos);
            if (lastNanos == 0l) {
                return 0l;
            }
            long elapsedMs = TimeUnit.NANOSECONDS.toMillis(currNanos - lastNanos);
            long compensationTime = elapsedMs - serverConfig.getEvictionIntervalTimerInMs();
            return compensationTime <= 0l ? 0l : compensationTime;
        }

这个是补偿清理逻辑：

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
public void evict(long additionalLeaseMs) {
        logger.debug("Running the evict task");
        if (!isLeaseExpirationEnabled()) {
            logger.debug("DS: lease expiration is currently disabled.");
            return;
        }
        // We collect first all expired items, to evict them in random order. For large eviction sets,
        // if we do not that, we might wipe out whole apps before self preservation kicks in. By randomizing it,
        // the impact should be evenly distributed across all applications.
        List<Lease<InstanceInfo>> expiredLeases = new ArrayList<>();
        for (Entry<String, Map<String, Lease<InstanceInfo>>> groupEntry : registry.entrySet()) {
            Map<String, Lease<InstanceInfo>> leaseMap = groupEntry.getValue();
            if (leaseMap != null) {
                for (Entry<String, Lease<InstanceInfo>> leaseEntry : leaseMap.entrySet()) {
                    Lease<InstanceInfo> lease = leaseEntry.getValue();
                    if (lease.isExpired(additionalLeaseMs) && lease.getHolder() != null) {
                        expiredLeases.add(lease);
                    }
                }
            }
        }
        // To compensate for GC pauses or drifting local time, we need to use current registry size as a base for
        // triggering self-preservation. Without that we would wipe out full registry.
        int registrySize = (int) getLocalRegistrySize();
        int registrySizeThreshold = (int) (registrySize * serverConfig.getRenewalPercentThreshold());
        int evictionLimit = registrySize - registrySizeThreshold;
        int toEvict = Math.min(expiredLeases.size(), evictionLimit);
        if (toEvict > 0) {
            logger.info("Evicting {} items (expired={}, evictionLimit={})", toEvict, expiredLeases.size(), evictionLimit);
            Random random = new Random(System.currentTimeMillis());
            for (int i = 0; i < toEvict; i++) {
                // Pick a random item (Knuth shuffle algorithm)
                int next = i + random.nextInt(expiredLeases.size() - i);
                Collections.swap(expiredLeases, i, next);
                Lease<InstanceInfo> lease = expiredLeases.get(i);
                String appName = lease.getHolder().getAppName();
                String id = lease.getHolder().getId();
                EXPIRED.increment();
                logger.warn("DS: Registry: expired lease for {}/{}", appName, id);
                internalCancel(appName, id, false);
            }
        }
    }

从上面的逻辑上看，配置的时间间隔有2个：

  • eureka server端：eviction-interval-timer-in-ms
  • eureka client端：lease-expiration-duration-in-seconds

这2个参数同时作用着清理逻辑，配置时就要注意，eviction-interval-timer-in-ms要比lease-expiration-duration-in-seconds配置的要小，产生的结果就完全不一样。

ResponseCache

通过readOnlyCache和readWriteCache实现：

  • readOnlyCache: java.util.concurrent.ConcurrentMap
  • readWriteCache: com.google.common.cache.LoadingCache

readWriteCache会自动从registry中更新。

这个缓存只作用于获取整个注册表和app的实例注册信息。

缓存更新CacheUpdateTask

这个任务用来定期更新只读缓存,逻辑上比较简单，定期从读写缓存中取出K/V，比较是否一致，不一致更新。

更新间隔通过参数eureka.server.response-cache-update-interval-ms来配置，默认是30s。

readWriteCache

过期时间通过参数eureka.server.response-cache-auto-expiration-in-seconds来配置，默认是180s。

eureka Server几个重要的配置

eureka.server.enable-self-preservation

是否开启自我保护模式，默认为true。

在开启状态下，Eureka Server会保持

默认情况下，如果Eureka Server在一定时间内没有接收到某个微服务实例的心跳，Eureka Server将会注销该实例（默认90秒）。但是当网络分区故障发生时，微服务与Eureka Server之间无法正常通信，以上行为可能变得非常危险了——因为微服务本身其实是健康的，此时本不应该注销这个微服务。

Eureka通过“自我保护模式”来解决这个问题——当Eureka Server节点在短时间内丢失过多客户端时（可能发生了网络分区故障），那么这个节点就会进入自我保护模式。一旦进入该模式，Eureka Server就会保护服务注册表中的信息，不再删除服务注册表中的数据（也就是不会注销任何微服务）。当网络故障恢复后，该Eureka Server节点会自动退出自我保护模式。

综上，自我保护模式是一种应对网络异常的安全保护措施。它的架构哲学是宁可同时保留所有微服务（健康的微服务和不健康的微服务都会保留），也不盲目注销任何健康的微服务。使用自我保护模式，可以让Eureka集群更加的健壮、稳定。

1
2
3
4
5
6
7
8
9
10
11
12
#是否开启自我保护模式，默认为true。
enableSelfPreservation: true
#默认是85%
renewal-percent-threshold: 0.85
#默认是15分钟
renewal-threshold-update-interval-ms: 15
#缓存更新时间，默认30s
response-cache-update-interval-ms: 10
#缓存过期时间，默认180s
response-cache-auto-expiration-in-seconds: 30
# 实例过期清理时间间隔，默认60秒
eviction-interval-timer-in-ms: 10

eureka client几个重要的配置

eureka.client.registry-fetch-interval-seconds

表示eureka client间隔多久去拉取服务注册信息，默认为30秒，对于api-gateway，如果要迅速获取服务注册状态，可以缩小该值，比如5秒

eureka.instance.lease-expiration-duration-in-seconds

leaseExpirationDurationInSeconds，表示eureka server至上一次收到client的心跳之后，等待下一次心跳的超时时间，在这个时间内若没收到下一次心跳，则将移除该instance。

默认为90秒
如果该值太大，则很可能将流量转发过去的时候，该instance已经不存活了。
如果该值设置太小了，则instance则很可能因为临时的网络抖动而被摘除掉。
该值至少应该大于leaseRenewalIntervalInSeconds
eureka.instance.lease-renewal-interval-in-seconds

leaseRenewalIntervalInSeconds，表示eureka client发送心跳给server端的频率。如果在leaseExpirationDurationInSeconds后，server端没有收到client的心跳，则将摘除该instance。除此之外，如果该instance实现了HealthCheckCallback，并决定让自己unavailable的话，则该instance也不会接收到流量。

| |

安装Nginx Lua模块

环境准备：

$ yum -y install pcre-devel

$ yum -y install openssl openssl-devel

下载所需文件

亦可参考官方安装指南：lua-nginx-module Installation

这是我总结的安装，供参考：

需要最新版的Nginx，LuaJIT，ngx_devel_kit，lua-nginx-module等安装文件:

参考命令下载：

$ curl -O http://nginx.org/download/nginx-1.10.1.tar.gz

$ curl -O http://luajit.org/download/LuaJIT-2.1.0-beta2.tar.gz

$ curl -L -O https://github.com/simpl/ngx_devel_kit/archive/v0.2.19.tar.gz

$ curl -L -O https://github.com/openresty/lua-nginx-module/archive/v0.10.5.tar.gz

安装LuaJIT

$ wget http://luajit.org/download/LuaJIT-.tar.gz

$ tar zxvf LuaJIT-.tar.gz

$ cd LuaJIT-

$ make

$ sudo make install

编译安装lua-nginx-module

1
2
3
4
5
6
7
8
export LUAJIT_LIB=/usr/local/lib
export LUAJIT_INC=/usr/local/include/luajit-2.1
cd nginx-1.10.1
./configure --prefix=/opt/nginx  --with-ld-opt="-Wl,-rpath,/usr/local/lib"  --add-module=/path/to/ngx_devel_kit-0.3.0  --add-module=/path/to/nginx/lua-nginx-module-0.10.5
make
make install

动态module方式

Nginx 1.9.11 开始可以编译module为一个动态module，在执行./configure命令时用–add-dynamic-module=PATH替换–add-module=PATH。编译后可以在nginx.conf配置中使用 load_module 来动态加载这个module：

load_module /path/to/modules/ndk_http_module.so;

load_module /path/to/modules/ngx_http_lua_module.so;

lua cjson install

Eureka Zuul 网关
安装编译cjson需要lua运行时，编译前需要修改MakeFile中的LUA_VERSION、LUA_INCLUDE_DIR、LUA_CMODULE_DIR参数，让cjson模块知道lua库文件和头文件目录，才可以正常编译安装。

下载解压：

1
2
3
wget http://www.kyne.com.au/~mark/software/download/lua-cjson-2.1.0.tar.gz
tar -zxvf cjson-2.1.0.tar.gz
cd cjson-2.1.0

修改MakeFile：

1
2
3
4
5
6
7
LUA_VERSION =  luajit-2.1
LUA_INCLUDE_DIR =   $(PREFIX)/include/$(LUA_VERSION)
LUA_CMODULE_DIR =   $(PREFIX)/lib
make & sudo make install
sudo cp cjson.so /usr/local/lib/lua/5.1/

| ， |

Hystrix semaphore和thread隔离策略的区别及配置参考

通用设置说明

Hystrix所有的配置都是hystrix.command.[HystrixCommandKey]开头,其中[HystrixCommandKey]是可变的，默认是default,即hystrix.command.default；另外Hystrix内置了默认参数，如果没有配置Hystrix属性，默认参数就会被设置，其优先级：

  • hystrix.command.[HystrixCommandKey].XXX
  • hystrix.command.default.XXX
  • Hystrix代码内置属性参数值

Hystrix隔离策略相关的参数

策略参数设置

execution.isolation.strategy= THREAD|SEMAPHORE

execution.isolation.thread.timeoutInMilliseconds

hystrix.command.default.execution.isolation.thread.timeoutInMilliseconds用来设置thread和semaphore两种隔离策略的超时时间，默认值是1000。

  • 建议设置这个参数，在Hystrix 1.4.0之前，semaphore-isolated隔离策略是不能超时的，从1.4.0开始semaphore-isolated也支持超时时间了。
  • 建议通过CommandKey设置不同微服务的超时时间,对于zuul而言，CommandKey就是service id：hystrix.command.[CommandKey].execution.isolation.thread.timeoutInMilliseconds

这个超时时间要根据CommandKey所对应的业务和服务器所能承受的负载来设置，要根据CommandKey业务的平均响应时间设置，一般是大于平均响应时间的20%~100%,最好是根据压力测试结果来评估，这个值设置太大，会导致线程不够用而会导致太多的任务被fallback；设置太小，一些特殊的慢业务失败率提升，甚至会造成这个业务一直无法成功，在重试机制存在的情况下，反而会加重后端服务压力。

根据微服务实际情况来定：

  • 太小：会导致很多正常业务失败
  • 太大：会导致实际的熔断效果很差，严重会导致雪崩。

一般实际大小为：

  • 要保证该服务的可用性为几个9？99.5 99.9 99.99
  • 要保证的N个9的最大响应时间。

execution.isolation.semaphore.maxConcurrentRequests

这个值并非TPS、QPS、RPS等都是相对值，指的是1秒时间窗口内的事务/查询/请求，semaphore.maxConcurrentRequests是一个绝对值，无时间窗口，相当于亚毫秒级的，指任意时间点允许的并发数。当请求达到或超过该设置值后，其其余就会被拒绝。默认值是100。

execution.timeout.enabled

是否开启超时，默认是true，开启。

execution.isolation.thread.interruptOnTimeout

发生超时是是否中断线程，默认是true。

execution.isolation.thread.interruptOnCancel

取消时是否中断线程，默认是false。

| ， |

zuul 参数调优

适用版本：
spring-boot: 1.4.x.RELEASE
spring-cloud：Camden.SR3
Hystrix: 1.5.6

spring-boot-tomcat 优化参数：

主要只有2个，最大和最小worker线程：

1
2
server.tomcat.max-threads=128 # Maximum amount of worker threads.
server.tomcat.min-spare-threads=64 # Minimum amount of worker threads.

spring-boot-undertow 优化参数：

ioThreads

设置IO线程数, 它主要执行非阻塞的任务,它们会负责多个连接,默认取CPU核心数量,最小值为2。
Math.max(Runtime.getRuntime().availableProcessors(), 2);
spring-boot 参数：server.undertow.io-threads=

worker-threads

阻塞任务线程池, 当执行类似servlet请求阻塞操作, undertow会从这个线程池中取得线程,它的值设置取决于系统的负载，默认值为io-threads*8。

spring-boot 参数：server.undertow.worker-threads=

buffer

buffer-size:

每块buffer的空间大小,越小的空间被利用越充分。

buffers-per-region:

每个区分配的buffer数量 , 所以pool的大小是buffer-size * buffers-per-region。

directBuffers

是否分配的直接内存。

获取JVM最大可用内存maxMemory=Runtime.getRuntime().maxMemory();

maxMemory<64M,不开启directBuffers， bufferSize = 512,buffersPerRegion = 10;

64<=maxMemory<128M,开启directBuffers， bufferSize = 1024 bytes,buffersPerRegion = 10;

maxMemory>128M,开启directBuffers， bufferSize = 16*1024 bytes,buffersPerRegion = 20;

spring-boot 参数：

1
2
3
4
5
6
7
8
# 最大可用内存<64M,不开启
server.undertow.buffer-size= # Size of each buffer in bytes.
server.undertow.buffers-per-region= # Number of buffer per region.
server.undertow.direct-buffers= # Allocate buffers outside the Java heap.
//默认值：cpu数量，最小为2
server.undertow.io-threads= # Number of I/O threads to create for the worker.
//默认值：io-threads*8
server.undertow.worker-threads= # Number of worker threads.

zuul 内置参数

zuul.host.maxTotalConnections

适用于ApacheHttpClient，如果是okhttp无效。每个服务的http客户端连接池最大连接，默认是200.

zuul.host.maxPerRouteConnections

适用于ApacheHttpClient，如果是okhttp无效。每个route可用的最大连接数，默认值是20。

zuul.semaphore.max-semaphores

Hystrix最大的并发请求execution.isolation.semaphore.maxConcurrentRequests，这个值并非TPS、QPS、RPS等都是相对值，指的是1秒时间窗口内的事务/查询/请求，semaphore.maxConcurrentRequests是一个绝对值，无时间窗口，相当于亚毫秒级的。当请求达到或超过该设置值后，其其余就会被拒绝。默认值是100。参考: Hystrix semaphore和thread隔离策略的区别及配置参考

这个参数本来直接可以通过Hystrix的命名规则来设置，但被zuul重新设计了，使得在zuul中semaphores的最大并发请求有4个方法的参数可以设置，如果4个参数都存在优先级（1~4）由高到低：

  • [优先级1]zuul.eureka.api.semaphore.maxSemaphores
  • [优先级2]zuul.semaphore.max-semaphores
  • [优先级3]hystrix.command.api.execution.isolation.semaphore.maxConcurrentRequests
  • [优先级4]hystrix.command.default.execution.isolation.semaphore.maxConcurrentRequests

需要注意的是：在Camden.SR3版本的zuul中hystrix.command.default.execution.isolation.semaphore.maxConcurrentRequests设置不会起作用，这是因为在org.springframework.cloud.netflix.zuul.filters.ZuulProperties.HystrixSemaphore.maxSemaphores=100设置了默认值100，因此zuul.semaphore.max-semaphores的优先级高于hystrix.command.default.execution.isolation.semaphore.maxConcurrentRequests。

zuul.eureka.[commandKey].semaphore.maxSemaphores：
其中commandKey为

参考设置参数：

1
2
3
4
5
6
#
zuul.host.maxTotalConnections: 200
zuul.host.maxPerRouteConnections: 10
#zuul.semaphore.max-semaphores: 128
# 建议使用这种方式来设置，可以给每个不同的后端微服务设置不同的信号量
zuul.eureka.[service id].semaphore.maxSemaphores: 128

其他Hystrix参数：

hystrix.command.default.execution.isolation.thread.timeoutInMilliseconds用来设置thread和semaphore两种隔离策略的超时时间，默认值是1000。

  • 建议设置这个参数，在Hystrix 1.4.0之前，semaphore-isolated隔离策略是不能超时的，从1.4.0开始semaphore-isolated也支持超时时间了。
  • 建议通过CommandKey设置不同微服务的超时时间,对于zuul而言，CommandKey就是service id：hystrix.command.[CommandKey].execution.isolation.thread.timeoutInMilliseconds

ribbon参数

1
2
3
4
5
6
7
8
9
10
11
12
13
ribbon:
#  # Max number of next servers to retry (excluding the first server)
#  MaxAutoRetries: 1
#  # Whether all operations can be retried for this client
#  MaxAutoRetriesNextServer: 1
#  # Interval to refresh the server list from the source
#  OkToRetryOnAllOperations: true
#  # Interval to refresh the server list from the source
#  ServerListRefreshInterval: 2000
#  # Connect timeout used by Apache HttpClient
  ConnectTimeout: 3000
#  # Read timeout used by Apache HttpClient
  ReadTimeout: 3000

主要是ConnectTimeout和ReadTimeout2个参数，最终会设置到http Client中。

注意这个参数很重要了，会配合execution.isolation.thread.timeoutInMilliseconds一起来用，多少合适要根据微服务实际情况来定：

  • 太小：会导致很多正常业务失败
  • 太大：会导致实际的熔断效果很差，严重会导致雪崩。

一般实际大小为：

  • 要保证该服务的可用性为几个9？99.5 99.9 99.99
  • 要保证的N个9的最大响应时间。

| |

maxwell 相关资源

http://maxwells-daemon.io/
https://github.com/zendesk/maxwell
https://github.com/zendesk/open-replicator

配置MySQL master数据源

1
2
3
4
[mysqld]
server-id=1
log-bin=master
binlog_format=row

注意：

  1. MySQL必须开启了binlogs，即log-bin指定了目录
  2. binlog_format必须是row

master数据源配置REPLICATION权限：

Maxwell需要储存他自己的一些状态数据，启动参数schema_database选型来指定，默认是maxwell.

1
2
GRANT ALL on maxwell.* to 'maxwell'@'%' identified by '123456';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'%';

问题列表

  1. 当binlog文件不存在时（被删除、移除、过期）
    • 无法启动maxwell
    • 正在运行的maxwell可能会stop

在阿里云RDS下的风险问题

  1. binlog文件清理问题
  2. binlog文件名在切换master主备或者阿里运维维护时会重置

RDS for MySQL 的 Binlog 生成和清理规则：

参考：RDS for MySQL 之 Binlog 日志生成和清理规则

其他问题

  1. 阿里RDS的binlog在被复制完成后，会将之前的最后的binlog文件复制到其他地方，如果maxwell挂起时间较长，有可能未复制的binlog不能被复制过来。

当master中的binlog文件被删除后，无法启动

1
2
3
4
5
6
7
8
9
10
11
09:35:06,785 ERROR MaxwellReplicator - Missing binlog 'mysql-bin.007213' on rdst5ai4d32fe3qd6if46.mysql.rds.aliyuncs.com
09:35:06,785 ERROR MaxwellReplicator - Transport exception #1236
com.google.code.or.net.TransportException: Could not find first log file name in binary log index file
	at com.google.code.or.OpenReplicator.dumpBinlog(OpenReplicator.java:343)
	at com.google.code.or.OpenReplicator.start(OpenReplicator.java:115)
	at com.zendesk.maxwell.replication.MaxwellReplicator.startReplicator(MaxwellReplicator.java:140)
	at com.zendesk.maxwell.replication.MaxwellReplicator.beforeStart(MaxwellReplicator.java:155)
	at com.zendesk.maxwell.util.RunLoopProcess.runLoop(RunLoopProcess.java:29)
	at com.zendesk.maxwell.Maxwell.start(Maxwell.java:160)
	at com.zendesk.maxwell.Maxwell.main(Maxwell.java:181)
09:35:06,786 INFO  Maxwell - starting shutdown