[practice-mistake] 并发工具类库

[Alice52]

1. Juc

1. ThreadLocal: 线程池中使用之后没有 remove, 导致用户信息错乱, 且会导致内存泄漏
2. ConcurrentHashMap 只保证提供原子性的读写操作是线程安全的
   • containsKey/size/isEmpty/containsValue/putAll 等都有可能获取中间态数据, 不能作为流程控制
   • sample: 统计大文本每个key出现的次数{CHM+LongAdder+putInAbsent+多线程}
3. CopyOnWriteArrayList 在大量写的情况下会有性能问题
   • 修改时会先复制出来一份
4. Map 方法
   • putIfAbsent: HashMap 可以KV 为null, CHM 不可以为null会直接报错
   • computeIfAbsent: Functional 为null 则不会真正的put 进去

[Alice52]

2. Lock

1. 8 lock
   • 静态字段问题
   • 同步方法与非同步方法
2. 锁的粒度和场景问题
   • 尽量小的锁粒度{范围} + 区分读写场景/资源访问冲突
   • 读多写少的 ReentrantReadWriteLock: 共享度互斥写, 读写冲突{悲观的读锁}
   • 冲突概率不大可以使用 StampedLock: 也是读写锁, 乐观的读锁
   • 没有明确的需求不要开启公平锁: 效率很低{尤其在任务很轻时}
3. 死锁
   • 减小锁粒度就会带有不同的粒度锁, 锁多了就有可能带来死锁问题
   • sample: 多用户并发下单多个商品, 可能出现死锁{对商品排序之后获取锁}
   • 一般性方案: 避免无限等待和循环等待
   • 锁是否配对及对应的释放
   • 超时自动释放, 耗时逻辑时的锁自动续期, 重复执行等

[Alice52]

3. 线程池

1. 使用 Executors 创建线程池会有问题: OOM
2. 线程池可以在一个项目中根据不同的场景创建多个, 避免混用: 每个线程池中线程数量都不应该太大[1~100] cpu核心数相关
3. ForkJoinPool 的默认并发数是 CPU-1, 适合Cpu密集型; 如果是IO密集型建议新建一个pool
4. tips
   • threadPool.allowCoreThreadTimeOut(true): core size 可以被回收
   • threadPool.prestartAllCoreThreads(): 优先初始化core size
5. 线程池应该尽量有相关的监控
6. 使用激进的线程创建, 改变之前的先入queue之后才开辟到max-pool-size
   • offer: response false
   • reject 中put queue

   public void execute(Runnable command) {
       if (command == null)
           throw new NullPointerException();
   
       int c = ctl.get();
       // 当前线程总数小于 core-pool-size 则创建线程执行
       if (workerCountOf(c) < corePoolSize) {
           if (addWorker(command, true))
               return;
           c = ctl.get();
       }
       // core-pool-size 最大则入 queue
       if (isRunning(c) && workQueue.offer(command)) {
           int recheck = ctl.get();
           if (! isRunning(recheck) && remove(command))
               reject(command);
           else if (workerCountOf(recheck) == 0)
               addWorker(null, false);
       }
       // 如果入 queue 失败或者使用 SynchronousQueue 则执行 reject
       else if (!addWorker(command, false))
           reject(command);
   }
   
   BlockingQueue<Runnable> queue =
           new LinkedBlockingQueue<Runnable>(10) {
               @Override
               public boolean offer(Runnable e) {
                   // 先返回false，造成队列满的假象，让线程池优先扩容
                   return false;
               }
           };
   RejectedExecutionHandler handler =
           (r, executor) -> {
               try {
                   // 等出现拒绝后再加入队列,
                   // 如果希望队列满了阻塞线程而不是抛出异常，那么可以注释掉下面三行代码，修改为 executor.getQueue().put(r);
                   if (!executor.getQueue().offer(r, 0, TimeUnit.SECONDS)) {
                       throw new RejectedExecutionException(
                               "ThreadPool queue full, failed to offer " + r.toString());
                   }
               } catch (InterruptedException e) {
                   Thread.currentThread().interrupt();
               }
           };

[Alice52]

4. 连接池

1. 作用
   

2. 常见的连接池

   • mysql连接池
   • redis连接池
   • http连接池

3. 创建的3种实现池化: 一般约束

   • 连接池和链接分离的api: JedisPool
     1. 有 XxxPool[线程安全] 类负责连接池实现: XxxConnection[非线程安全]
     2. 使用端先获取连接, 使用, 归还给连接池: 对XxxPool内资源操作可以并发
   • 内部带有连接池的 api: RedisTemplate/CloseableHttpClient/@Transactional
     1. 有 XxxClient 使用: 内部维护线程池[使用时无需考虑获取与归还问题]
     2. XxxClient 一般会是线程安全的
   • 非连接池的api
     1. 有XxxConnection, 每次使用都需要创建和断开连接
     2. 所以性能很一般, 且一般都是线程不安全的

4. jedis: 一定要使用池化的, 否则使用一个 Connection 在并发下会导致很多问题{stream closed, socket closed, value unexpected}

   • connection 的实现本身就是非线程安全的
   • 使用线程池让每个线程都是用各自的 connection: jedis pool 维护另一个线程池{+stateStore}
     

5. 连接池使用

   • 确保连接池复用, 且实现线程安全
   • 尽量确保程序退出显示关闭连接池释放资源
   • 最大连接数非常重要

6. 超时

   // 假设我们希望设置 连接超时5s. 获取连接超时10s
   
   // 1. hikari两个参数设置方式
   spring.datasource.hikari.connection-timeout=10000
   spring.datasource.url=jdbc:mysql://localhost:6657/common_mistakes?connectTimeout=5000&characterEncoding=UTF-8&useSSL=false&rewriteBatchedStatements=true
   
   // 2. jedis两个参数设置
   JedisPoolConfig config = new JedisPoolConfig();
           config.setMaxWaitMillis(10000);
           try (JedisPool jedisPool = new JedisPool(config, "127.0.0.1", 6379, 5000);
                Jedis jedis = jedisPool.getResource()) {
               return jedis.set("test", "test");
           }
   
   // 3. httpclient两个参数设置
   RequestConfig requestConfig = RequestConfig.custom()
                   .setConnectTimeout(5000)
                   .setConnectionRequestTimeout(10000)
                   .build();

[Alice52]

• openfeign-mis-use
• feign-hystrix-concurrent
• hystrix-config

5. Http 超时/重试/并发

• 一般性认为连接超时是网络问题或者服务不在线; 读取超时是服务处理超时

1. 连接超时: ConnectTimeout, 建立连接的最长等待时间 [1-5s]

   • 误区1: 【错误】设置很长时间, tcp 三次握手建立连接需要的时间是很短的, 连接不上的话配置很长时间就是浪费
   • 误区2: 【事实】排查超时时, 需要注意上下游, 比如 nginx

2. 读取超时: ReadTimeout, 从 socket 上读取数据的最长等待时间 [1~30s]

   • 误区1: 【事实】读取超时, 服务端的执行是不会终止的
   • 误区2: 【错误】将 readtimeout 设置的很小[认为只是 socket网络层的]; 事实是发生读取超时时, 网络层面无法区分是服务端没有把数据返回给客户端, 还是数据在网络上耗时或丢包
   • 误区3: 【错误】认为任务超时时间越长成功率就越高, 所以讲超时配的很大
     1. http 请求是需要结果的, 属于同步调用: 异步任务定时任务可以超时配置大一些
     2. 如果超时很大, 当很多超时时会被大量的线程拖垮

3. feign & robbin 超时

   • feign 默认读取超时是 1s: 很坑

     # 修改 feign 自身的默认超时
     feign.client.config.default.readTimeout=3_000
     feign.client.config.default.connectTimeout=3_000
     # 配置指定的 FeignClient 的超时时间
     feign.client.config.CLIENGT_NAME.connectTimeout=3_000
     # 注意！！！: 这里如果只配置 readTimeout 是不会生效的, 必须同时配置: 新版本中已修复
   • ribbon 的超时

     # 修改 ribbon 自身的默认超时
     ribbon.ReadTimeout=4_000
     ribbon.ConnectTimeout=4_000
   • feign & robbin 两个都有配置以 feign 为准
   • robbin Get请求会默认重试一次, POST 不会: 可以通过 MaxAutoRetrieNextServer=0 配置

4. 并发数: 前期http1.0时有些实现框架对每个域名限制2个并发 defaultMaxPerRoute

5. 写入超时

   • 客户端发送数据到服务端, 首先接力连接[TCP], 然后写入TCP缓冲区
   • TCP缓冲区根据时间窗口, 发送数据到服务端
   • 因此写入操作可以任务是自己本地的操作，本地操作是不需要什么超时时间的
   • 其实也就是到socket sendbuffer的超时: 满的话等待空间释放的超时