手把手實作RPC架構--簡易版Dubbo構造（九）Hessian序列化、線程池工具

本節commit位址：1ed2159

上一節我們實作了Kryo序列化，抱着學習的心态，本節基于Hessian協定來實作序列化，實作步驟和之前的Kryo類似，關于Hessian和Kryo的差別可以參考另一篇：https://blog.csdn.net/qq_38685503/article/details/114633168?spm=1001.2014.3001.5501

本節還會把Socket方式傳輸中用到的線程池封裝成一個通用Util，要用到guava中的ThreadFactoryBuilder()，自定義線程名，友善出錯定位問題。

Hessian序列化器

public class HessianSerializer implements CommonSerializer{

    private static final Logger logger = LoggerFactory.getLogger(HessianSerializer.class);

    @Override
    public byte[] serialize(Object obj) {
        HessianOutput hessianOutput = null;
        try(ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()){
            hessianOutput = new HessianOutput(byteArrayOutputStream);
            hessianOutput.writeObject(obj);
            return byteArrayOutputStream.toByteArray();
        }catch (IOException e){
            logger.error("序列化時有錯誤發生" + e);
            throw new SerializeException("序列化時有錯誤發生");
        }finally {
            if(hessianOutput != null){
                try {
                    hessianOutput.close();
                }catch (IOException e){
                    logger.error("關閉output流時有錯誤發生" + e);
                }
            }
        }
    }

    @Override
    public Object deserialize(byte[] bytes, Class<?> clazz) {
        HessianInput hessianInput = null;
        try(ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes)){
            hessianInput = new HessianInput(byteArrayInputStream);
            return hessianInput.readObject();
        }catch (IOException e){
            logger.error("反序列化時有錯誤發生" + e);
            throw new SerializeException("反序列化時有錯誤發生");
        }finally {
            if(hessianInput != null) {
                hessianInput.close();
            }
        }
    }
           

線程池工具類

public class ThreadPoolFactory {
    /**
     * 線程池參數
     */
    private static final int CORE_POOL_SIZE = 10;
    private static final int MAXIMUM_POOL_SIZE = 100;
    private static final int KEEP_ALIVE_TIME = 1;
    private static final int BLOCKING_QUEUE_CAPACITY = 100;

    private ThreadPoolFactory(){
    }

    public static ExecutorService createDefaultThreadPool(String threadNamePrefix){
        return createDefaultThreadPool(threadNamePrefix, false);
    }

    public static ExecutorService createDefaultThreadPool(String threadNamePrefix, Boolean daemon){
        /**
         * 設定上限為100個線程的阻塞隊列
         */
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(BLOCKING_QUEUE_CAPACITY);
        ThreadFactory threadFactory = createThreadFactory(threadNamePrefix, daemon);
        return new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.MINUTES, workQueue, threadFactory);
    }

    /**
     * @description 建立ThreadFactory，如果threadNamePrefix不為空則使用自建ThreadFactory，否則使用defaultThreadFactory
     * @param threadNamePrefix 作為建立的線程名字的字首，指定有意義的線程名稱，友善出錯時回溯
     * @param daemon 指定是否為Daemon Thread(守護線程)，當所有的非守護線程結束時，程式也就終止了，同時會殺死程序中的所有守護線程
     * @return [java.util.concurrent.ThreadFactory]
     * @date [2021-03-10 17:50]
     */
    private static ThreadFactory createThreadFactory(String threadNamePrefix, Boolean daemon) {
        if (threadNamePrefix != null) {
            if (daemon != null) {
                //利用guava中的ThreadFactoryBuilder自定義建立線程工廠
                return new ThreadFactoryBuilder().setNameFormat(threadNamePrefix + "-%d").setDaemon(daemon).build();
            } else {
                return new ThreadFactoryBuilder().setNameFormat(threadNamePrefix + "-%d").build();
            }
        }
        return Executors.defaultThreadFactory();
    }
}
           

本節over……