[hadoop] hadoop性能优化模拟读写分离核心代码

雨点打透心脏的1/2处 2021-11-17 10:31 250阅读 0赞

package com.simulation.core;
    
    import java.util.LinkedList;
    import java.util.concurrent.atomic.LongAdder;
    /***
    AtomicLong的实现原理是：利用底层操作系统的CAS来保证原子性，在一个死循环内不断执行CAS操作，直到操作成功。不过，CAS操作的一个问题是在并发量比较大的时候，可能很多次的执行CAS操作都不成功，这样性能就受到较大影响。 
    那我们知道，在ConcurrentHashMap中，对Map分割成多个segment，这样多个Segment的操作就可以并行执行，从而可以提高性能。在JDK8中，LongAdder与ConcurrentHashMap类似，将内部操作数据value分离成一个Cell数组，每个线程访问时，通过Hash等算法映射到其中一个Cell上。 
    计算最终的数据结果，则是各个Cell数组的累计求和。
    */
    public class runHadoop {
        /**
         * 测试
         * 1000个线程
         * 刷10000条log
         * */
        public static void main(String[] args) {
            runHadoop run = new runHadoop();
            for(int i = 0;i < 1000;i++){
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        for(int j = 0;j < 10000; j++){
                            run.logEdit("日志");
                        }
                    }
                }).start();
            }
    
        }
    
        //long taxID = 0L;
        DoubleBuffer doubleBuffer = new DoubleBuffer();
        //每个线程拥有自己的副本
        ThreadLocal<Long> threadLocal = new ThreadLocal<Long>();
        //是否后台正在把数据同步到磁盘上
        public boolean isSyncRunning = false;
        //正在同步磁盘的内存块里面最大的一个ID号
        long maxtaxid = 0L;
        boolean isWait = false;
        LongAdder taxID=new LongAdder();
        /**
         * 写元数据的核心方法
         * */
        private void logEdit(String log){
            taxID.increment();
            synchronized (this){
                threadLocal.set(taxID.longValue());
                Edlog edlog = new Edlog(taxID.longValue(),log);
                //往内存里写东西
                doubleBuffer.write(edlog);
            }//释放锁
            //把数据持久化到磁盘
    
            //重新枷锁
            logFlush();
        }
        private void logFlush(){
            synchronized(this){
                //多线程同步
                if(isSyncRunning){//false
                    //获取当前事务Id
                    long localTaxID = threadLocal.get();
                    //2 < 3
                    if(localTaxID <= maxtaxid){ //数据是否一样
                        return;
                    }
                    if (isWait){    //是否有线程在写
                        return;
                    }
                    isWait = true; //标记在写
                    while (isSyncRunning){
                        try{
                            this.wait(1000); //释放锁
                        }catch (InterruptedException e){
                            e.printStackTrace();
                        }
                    }
                    isWait = false;
                }
                //程序直接走到这
                doubleBuffer.exchange();//交换内存
                //maxTaxid = 3 ;
                if(doubleBuffer.syncBuffer.size() > 0){
                    maxtaxid = doubleBuffer.getMaxTaxid();
                }
                isSyncRunning = true;//
            }//释放锁
            //数据持久化到磁盘 IO比较慢费性能
            doubleBuffer.flush();
            synchronized(this){
                isSyncRunning = false; //修改标志位
                this.notifyAll(); //唤醒线程
            }
        }
        /**
         *
         * edLog日志对象
         */
        class Edlog{
    
            private long taxID;
            private String log;
    
            public Edlog(long taxID,String log) {
                this.taxID = taxID;
                this.log = log;
            }
    
            @Override
            public String toString() {
                return "Edlog{" +
                        "taxID=" + taxID +
                        ", log='" + log + '\'' +
                        '}';
            }
        }
    
        /**
         * 日志读写过程
         * */
        class DoubleBuffer{
    
            //写数据 有序队列
            LinkedList<Edlog> currentBuffer = new LinkedList<>();
            //持久数据化
            LinkedList<Edlog> syncBuffer = new LinkedList<>();
    
    
            /**
             * 写元数据
             * */
            public void write(Edlog log){
                currentBuffer.add(log);
            }
    
            /**
             * 持久化到磁盘
             * */
            public void flush(){
                for(Edlog log:syncBuffer){
                    System.out.println(log); //模拟打印出来
                }
                syncBuffer.clear(); //释放内存
            }
    
            /**
             * 读写分离
             * 交换内存
             */
            public void exchange(){
                LinkedList<Edlog> tmp = currentBuffer;
                currentBuffer = syncBuffer;
                syncBuffer = tmp;
            }
            /**
             * 获取正在同步的内存中最大的事务ID
             * */
            public long getMaxTaxid(){
                return syncBuffer.getLast().taxID;
            }
        }
    }