import org.apache.shiro.crypto.SecureRandomNumberGenerator;
import org.apache.shiro.crypto.RandomNumberGenerator;
import org.apache.shiro.crypto.hash.SimpleHash;
import org.apache.shiro.util.ByteSource;

/**
  * 密码加密工具类
  */
public class PasswordHelper{
    //随机字符串生成器
    private RandomNumberGenerator randomOb = new RandomNumberGenerator();

    //散列算法名（md5方式）
    private static final String ALGORITHM_NAME = "md5";
    
    //散列迭代次数
    private static final int HASH_ITERATION = 2;

    /**
      *  加密用户
      * @param user 用户，用户名(name)、密码(pwd)、盐\加密因子(salt)
      */
    public void encryptPassword(User user){
        //生成加密因子，保存盐。
        user.setSalt(randomOb.nextBytes().toHex());
        //加密密码 SimpleHash（算法名，密码，盐的byte，次数）.toHex()
        String newPassword = new SimpleHash(ALGORITHM_NAME ,user.getPwd,ByteSource.Util.bytes(user.getSalt()),HASH_ITERATION).toHex();
        //更新密码
        user.setPwd(newPassword );
    }
}

/*
 * 为shiro提供登录验证和权限提取
 */
public class EnceladusShiroRealm extends AuthorizingRealm{
    //用户操作业务层
    @Autowired
    private UserService userService;

    /*
     * 提供用户角色和权限获取的逻辑
     */
    @Override
     protected AuthorizationInfo doGetAuthorizationInfo(PrincipalCollection principals) {
         //需要返回的权限信息
         SimpleAuthorizationInfo authorizationInfo = new SimpleAuthorizationInfo();
         String userName = (String) principals.getPrimaryPrincipal();
         User user = userService.findUserByName(userName);
         for (SysRole role : user.getRoles()) {
            //设置角色
            authorizationInfo.addRole(role.getRole());
            for (SysPermission permission : role.getPermissions()) {
                //设置权限
                authorizationInfo.addStringPermission(permission.getName());
            }
         }
         return authorizationInfo;
     }

     @Override
     protected AuthenticationInfo doGetAuthenticationInfo(AuthenticationToken token) throws AuthenticationException {
     //通过token获取用户名
     String userName = (String) token.getPrincipal();
     User user = userService.findUserByName(username);
     if (user == null) return null;
     SimpleAuthenticationInfo authenticationInfo = new SimpleAuthenticationInfo(
                                         user.getUsername(),
                                         user.getPassword(),
                                         ByteSource.Util.bytes(user.getCredentialsSalt()),
                                         getName());
      return authenticationInfo;
  }
}

@Configuration
public class ShiroConfig {
    //配置权限
    @Bean
    public ShiroFilterFactoryBean shirFilter(SecurityManager securityManager) {
        ShiroFilterFactoryBean shiroFilterFactoryBean = new ShiroFilterFactoryBean();
        shiroFilterFactoryBean.setSecurityManager(securityManager);

        Map<String, String> filterChainDefinitionMap = new HashMap<String, String>();
        shiroFilterFactoryBean.setLoginUrl("/login");
        shiroFilterFactoryBean.setUnauthorizedUrl("/unauthc");
        shiroFilterFactoryBean.setSuccessUrl("/home/index");
        
        filterChainDefinitionMap.put("/*", "anon");
        filterChainDefinitionMap.put("/authc/index", "authc");
        filterChainDefinitionMap.put("/authc/admin", "roles[admin]");
        filterChainDefinitionMap.put("/authc/renewable", "perms[Create,Update]");
        filterChainDefinitionMap.put("/authc/removable", "perms[Delete]");
        shiroFilterFactoryBean.setFilterChainDefinitionMap(filterChainDefinitionMap);
        return shiroFilterFactoryBean;
    }

    /*
     * 提供散列算法的信息
     */
    @Bean
    public HashedCredentialsMatcher hashedCredentialsMatcher() {
        HashedCredentialsMatcher hashedCredentialsMatcher = new HashedCredentialsMatcher();
        hashedCredentialsMatcher.setHashAlgorithmName(PasswordHelper.ALGORITHM_NAME); // 散列算法
        hashedCredentialsMatcher.setHashIterations(PasswordHelper.HASH_ITERATIONS); // 散列次数
        return hashedCredentialsMatcher;
    }

    /*
     * 设置加密算法
     */
    @Bean
    public EnceladusShiroRealm shiroRealm() {
        //自己实现的Realm
        EnceladusShiroRealm shiroRealm = new EnceladusShiroRealm();
        shiroRealm.setCredentialsMatcher(hashedCredentialsMatcher()); // 设置加密算法
        return shiroRealm;
    }

    /*
     * 设置安全管理类的realm为上面写的realm
     */
    @Bean
    public SecurityManager securityManager() {
        DefaultWebSecurityManager securityManager = new DefaultWebSecurityManager();
        securityManager.setRealm(shiroRealm());
        return securityManager;
    }

    @Bean
    public PasswordHelper passwordHelper() {
        return new PasswordHelper();
    }
}

import org.apache.shiro.SecurityUtils;
import org.apache.shiro.subject.Subject;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.learnhow.springboot.web.entity.User;

@RestController
@RequestMapping("authc")
public class AuthcController {

    @GetMapping("index")
    public Object index() {
        Subject subject = SecurityUtils.getSubject();
        User user = (User) subject.getSession().getAttribute("user");
        return user.toString();
    }

    @GetMapping("admin")
    public Object admin() {
        return "Welcome Admin";
    }

    // delete
    @GetMapping("removable")
    public Object removable() {
        return "removable";
    }

    // insert & update
    @GetMapping("renewable")
    public Object renewable() {
        return "renewable";
    }
}

import lombok.extern.slf4j.Slf4j;
import javax.net.ssl.*;
import java.io.*;
import java.net.Socket;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.security.KeyFactory;
import java.security.KeyStore;
import java.security.PrivateKey;
import java.security.SecureRandom;
import java.security.cert.Certificate;
import java.security.cert.CertificateException;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.security.spec.PKCS8EncodedKeySpec;

/***
 * ssl工具类
 * 
 * @author colin.cheng
 * @date 2021-02-03 14:39
 * @since 1.0.0
 */
@Slf4j
public class SslUtil {

    /**
     * 读取x509格式的证书
     * 
     * @param certPath
     * @return
     * @throws FileNotFoundException
     * @throws CertificateException
     */
    public static X509Certificate getX509Certificate(String certPath) throws Exception {
        try (InputStream inStream = new FileInputStream(certPath)) {
            return getX509Certificate(inStream);
        }
    }

    /**
     * 通过输入流获取证书
     * 
     * @param inStream
     * @return
     * @throws Exception
     */
    public static X509Certificate getX509Certificate(InputStream inStream) throws Exception {
        CertificateFactory cf = CertificateFactory.getInstance("X.509");
        X509Certificate caCert = (X509Certificate)cf.generateCertificate(inStream);
        return caCert;
    }

    /**
     * 从keystore中获取证书
     * 
     * @param keyStorePath keystore位置
     * @param keyStorePwd keyStore密码
     * @param keyStoreType keyStore类型
     * @param alias 证书存储的别名
     * @return
     * @throws Exception
     */
    public static Certificate getCertificate(String keyStorePath, String keyStorePwd, String keyStoreType, String alias)
        throws Exception {
        final KeyStore keyStore = getKeyStore(keyStorePath, keyStorePwd, keyStoreType);
        return keyStore.getCertificate(alias);
    }

    /**
     * 获取KeyStore
     * 
     * @param keyStorePath 路径
     * @param keyStorePwd 密码
     * @param keyStoreType 类型
     * @return
     */
    public static KeyStore getKeyStore(String keyStorePath, String keyStorePwd, String keyStoreType) throws Exception {
        try (InputStream inputStream = new FileInputStream(keyStorePath)) {
            KeyStore keyStore = KeyStore.getInstance(keyStoreType);
            keyStore.load(inputStream, keyStorePwd.toCharArray());
            return keyStore;
        }
    }

    /**
     * 加载DER格式的私钥证书
     * @param filename
     * @return
     * @throws Exception
     */
    public static PrivateKey getDERPrivateKey(String filename) throws Exception {
        byte[] keyBytes = Files.readAllBytes(Paths.get(filename));
        PKCS8EncodedKeySpec spec = new PKCS8EncodedKeySpec(keyBytes);
        KeyFactory kf = KeyFactory.getInstance("RSA");
        return kf.generatePrivate(spec);
    }

    /**
     * 根据证书库和受信证书库构建双向认证sslSocket工厂
     * 
     * @param p12TrustStore p12受信库
     * @param p12KeyStore p12证书库
     * @param ksPwd 证书库密码
     * @return
     * @throws Exception
     */
    public static SSLSocketFactory getSocketFactoryByPKCS12(String p12TrustStore, String tsPwd, String p12KeyStore,
        String ksPwd) throws Exception {
        final String P12_TYPE = "pkcs12";
        KeyStore trustStore = getKeyStore(p12TrustStore, tsPwd, P12_TYPE);
        TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
        tmf.init(trustStore);
        KeyStore keyStore = getKeyStore(p12KeyStore, ksPwd, P12_TYPE);
        // 创建管理JKS密钥库的密钥管理器 (SunX509)
        KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
        // 使用密钥内容源初始化此工厂。 提供者通常使用 KeyStore 来获取在安全套接字协商期间所使用的密钥内容
        kmf.init(keyStore, ksPwd.toCharArray());
        // 初始sslcontext
        SSLContext sslContext = SSLContext.getInstance("TLSv1.2");
        sslContext.init(kmf.getKeyManagers(), tmf.getTrustManagers(), new SecureRandom());
        return sslContext.getSocketFactory();
    }

    /**
     * 通过证书链构造SSLSocketFactory（单向认证）
     * @param keyPath
     * @param pwd
     * @param ca
     * @param crt
     * @return
     * @throws Exception
     */
    public static SSLSocketFactory getSocketFactoryByCrtList(String keyPath, String pwd, String ca, String crt) throws Exception {
        final KeyStore keyStore = KeyStore.getInstance("pkcs12");
        keyStore.load(null);
        PrivateKey privateKey = getDERPrivateKey(keyPath);
        keyStore.setKeyEntry("cert", privateKey, pwd.toCharArray(), new Certificate[]{getX509Certificate(crt), getX509Certificate(ca)});
        // 创建管理JKS密钥库的密钥管理器 (SunX509)
        KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
        // 使用密钥内容源初始化此工厂。 提供者通常使用 KeyStore 来获取在安全套接字协商期间所使用的密钥内容
        kmf.init(keyStore, null);
        // 初始sslcontext
        SSLContext sslContext = SSLContext.getInstance("TLSv1.2");
        // 单向认证，不验证证书以及ip。方便测试
        final TrustManager trustManager = new X509ExtendedTrustManager(){

            @Override
            public void checkClientTrusted(X509Certificate[] chain, String authType, Socket socket) throws CertificateException {

            }

            @Override
            public void checkServerTrusted(X509Certificate[] chain, String authType, Socket socket) throws CertificateException {

            }

            @Override
            public void checkClientTrusted(X509Certificate[] chain, String authType, SSLEngine engine) throws CertificateException {

            }

            @Override
            public void checkServerTrusted(X509Certificate[] chain, String authType, SSLEngine engine) throws CertificateException {

            }

            @Override
            public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException {

            }

            @Override
            public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException {

            }

            @Override
            public X509Certificate[] getAcceptedIssuers() {
                return new X509Certificate[0];
            }
        };
        sslContext.init(kmf.getKeyManagers(), new TrustManager[] {trustManager}, new SecureRandom());
        return sslContext.getSocketFactory();
    }

    /**
     * 需要添加受信的证书
     * 
     * @param certAlias 证书别名
     * @param cert 证书
     * @param trustStore 受信库路径
     * @param pwd 受信库密码
     * @return
     */
    public static boolean trustCert(String certAlias, Certificate cert, String trustStore, String pwd,
        String storeType) {
        boolean isOk = true;
        try {
            initTrustStoreIfNotExist(trustStore, pwd, storeType);
        } catch (Exception e) {
            log.error("create trustStore error", e);
            isOk = false;
        }
        if (isOk) {
            try (FileInputStream ips = new FileInputStream(trustStore)) {
                KeyStore ks = KeyStore.getInstance(storeType);
                ks.load(ips, pwd.toCharArray());
                ks.deleteEntry(certAlias);
                ks.setCertificateEntry(certAlias, cert);
                try (FileOutputStream ops = new FileOutputStream(trustStore)) {
                    ks.store(ops, pwd.toCharArray());
                }
            } catch (Exception e) {
                log.error("authorization certificate error", e);
                isOk = false;
            }
        }
        return isOk;
    }

    /**
     * 如果不存在创建文件
     * 
     * @param trustStore 文件路径
     * @return 是否新建
     */
    public static void initTrustStoreIfNotExist(String trustStore, String pwd, String storeType) throws Exception {
        File trustStoreFile = new File(trustStore);
        if (!trustStoreFile.getParentFile().exists()) {
            trustStoreFile.getParentFile().mkdirs();
        }
        if (!trustStoreFile.exists()) {
            trustStoreFile.createNewFile();
            KeyStore ks = KeyStore.getInstance(storeType);
            try (FileOutputStream ops = new FileOutputStream(trustStoreFile)) {
                ks.load(null);
                ks.store(ops, pwd.toCharArray());
            }
        }
    }

    /**
     * 使用TrustStore检测证书（链）是否受信
     * 
     * @param crt
     * @param certPath
     * @return
     */
    public static boolean authCertByTrustStore(String crt, String... certPath) {
        boolean res = true;
        try {
            // 需要校验的证书
            X509Certificate[] crts = new X509Certificate[certPath.length];
            for (int i = 0; i < certPath.length; i++) {
                crts[i] = getX509Certificate(certPath[i]);
            }
            KeyStore trustStore = KeyStore.getInstance("pkcs12");
            trustStore.load(null);
            trustStore.setCertificateEntry("ca", getX509Certificate(crt));
            TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
            tmf.init(trustStore);
            final TrustManager[] trustManagers = tmf.getTrustManagers();
            for (int i = 0; i < trustManagers.length; i++) {
                X509TrustManager manager = (X509TrustManager)trustManagers[i];
                // 检测不成功会抛出异常
//                manager.checkClientTrusted(new X509Certificate[] {certificate}, "RSA");
                manager.checkServerTrusted(crts, "RSA");
            }
        } catch (Exception e) {
            log.error("auth cert error ", e);
            res = false;
        }
        return res;
    }
}

private final static String CONNECT = "192.168.1.1:8088,xxxx";// 多个用逗号隔开
private static CountDownLatch cdl = new CountDownLatch(1);
ZooKeeper zk = new ZooKeeper(CONNECT, 5000, new Watch(){// 5000超时
    public void process(WatchedEvent watchedEvent) {
        // 如果获取到了连接
        if(watchedEvent.getState() == Event.keeperState.SyncConnected) {
            countDownLatch.countDown();
        }
    }
});

countDownLatch.watch();//等待连接
zookeeper.create("/path", "val".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
Stat stat = new Stat();// 节点状态
// watch为true代表监听此节点，节点内容发生变化会回调连接时注册的watch。watch是一致性的
// watch之后只会返回当前session最后一次修改此节点的内容，即多个setData,只会最后一次回调
byte[] data = zookeeper.getData("/path", true, stat);// 返回值

List<String> childrens = zookeeper.getChildren("/path", true);//[a,b,c]的形式返回子节点

ACL acl = new ACL(ZooDefs.Perms.ALL, new Id("digest", DigestAuthenticationProvicer.generateDigest("root:root")));
ACL acl2 = new ACL(ZooDefs.Perms.CREATE, new Id("ip", "192.168.1.1"));
List<ACL> acls = new ArrayList<>();
acls.add(acl);
acls.add(acl2);
// 创建持久节点
zookeeper.create("path", "val".getBytes(), acls, CreateMode.PERSISTENT);
// 添加digest方案的权限
zookeeper.addAuthInfo("digest", "root:root".getBytes());

private final static String CONNECT_HOST = "192.168.1.1:8000";
// zkClient不用监听
ZkClient zkClient = new ZkClient(CONNECT_HOST, 4000);
// 提供了递归创建父节点的功能,true代表级联新增，false代表不允许（父节点不存在就会报错）
zkClient.createPersistent("/zk/zkclient/zkclient1", true);
// 获取子节点 [node1,node2,node3]
List<String> list = zkClient.getChildren("/zkclient");
// 监听
// 节点内容修改
zkClient.subscribeDataChange("/nodeName", new IZkDataListener() {
    public void handleDataChange(String nodeName, Object newVal) throws Exception {
        // nodeName节点名称  newVal修改后的值
    }
    
    public void handleDataDeleted(String nodeName) throws Exception {
        // ...
    }
});

// nodeName中的子节点发升变化触发
zkClient.subscribeChildChanges("/nodeName", new IZkChildListener()) {
    public void handleChildChange(String nodeName, List<String> list) throws Exception {
        // nodeName节点名 list节点列表
    }
}
// 监听器
// subscriptStateChanges
// 权限
public void addAuthInfo(String scheme, final byte[] auth);
public void setAcl(final String path, final List<ACL> acl);

private final static String CONNECT_HOST="192.168.1.1:8000";
// 尝试三次连接，失败1000毫秒后重试，第二次重试时间2*1000毫秒，第三次3*1000
CuratorFramework curatorFramework = CuratorFrameworkFactory.
    newCliebt(CONNECT_HOST, 5000, 5000, new ExponentialBackoffRetry(1000, 3));
curatorFramework.start();// 启动连接

// 另一种写法
CuratorFramework curatorFramework = CuratorFrameworkFactory.builder().connectString(CONNECT_HOST).sessionTimeoutMs(5000).retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
curatorFramework.start();// 启动连接
// 创建节点,返回节点路径
String val = curatorFramework.create()
    .createingParentsIfNeeded()//级联创建父节点
    .withMode(CreateMode.PERSISTENT)//持久节点
    .forPath("/path/path1/path2", "val".getBytes());
// 删除节点,级联删除
curatorFramework.delete().deleteChildrenIfNeed().forPath("/path");

// 异步执行
final CountDownLatch countDownLatch = new CountDownLatch(1);
ExcutorService service = Executors.newFixedThreadPool(1);
curatorFramework.create().createingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
    .inBackground(new BackgroundCallback(){
        public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throw Exception {
            curatorFramework.getResultCode();// 结果
            curatorFramework.getType();
            countDownLatch.countDown();
        }
    },server).forPath("/path/path1/path2", "val".getBytes());    
// 事务, 同时成功才成功
Collection<CuratorTransactionResult> resultCollections = curatorFramework.inTransaction()
    .create().forPath("/path/path1/path2", "val".getBytes()).and()
    .setData().forPath("/path/path1/path2", "v1".getBytes()).and().commit();

for(CuratorTransactionResult res: resultCollections) {
    res.getForPath();// 节点路径
    res.getType();// 结果
}

// watch机制
// Pathcache 监听一个路径下子节点的创建、删除、数据更新
// NodeCache 节点的创建、删除、更新
// TreeCache Pathcache+NodeCache
NodeCache nodeCache = new NodeCache(curatorFramework, "/curator", false);
nodeCache.start(true);
nodeCache.getListenable().addListener(()->System.out.println("节点变化，变成了" + new String(nodeCache.getCurrentData().getData())));

pathCache.getListenable().addListener((framework, event)->{
    event.getType();//CHILD_ADD CHILD_REMOVED CHILD_UPDATE 子节点增加、删除、更新
});

class Solution {
    public int maxSubArray(int[] nums) {
        if (nums.length == 1) {
            return nums[0];
        }
        int[] dp = new int[nums.length];
        dp[0] = nums[0];
        for (int i = 1; i < nums.length; i++) {
            dp[i] = Math.max(dp[i - 1] + nums[i], nums[i]);
        }
        int max = dp[0];
        for (int i : dp) {
            max = Math.max(i, max);
        }
        return max;
    }
}

class Solution {
    public int climbStairs(int n) {
        if(n == 1) {
            return 1;
        }
        if(n == 2) {
            return 2;
        }
        int[] dp = new int[n];
        dp[0] = 1;
        dp[1] = 2;
        for(int i=2; i<n ;i++) {
            dp[i] = dp[i-1] + dp[i-2];
        }
        return dp[n - 1];
    }
}

class Solution {
    public int rob(int[] nums) {
        // 边界情况
        if (nums.length == 0) {
            return 0;
        } else if (nums.length == 1) {
            return nums[0];
        }
        int[] dp = new int[nums.length];
        dp[0] = nums[0];
        dp[1] = Math.max(nums[0], nums[1]);
        for (int i = 2; i < nums.length; i++) {
            dp[i] = Math.max(dp[i - 2] + nums[i], dp[i - 1]);
        }
        return dp[nums.length - 1];
    }
}

class Solution {
    public int calculateMinimumHP(int[][] dungeon) {
        int n = dungeon.length;
        int m = dungeon[0].length;
        int[][] dp = new int[n][m];
        dp[n - 1][m - 1] = Math.max((1 - dungeon[n - 1][m - 1]), 1);
        for (int i = m - 2; i >= 0; i--) {
            dp[n - 1][i] = Math.max(dp[n - 1][i + 1] - dungeon[n - 1][i], 1);
        }
        // 只有一行的情况下，直接返回
        if (dungeon.length == 1) {
            return dp[0][0];
        }
        for (int i = n - 2; i >= 0; i--) {
            dp[i][m - 1] = Math.max(dp[i + 1][m - 1] - dungeon[i][m - 1], 1);
        }
        for (int i = n - 2; i >= 0; i--) {
            for (int j = m - 2; j >= 0; j--) {
                dp[i][j] = Math.max(1, Math.min(dp[i + 1][j] - dungeon[i][j], dp[i][j + 1] - dungeon[i][j]));
            }
        }
        return dp[0][0];
    }
}

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Deque;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

/**
 * @author colin.cheng
 * @date 2021-10-30
 * @since 1.0.0
 */
public class SocketChannelProcessorThread extends Thread {

    private int bufferSize;
    private Selector selector;
    private ExecutorService workerPool;
    private Function<String, String> dealPackage;
    private Charset defaultCharset = StandardCharsets.UTF_8;
    private volatile boolean flag = true;

    public SocketChannelProcessorThread(Function<String, String> dealPackage, int bufferSize, int workerSize, Selector selector) {
        super("TcpServer-MessageProcessor-Thread");
        this.selector = selector;
        this.dealPackage = dealPackage;
        this.bufferSize = bufferSize;
        this.workerPool = Executors.newFixedThreadPool(workerSize, new ThreadFactory() {
            final AtomicInteger index = new AtomicInteger();
            @Override
            public Thread newThread(Runnable runnable) {
                return new Thread(runnable, "TcpServer-MessageProcessor-Worker" + index.getAndIncrement());
            }
        });
    }

    @Override
    public void run() {
        while (flag) {
            try {
                int readyChannels = this.selector.select(100);
                if(readyChannels == 0) {
                    continue;
                }
                Set<SelectionKey> selectedKeys = this.selector.selectedKeys();
                Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
                while(keyIterator.hasNext()) {
                    final SelectionKey selectionKey = keyIterator.next();
                    try {
                        dealSelectedKey(selectionKey);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    keyIterator.remove();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public void close() throws IOException {
        this.flag = false;
        this.selector.close();
        this.workerPool.shutdown();
    }

    private void dealSelectedKey(SelectionKey key) throws IOException {
        if(key.isConnectable()) {

        } else if (key.isReadable()) {
            final byte[] bytes = new byte[bufferSize];
            bytes[0] = 1;
            bytes[1] = 1;
            final ByteBuffer buffer = ByteBuffer.wrap(bytes);
            SocketChannel channel = (SocketChannel)key.channel();
            int len = channel.read(buffer);
            if(len > 0) {
                final String str = new String(buffer.array(), defaultCharset);
                workerPool.execute(()->{
                    final String res = dealPackage.apply(str);
                    if(res != null) {
                        ((Deque<String>)key.attachment()).addLast(res);
                    }
                });
            }
        } else if(key.isWritable()) {
            SocketChannel channel = (SocketChannel)key.channel();
            Deque<String> messList = (Deque<String>)key.attachment();
            final Iterator<String> iterator = messList.iterator();
            while (iterator.hasNext()) {
                String str = iterator.next();
                System.out.println("send >> " + str);
                channel.write(ByteBuffer.wrap(str.getBytes(defaultCharset)));
                iterator.remove();
            }
        }
    }
}

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Deque;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.function.Function;

/**
 * @author colin.cheng
 * @date 2021-10-30
 * @since 1.0.0
 */
public class TcpClient {

    public SocketChannel socketChannel;
    private Selector selector;
    private SelectionKey selectionKey;
    private SocketChannelProcessorThread socketChannelProcessorThread;

    public TcpClient(String ip, int port, Function<String, String> dealPackage, int bufferSize, int workerSize) throws IOException {
        this.socketChannel = SocketChannel.open();
        this.socketChannel.connect(new InetSocketAddress(ip, port));
        this.socketChannel.configureBlocking(false);
        this.selector = Selector.open();
        this.selectionKey = this.socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
        this.selectionKey.attach(new ConcurrentLinkedDeque<>());
        this.socketChannelProcessorThread = new SocketChannelProcessorThread(dealPackage, bufferSize, workerSize, selector);
        this.socketChannelProcessorThread.start();
    }

    public void write(String mess) throws IOException {
        Deque<String> list = (Deque<String>)this.selectionKey.attachment();
        list.add(mess);
    }

    public void close() throws IOException {
        this.socketChannelProcessorThread.close();
        this.socketChannel.close();
        this.selector.close();
    }
}

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.*;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.function.Function;

/**
 * @author colin.cheng
 * @date 2021-10-30
 * @since 1.0.0
 */
@Slf4j
public class TcpServer {

    private Selector selector;
    private BindPortThread bindThread;
    private SocketChannelProcessorThread messageProcessorThread;
    private ServerSocketChannel serverSocketChannel;

    public TcpServer(int port, Function<String, String> dealPackage, int bufferSize, int workerSize) throws IOException {
        this.selector = Selector.open();
        this.serverSocketChannel = ServerSocketChannel.open();
        this.serverSocketChannel.socket().bind(new InetSocketAddress(port));
        this.serverSocketChannel.configureBlocking(false);
        this.messageProcessorThread = new SocketChannelProcessorThread(dealPackage, bufferSize, workerSize, selector);
        this.messageProcessorThread.start();
        this.bindThread = new BindPortThread(selector, serverSocketChannel);
        this.bindThread.start();
    }

    public void close() throws IOException {
        this.bindThread.close();
        this.messageProcessorThread.close();
        this.serverSocketChannel.close();
        this.selector.close();
    }

    private class BindPortThread extends Thread {

        private Selector selector;
        private ServerSocketChannel serverSocketChannel;
        private volatile boolean flag = true;

        public BindPortThread(Selector selector, ServerSocketChannel serverSocketChannel) {
            super("TcpServer-BindPort-Thread");
            this.selector = selector;
            this.serverSocketChannel = serverSocketChannel;
        }

        @Override
        public void run() {
            while (flag) {
                try {
                    if(serverSocketChannel.isOpen()) {
                        final SocketChannel accept = serverSocketChannel.accept();
                        if(accept != null) {
                            accept.configureBlocking(false);
                            accept.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE).attach(new ConcurrentLinkedDeque<>());
                        }
                    }
                } catch (ClosedChannelException e) {
                    e.printStackTrace();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        public void close() throws IOException {
            this.serverSocketChannel.close();
            this.flag = false;
        }
    }
}

public static void main(String[] args) {
    try {
        TcpServer server = new TcpServer(8111, mess->{
            System.out.println("server收到 [" + mess + "]");
            return "来自server的回复=》" + mess;
        }, 1024, 2);

        TcpClient client = new TcpClient("127.0.0.1", 8111, mess->{
            System.out.println("client收到 [" + mess + "]");
            return null;
        }, 1024, 2);

        client.write("test1");
        client.write("test2");
    } catch (Exception e) {
        e.printStackTrace();
    }
}