解决高并发相关问题

超、凢脫俗 2022-12-04 07:36 202阅读 0赞

## 一、用于解决高并发下System.currentTimeMillis卡顿 ##

import java.util.concurrent.Executors;
    import java.util.concurrent.ScheduledExecutorService;
    import java.util.concurrent.ThreadFactory;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicLong;
    
    /** * 用于解决高并发下System.currentTimeMillis卡顿 */
    public class SystemClock { 
    
        private final int period;
    
        private final AtomicLong now;
    
        private static class InstanceHolder { 
            private static final SystemClock INSTANCE = new SystemClock(1);
        }
    
        private SystemClock(int period) { 
            this.period = period;
            this.now = new AtomicLong(System.currentTimeMillis());
            scheduleClockUpdating();
        }
    
        private static SystemClock instance() { 
            return InstanceHolder.INSTANCE;
        }
    
        private void scheduleClockUpdating() { 
            ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(runnable -> { 
                Thread thread = new Thread(runnable, "System Clock");
                thread.setDaemon(true);
                return thread;
            });
            scheduler.scheduleAtFixedRate(() -> now.set(System.currentTimeMillis()), period, period, TimeUnit.MILLISECONDS);
        }
    
        private long currentTimeMillis() { 
            return now.get();
        }
    
        /** * 用来替换原来的System.currentTimeMillis() */
        public static long now() { 
            return instance().currentTimeMillis();
        }
    }

## 二、解决取cpu核数，新版jdk在docker里取的就是真实分配的，老版jdk取的是宿主机的，可能特别大，如32核 ##

public class CpuNum { 
    
        /** * netty worker线程数量. cpu密集型 */
        public static int workerCount() { 
            //取cpu核数，新版jdk在docker里取的就是真实分配的，老版jdk取的是宿主机的，可能特别大，如32核
            int count = Runtime.getRuntime().availableProcessors();
            if (isNewerVersion()) { 
                return count;
            } else { 
                count = count / 2;
                if (count == 0) { 
                    count = 1;
                }
            }
            return count;
        }
    
        public static void main(String[] args) { 
            System.out.println(isNewerVersion());
            System.out.println(Runtime.getRuntime().availableProcessors());
        }
    
    
        private static boolean isNewerVersion() { 
            try { 
                //如1.8.0_20, 1.8.0_181,1.8.0_191-b12
                String javaVersion = System.getProperty("java.version");
                //1.8.0_191之前的java版本，在docker内获取availableProcessors的数量都不对，会取到宿主机的cpu数量，譬如宿主机32核，
                //该docker只分配了4核，那么老版会取到32，新版会取到4。
                //线上事故警告！！！！！！老版jdk取到数值过大，线程数太多，导致cpu瞬间100%，大量的线程切换等待
                //先取前三位进行比较
                String topThree = javaVersion.substring(0, 5);
                if (topThree.compareTo("1.8.0") > 0) { 
                    return true;
                } else if (topThree.compareTo("1.8.0") < 0) { 
                    return false;
                } else { 
                    //前三位相等，比小版本. 下面代码可能得到20，131，181，191-b12这种
                    String smallVersion = javaVersion.replace("1.8.0_", "");
                    //继续截取，找"-"这个字符串，把后面的全截掉
                    if (smallVersion.contains("-")) { 
                        smallVersion = smallVersion.substring(0, smallVersion.indexOf("-"));
                    }
                    return Integer.valueOf(smallVersion) >= 191;
                }
            } catch (Exception e) { 
                return false;
            }
        }
    }

## 三、解决高并发数量统计都放到延迟队列里，10几秒后再处理 ##

import java.util.List;
    import java.util.concurrent.DelayQueue;
    import java.util.concurrent.Delayed;
    import java.util.concurrent.TimeUnit;
    
    /** * 将收到的数量统计都放到延迟队列里，10几秒后再处理 */
    public class KeyCountItem implements Delayed { 
        private String appName;
        private long createTime;
        private List<KeyCountModel> list;
    
        public KeyCountItem(String appName, long createTime, List<KeyCountModel> list) { 
            this.appName = appName;
            this.createTime = createTime;
            this.list = list;
        }
    
        public KeyCountItem() { 
        }
    
        public long getCreateTime() { 
            return createTime;
        }
    
        public void setCreateTime(long createTime) { 
            this.createTime = createTime;
        }
    
        public String getAppName() { 
            return appName;
        }
    
        public void setAppName(String appName) { 
            this.appName = appName;
        }
    
        public List<KeyCountModel> getList() { 
            return list;
        }
    
        public void setList(List<KeyCountModel> list) { 
            this.list = list;
        }
    
        @Override
        public String toString() { 
            return "KeyCountItem{" +
                    "appName='" + appName + '\'' +
                    ", createTime=" + createTime +
                    ", list=" + list +
                    '}';
        }
    
        @Override
        public long getDelay(TimeUnit unit) { 
            //固定延迟15秒后再处理
            return createTime - System.currentTimeMillis() + 15000;
        }
    
        @Override
        public int compareTo(Delayed o) { 
            KeyCountItem item = (KeyCountItem) o;
            long diff = this.createTime - item.getCreateTime();
            if (diff <= 0) { // 改成>=会造成问题
                return -1;
            } else { 
                return 1;
            }
        }
    
        class KeyCountModel { 
            /** * 对应的规则名，如 pin_2020-08-09 11:32:43 */
            private String ruleKey;
            /** * 总访问次数 */
            private int totalHitCount;
            /** * 热后访问次数 */
            private int hotHitCount;
            /** * 发送时的时间 */
            private long createTime;
        }
    
        public static void main(String[] args) { 
            KeyCountItem keyCountItem = new KeyCountItem();
            keyCountItem.setAppName("a");
            keyCountItem.setCreateTime(System.currentTimeMillis() + 2000);
            KeyCountItem keyCountItem1 = new KeyCountItem();
            keyCountItem1.setAppName("b");
            keyCountItem1.setCreateTime(System.currentTimeMillis() + 1000);
            KeyCountItem keyCountItem2 = new KeyCountItem();
            keyCountItem2.setAppName("c");
            keyCountItem2.setCreateTime(System.currentTimeMillis());
            DelayQueue<KeyCountItem> queue = new DelayQueue<>();
            queue.put(keyCountItem);
            queue.put(keyCountItem1);
            queue.put(keyCountItem2);
            for (int i = 0; i < 3; i++) { 
                try { 
                    System.out.println(queue.take());
                    System.out.println(System.currentTimeMillis());
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    }

## 四、解决List拿取特征元素值 ##

import java.util.Arrays;
    import java.util.Iterator;
    import java.util.List;
    import java.util.stream.Collectors;
    import java.util.stream.Stream;
    import java.util.stream.StreamSupport;
    
    public class StreamUtils { 
    
        private StreamUtils() { 
            throw new UnsupportedOperationException("Construct StreamUtils");
        }
    
        public static <T> Stream<T> asStream(Iterator<T> sourceIterator) { 
            return asStream(sourceIterator, false);
        }
    
        public static <T> Stream<T> asStream(Iterator<T> sourceIterator, boolean parallel) { 
            Iterable<T> iterable = () -> sourceIterator;
            return StreamSupport.stream(iterable.spliterator(), parallel);
        }
    
        public static void main(String[] args) { 
            List<String> list = Arrays.asList("a", "b", "c");
            List<String> ret = StreamUtils.asStream(list.iterator())
                    .filter(item -> item.equals("a"))
                    .collect(Collectors.toList());
            System.out.println(JSONUtils.toJsonString(ret));
        }
    }

## 五、解决网络延迟或者不稳定、采用重试方法 ##

依赖：可能存在依赖的guava低或者高版本、使用时候排除guava

<dependency>
                <groupId>com.google.guava</groupId>
                <artifactId>guava</artifactId>
                <version>29.0-jre</version>
            </dependency>
    
            <dependency>
                <groupId>com.github.rholder</groupId>
                <artifactId>guava-retrying</artifactId>
                <version>2.0.0</version>
                <exclusions>
                    <exclusion>
                        <artifactId>slf4j-api</artifactId>
                        <groupId>org.slf4j</groupId>
                    </exclusion>
                    <exclusion>
                        <groupId>org.slf4j</groupId>
                        <artifactId>slf4j-log4j12</artifactId>
                    </exclusion>
                    <exclusion>
                        <groupId>ch.qos.logback</groupId>
                        <artifactId>logback-classic</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>

**guava-retrying** 的核心是Attempt类、Retryer类以及一些Strategy(策略)相关的类。  
1、Attempt既是一次重试请求(call)，也是请求的结果，并记录了当前请求的次数、是否包含异常和请求的返回值。  
2、Retryer通过RetryerBuilder这个工厂类进行构造。RetryerBuilder负责将定义的重试策略赋值到Retryer对象中。

import com.github.rholder.retry.Retryer;
    import com.github.rholder.retry.RetryerBuilder;
    import com.github.rholder.retry.StopStrategies;
    import com.google.common.base.Predicates;
    import lombok.extern.slf4j.Slf4j;
    import java.util.concurrent.Callable;
    
    @Slf4j
    public class GuavaRetrying { 
        /** * 重试策略 */
        private Retryer<Boolean> reTrying = RetryerBuilder.<Boolean>newBuilder()
                // retryIf 重试条件 如果Callable结果为null,继续重试
                .retryIfResult(Predicates.isNull())
                // IOException,继续重试
                .retryIfExceptionOfType(RuntimeException.class)
                // RuntimeException,继续重试
                .retryIfRuntimeException()
                // 指定重试策略,重试三次后停止
                .withStopStrategy(StopStrategies.stopAfterAttempt(3))
                .build();
        /** * 定义请求实现 */
        private Callable<Boolean> callable = new Callable<Boolean>() { 
            private int i = 1;
            @Override
            public Boolean call() { 
                System.out.println("重试次数：" + i);
                if (i==1){ 
                    i++;
                    throw new NullPointerException();
                }
                if (i == 2){ 
                    return true;
                }
                return true;
            }
        };
        public static void main(String[] args) { 
            try { 
                GuavaRetrying guavaRetrying = new GuavaRetrying();
                Boolean call = guavaRetrying.reTrying.call(guavaRetrying.callable);
                System.out.println("call: " + call);
            } catch (Exception e) { 
              log.error("重试失败：{}",e);
            }
        }
    }

参考：[https://www.jianshu.com/p/557eb67bb3d8][https_www.jianshu.com_p_557eb67bb3d8]  
参考：[https://gitee.com/jd-platform-opensource/hotkey][https_gitee.com_jd-platform-opensource_hotkey]

[https_www.jianshu.com_p_557eb67bb3d8]: https://www.jianshu.com/p/557eb67bb3d8
[https_gitee.com_jd-platform-opensource_hotkey]: https://gitee.com/jd-platform-opensource/hotkey