ini复制代码# AutoConfiguration
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
com.github.jojotech.spring.cloud.webflux.auto.WebClientAutoConfiguration

less复制代码@Import(WebClientConfiguration.class)
@Configuration(proxyBeanMethods = false)
public class WebClientAutoConfiguration {
}

less复制代码@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties(WebClientConfigurationProperties.class)
public class WebClientConfiguration {
    @Bean
    public WebClientNamedContextFactory getWebClientNamedContextFactory() {
        return new WebClientNamedContextFactory();
    }
}

typescript复制代码@Data
@NoArgsConstructor
@ConfigurationProperties(prefix = "webclient")
public class WebClientConfigurationProperties {
    private Map<String, WebClientProperties> configs;
    @Data
    @NoArgsConstructor
    public static class WebClientProperties {
        private static AntPathMatcher antPathMatcher = new AntPathMatcher();
        private Cache<String, Boolean> retryablePathsMatchResult = Caffeine.newBuilder().build();
        /**
         * 服务地址，不填写则为 http://serviceName
         */
        private String baseUrl;
        /**
         * 微服务名称，不填写就是 configs 这个 map 的 key
         */
        private String serviceName;
        /**
         * 可以重试的路径，默认只对 GET 方法重试，通过这个配置增加针对某些非 GET 方法的路径的重试
         */
        private List<String> retryablePaths;
        /**
         * 连接超时
         */
        private Duration connectTimeout = Duration.ofMillis(500);
        /**
         * 响应超时
         */
        private Duration responseTimeout = Duration.ofSeconds(8);

        /**
         * 是否匹配
         * @param path
         * @return
         */
        public boolean retryablePathsMatch(String path) {
            if (CollectionUtils.isEmpty(retryablePaths)) {
                return false;
            }
            return retryablePathsMatchResult.get(path, k -> {
                return retryablePaths.stream().anyMatch(pattern -> antPathMatcher.match(pattern, path));
            });
        }
    }
}

xml复制代码<!--粘合 project-reactor 与 resilience4j，这个在异步场景经常会用到-->
<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-reactor</artifactId>
</dependency>

ini复制代码//由于还是在前面弄好的 spring-cloud 环境下，所以还是可以这样获取配置对应的 retry
Retry retry;
try {
    retry = retryRegistry.retry(name, name);
} catch (ConfigurationNotFoundException e) {
    retry = retryRegistry.retry(name);
}

Retry finalRetry = retry;
WebClient.builder().filter((clientRequest, exchangeFunction) -> {
    return exchangeFunction.exchange(clientRequest)
        //核心就是加入 RetryOperator
        .transform(RetryOperator.of(finalRetry));
})

swift复制代码@Override
public Publisher<T> apply(Publisher<T> publisher) {
    //对于 mono 的处理
    if (publisher instanceof Mono) {
        Context<T> context = new Context<>(retry.asyncContext());
        Mono<T> upstream = (Mono<T>) publisher;
        return upstream.doOnNext(context::handleResult)
            .retryWhen(reactor.util.retry.Retry.withThrowable(errors -> errors.flatMap(context::handleErrors)))
            .doOnSuccess(t -> context.onComplete());
    } else if (publisher instanceof Flux) {
        //对于 flux 的处理
        Context<T> context = new Context<>(retry.asyncContext());
        Flux<T> upstream = (Flux<T>) publisher;
        return upstream.doOnNext(context::handleResult)
            .retryWhen(reactor.util.retry.Retry.withThrowable(errors -> errors.flatMap(context::handleErrors)))
            .doOnComplete(context::onComplete);
    } else {
        //不可能是 mono 或者 flux 以外的其他的
        throw new IllegalPublisherException(publisher);
    }
}

ini复制代码//由于还是在前面弄好的 spring-cloud 环境下，所以还是可以这样获取配置对应的 circuitBreaker
CircuitBreaker circuitBreaker;
try {
    circuitBreaker = circuitBreakerRegistry.circuitBreaker(instancId, finalServiceName);
} catch (ConfigurationNotFoundException e) {
    circuitBreaker = circuitBreakerRegistry.circuitBreaker(instancId);
}

CircuitBreaker finalCircuitBreaker = circuitBreaker;
WebClient.builder().filter((clientRequest, exchangeFunction) -> {
    return exchangeFunction.exchange(clientRequest)
        //核心就是加入 CircuitBreakerOperator
        .transform(CircuitBreakerOperator.of(finalCircuitBreaker));
})

java复制代码class CircuitBreakerSubscriber<T> extends AbstractSubscriber<T> {

    private final CircuitBreaker circuitBreaker;

    private final long start;
    private final boolean singleProducer;

    private final AtomicBoolean successSignaled = new AtomicBoolean(false);
    private final AtomicBoolean eventWasEmitted = new AtomicBoolean(false);

    protected CircuitBreakerSubscriber(CircuitBreaker circuitBreaker,
        CoreSubscriber<? super T> downstreamSubscriber,
        boolean singleProducer) {
        super(downstreamSubscriber);
        this.circuitBreaker = requireNonNull(circuitBreaker);
        this.singleProducer = singleProducer;
        this.start = circuitBreaker.getCurrentTimestamp();
    }

    @Override
    protected void hookOnNext(T value) {
        if (!isDisposed()) {
             //正常完成时，断路器也标记成功，因为可能会触发多次（因为 onComplete 也会记录），所以需要 successSignaled 标记只记录一次
            if (singleProducer && successSignaled.compareAndSet(false, true)) {
                circuitBreaker.onResult(circuitBreaker.getCurrentTimestamp() - start, circuitBreaker.getTimestampUnit(), value);
            }
            //标记事件已经发出，就是已经执行完 WebClient 的请求，后面判断取消的时候会用到
            eventWasEmitted.set(true);

            downstreamSubscriber.onNext(value);
        }
    }

    @Override
    protected void hookOnComplete() {
        //正常完成时，断路器也标记成功，因为可能会触发多次（因为 onNext 也会记录），所以需要 successSignaled 标记只记录一次
        if (successSignaled.compareAndSet(false, true)) {
            circuitBreaker.onSuccess(circuitBreaker.getCurrentTimestamp() - start, circuitBreaker.getTimestampUnit());
        }

        downstreamSubscriber.onComplete();
    }

    @Override
    public void hookOnCancel() {
        if (!successSignaled.get()) {
            //如果事件已经发出，那么也记录成功
            if (eventWasEmitted.get()) {
                circuitBreaker.onSuccess(circuitBreaker.getCurrentTimestamp() - start, circuitBreaker.getTimestampUnit());
            } else {
                //否则取消
                circuitBreaker.releasePermission();
            }
        }
    }

    @Override
    protected void hookOnError(Throwable e) {
        //记录失败
        circuitBreaker.onError(circuitBreaker.getCurrentTimestamp() - start, circuitBreaker.getTimestampUnit(), e);
        downstreamSubscriber.onError(e);
    }
}

arduino复制代码 public final class String
    implements java.io.Serializable, Comparable<String>, CharSequence {
    /** The value is used for character storage. */
    private final char value[];

    /** Cache the hash code for the string */
    private int hash; // Default to 0

    public String(String original) {
        this.value = original.value;
        this.hash = original.hash;
    }

    public String concat(String str) {
        int otherLen = str.length();
        if (otherLen == 0) {
            return this;
        }
        int len = value.length;
        char buf[] = Arrays.copyOf(value, len + otherLen);
        str.getChars(buf, len);
        return new String(buf, true);
    }

    public boolean equals(Object anObject) {
        if (this == anObject) {
            return true;
        }
        if (anObject instanceof String) {
            String anotherString = (String) anObject;
            int n = value.length;
            if (n == anotherString.value.length) {
                char v1[] = value;
                char v2[] = anotherString.value;
                int i = 0;
                while (n-- != 0) {
                    if (v1[i] != v2[i])
                            return false;
                    i++;
                }
                return true;
            }
        }
        return false;
    }
    ...

ini复制代码public static void main(String[] args) {
        char[] value = {'a','b','c','d'};
        System.out.println(value); 
        for (int i = 0; i < value.length; i++) {
            if(i == 1){
                value[i] = 'B';
            }
        }
        System.out.println(value); //aBcd
    }

arduino复制代码public static void main(String[] args) throws Exception {
        //创建字符串"Hello World"， 并赋给引用s  
        String s = "Hello World";   
        System.out.println("s = " + s); //Hello World  

        //获取String类中的value字段  
        Field valueFieldOfString = String.class.getDeclaredField("value");  
        //改变value属性的访问权限  
        valueFieldOfString.setAccessible(true);  

        //获取s对象上的value属性的值  
        char[] value = (char[]) valueFieldOfString.get(s);  
        //改变value所引用的数组中的第5个字符  
        value[5] = '_';  

        System.out.println("s = " + s);  //Hello_World  
    }

arduino复制代码/** Cache the hash code for the string */
    private int hash; // Default to 0

ini复制代码输入： nums = [1,3,2,2,5,2,3,7]
输出： 5
解释： 最长的和谐子序列是 [3,2,2,2,3]

ini复制代码输入： nums = [1,2,3,4]
输出： 2

ini复制代码输入： nums = [1,1,1,1]
输出： 0

java复制代码class Solution {
    public int findLHS(int[] nums) {
        // 排序
        Arrays.sort(nums);
        int max = 0;
        for(int left = 0, right = 1; right < nums.length; ++right){
            // 判断如果左边元素与右边元素相差超过1，则指针右移
            while(left < right && nums[right] - nums[left] > 1){
                ++left;
            }
            // 判断是否为和谐数组
            if(nums[right] - nums[left] == 1){
                // 更新结果
                max = Math.max(max, right - left + 1);
            }
        }

        return max;
    }
}

sql复制代码CREATE TABLE `member` (
  `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
  `tel` bigint(20) DEFAULT NULL COMMENT '手机号码',
  `password` varchar(555) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '登录密码',
  `wx_token` varchar(125) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '微信TOKEN',
  `im_token` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '用户融云token',
  `open_id` varchar(125) COLLATE utf8mb4_unicode_ci DEFAULT NULL,
  `status` enum('1','-1') COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT '1' COMMENT '账号状态',
  `created_at` timestamp NULL DEFAULT NULL,
  `updated_at` timestamp NULL DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `member_tel_unique` (`tel`),
  UNIQUE KEY `member_wx_token_unique` (`wx_token`)
) ENGINE=InnoDB AUTO_INCREMENT=95 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

sql复制代码CREATE TABLE `member_address` (
  `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
  `member_id` int(11) NOT NULL COMMENT '用户编号',
  `nick_name` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '收货人姓名',
  `tel` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '手机号码',
  `prov` int(11) DEFAULT NULL COMMENT '省',
  `city` int(11) NOT NULL COMMENT '市',
  `area` int(11) DEFAULT NULL COMMENT '区',
  `address` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT '' COMMENT '街道地址',
  `number` int(11) NOT NULL COMMENT '邮政编码',
  `default` enum('0','1') COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT '0' COMMENT '默认收货地址 1=>默认',
  `deleted_at` timestamp NULL DEFAULT NULL,
  `created_at` timestamp NULL DEFAULT NULL,
  `updated_at` timestamp NULL DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=55 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

sql复制代码CREATE TABLE `member_card` (
  `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
  `member_id` int(11) NOT NULL COMMENT '用户编码',
  `card_name` varchar(25) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '持卡人姓名',
  `card_number` varchar(25) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '银行卡号',
  `created_at` timestamp NULL DEFAULT NULL,
  `updated_at` timestamp NULL DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `member_card_card_number_unique` (`card_number`)
) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

sql复制代码CREATE TABLE `member_cart` (
  `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
  `member_id` int(11) NOT NULL COMMENT '用户编码',
  `created_at` timestamp NULL DEFAULT NULL,
  `updated_at` timestamp NULL DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `member_cart_member_id_unique` (`member_id`),
  KEY `member_cart_member_id_index` (`member_id`)
) ENGINE=InnoDB AUTO_INCREMENT=28 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

sql复制代码CREATE TABLE `member_cart_item` (
  `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
  `cart_id` int(11) NOT NULL COMMENT '购物车编码',
  `product_desc` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '商品sku信息',
  `product_img` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '商品快照',
  `product_name` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '商品名称',
  `price` decimal(8,2) NOT NULL DEFAULT '0.00' COMMENT '价格',
  `product_id` int(11) NOT NULL COMMENT '商品编码',
  `supplier_id` int(11) NOT NULL COMMENT '店铺编码',
  `sku_id` int(11) NOT NULL COMMENT '商品sku编码',
  `number` int(11) NOT NULL DEFAULT '1' COMMENT '商品数量',
  `created_at` timestamp NULL DEFAULT NULL,
  `updated_at` timestamp NULL DEFAULT NULL,
  PRIMARY KEY (`id`),
  KEY `member_cart_item_cart_id_product_id_supplier_id_index` (`cart_id`,`product_id`,`supplier_id`)
) ENGINE=InnoDB AUTO_INCREMENT=24 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

sql复制代码CREATE TABLE `member_query_history` (
  `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
  `member_id` int(11) NOT NULL COMMENT '用户编码',
  `keyword` varchar(125) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '关键字',
  `created_at` timestamp NULL DEFAULT NULL,
  `updated_at` timestamp NULL DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

java复制代码
ResponseVO queryData(){
    
    // 先从缓存中查数据
    ResonseVO vo=queryFromCache();

    if(Objects.isNull(vo){
        
        // 缓存中没有数据，去数据库中查询数据
        DataBaseEntity entity=queryFromDataBase();
        vo=dataConvert(entity);
        
        // 设置缓存
        setCache(vo);    
    }
    return vo;
}

scss复制代码ResponseVO queryData() {

    int retry = 0;
    boolean locked;
    try {
        do {

            // 从缓存中查询数据
            ResponseVO vo = queryFromCache();
            if (Objects.isNull(vo) {
                return vo;
            }

            // 尝试获取锁,setnx命令
            locked = tryLock();
            if (locked) {

                // 获取到锁以后,再次从缓存查数据.
                // 因为当前线程获取到的锁是被另外一个线程释放掉的,而另外一个线程此时已经设置了缓存
                vo = queryFromCache();
                if (Objects.isNull(vo) {
                    return vo;
                }
                // 如果缓存中还是没有数据,只能查库了
                DataBaseEntity entity = queryFromDataBase();
                vo = dataConvert(entity);

                // 设置缓存
                setCache(vo);
                return vo;
            }
            retry++;
        } while (!locked && retry <= 10); // 如果某线程一直获取不到锁,就进入死循环了,设置一个循环次数

    } finally {
        if (locked) {
            // 释放锁
            releaseLock();
        }
    }
    return null;
}

java复制代码ResponseVO queryData() {

    int retry = 0;
    boolean locked;
    try {

        do{

            // 从缓存中查询数据
            ResponseVO vo = queryFromCache();
            if (Objects.isNull(vo) {
                return vo;
            }

            // 尝试获取锁,setnx命令,没获取到锁,就睡两秒钟
            locked = tryLock(2);
            if (locked) {

                // 获取到锁以后,再次从缓存查数据.
                // 因为当前线程获取到的锁是被另外一个线程释放掉的,而另外一个线程此时已经设置了缓存
                vo = queryFromCache();
                if (Objects.isNull(vo) {
                    return vo;
                }
                // 如果缓存中还是没有数据,只能查库了
                DataBaseEntity entity = queryFromDataBase();
                vo = dataConvert(entity);

                // 设置缓存
                setCache(vo);
                return vo;
            }
            retry++;
        }while (!locked && retry<=10);
    } finally {
        if (locked) {
            // 释放锁
            releaseLock();
        }
    }
    return null;
}

java复制代码
ResponseVO queryData(){
    
    // 先从缓存中查数据
    ResonseVO vo=queryFromCache();

    if(Objects.isNull(vo){
        
        // 缓存中没有数据，去数据库中查询数据
        DataBaseEntity entity=queryFromDataBase();
        if(Objects.isNull(entity){
            // 数据库中不存在，就设置一个空值
             vo=new ResonseVO()；
        }else{
            vo=dataConvert(entity);
        }
        // 设置缓存
        setCache(vo);    
    }
    return vo;
}

scss复制代码set(key，value,100+ThreadLocalRandom.current().nextInt(1,500));

yml复制代码spring:
	application:
		name: spring-cloud-sentinel
	cloud:
		sentinel:
			transport:	
				dashboard: 127.0.0.1:8080

java复制代码@RestController
@RequestMapping("/sentinel")
public class SentinelController {

    @GetMapping("limit")
    public String limit(){
        return "Sentinel DashBoard Flow Limit";
    }
}

scss复制代码Blocked By Sentinel(flow limiting)

scss复制代码Blocked By Sentinel(flow limiting)

java复制代码@Service
public class CustomUrlBlockHandler implements UrlBlockHandler{
    	@Override
    	public void blocked(HttpServletRequest httpServletRequest,HttpServletResponse httpServletResponse,BlockException e) throws IOException{
            httpServletResponse.setHeader("Content-Type","application/json;charset=UTF-8");
            String msg="请求数超过最大数，请稍后再试！";
            httpServletResponse.getWirter().write(msg);
        }
}

yml复制代码spring:
	cloud:
		sentinel:
			serlvet:
				block-page: 降级页面url地址

ini复制代码$client = new Client('UserSerivce');
$userInfo = $client->getUserInfo(1);

var_dump($userInfo);

php复制代码class Client{
    protected $serviceName;
    		
    public function __construct($serviceName){
    	$this->serviceName = $serviceName;
    }
    public function __call($name, $arguments){
    	$rpcClient = new \swoole_client(SWOOLE_SOCK_TCP);
    	$rpcClient->connect('127.0.0.1',9503,0.5);
    	// 我们将要发送的数据是事先约定好的，跟写对外开放的Api一样
    	$rpcClient->send(json_encode([
    	    'service'=>$this->serviceName,
    	    'action'=>$name,
    	    'params'=>$arguments[0]
    	]));
    	$rpcClient->close();
    }
}

bash复制代码$server = new swoole_server("127.0.0.1", 9503);
$server->on('connect', function ($server, $fd){
    echo "connection open: {$fd}\n";
});
$server->on('receive', function ($server, $fd, $reactor_id, $data) {
    // $data 则就是客户端发送过来的数据，我们可以这样做来做到去调用类,当然你必须遵守PSR-4 Autoloader
    $request   = json_decode ($data, true);
    $className = $request['service'];
    $app       = new $className;
	$response  = $app->{$request['action']}($request['params']);
    
    $server->send($fd, "Swoole: {$data}");
    $server->close($fd);
});
$server->on('close', function ($server, $fd) {
    echo "connection close: {$fd}\n";
});
$server->start();

java复制代码public class Server {
    public static void main(String[] args) {
        // 服务端占用端口
        int port = 8000;
        // 创建 serversocker
        ServerSocket serverSocket = null;
        try {
            serverSocket = new ServerSocket(port);
        } catch (IOException e) {
            e.printStackTrace();
        }
        if (serverSocket != null) {
            while (true) {
                InputStream is = null;
                OutputStream os = null;
                Socket client = null;
                try {
                    // accept()方法会阻塞，直到有client连接后才会执行后面的代码
                    client = serverSocket.accept();
                    is = client.getInputStream();
                    os = client.getOutputStream();

                    // 3.server收到消息在控制台的打印，并回复"Hi client,I am Server."
                    byte[] buffer = new byte[5];
                    int len = 0;
                    // 使用ByteArrayOutputStream，避免缓冲区过小导致中文乱码
                    ByteArrayOutputStream baos = new ByteArrayOutputStream();
                    while ((len = is.read(buffer)) != -1) {
                        baos.write(buffer,0,len);
                    }
                    System.out.println(baos.toString());
                    // 服务端回复客户端消息
                    os.write("Hi client,I am Server.".getBytes());
                    os.flush(); // 刷新缓存，避免消息没有发送出去
                    client.shutdownOutput();
                } catch (IOException e) {
                    e.printStackTrace();
                } finally {
                    // 程序异常或者执行完成，关闭流，防止占用资源
                    if (client != null) {
                        try {
                            client.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                    if (is != null) {
                        try {
                            is.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                    if (os != null) {
                        try {
                            os.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }

            }
        }


    }
}


public class Client {
    public static void main(String[] args) {
        int port = 8000;
        Socket client = null;
        InputStream is = null;
        OutputStream os = null;
        try {
            // 1.client连接server
            client = new Socket("localhost", port);
            is = client.getInputStream();
            os = client.getOutputStream();

            // 2.client发送"Hi Server,I am client."
            os.write("Hi Server,I am client.".getBytes());
            os.flush();
            // 调用shutdownOutput()方法表示客户端传输完了数据，否则服务端的
            // read()方法会一直阻塞
            // (你可能会问我这不是写了 read()!=-1, -1表示的文本文件的结尾字符串，而对于字节流数据，
            // 是没有 -1 标识的，这就会使服务端无法判断客户端是否发送完成，导致read()方法一直阻塞)
            client.shutdownOutput();

            // 4.client收到消息在控制台打印。
            int len = 0;
            byte[] buffer = new byte[5];
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            while ((len = is.read(buffer)) != -1) {
                baos.write(buffer,0,len);
            }
            System.out.println(baos.toString());
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            // 程序异常或者执行完成，关闭流，防止占用资源
            try {
                if (is != null) {
                    is.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            try {
                if (os != null) {
                    os.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            try {
                if (client != null) {
                    client.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }


    }
}

java复制代码public class NIOClient {

    public static void main(String[] args) {
        SocketAddress socketAddress = new InetSocketAddress(8000);
        SocketChannel socketChannel = null;
        try {
            socketChannel = SocketChannel.open(socketAddress);
            socketChannel.configureBlocking(false);
            if (socketChannel.finishConnect()) {

                ByteBuffer buffer = ByteBuffer.allocate(1024);
                // 客户端发送数据 "Hi Server,I am client."
                buffer.clear();
                buffer.put("Hi Server,I am client.".getBytes());
                buffer.flip();
                while (buffer.hasRemaining()) {
                    socketChannel.write(buffer);
                }
                // 客户端接收服务端数据打印在控制台

                buffer.clear();
                int len = socketChannel.read(buffer);
                while (len > 0) {
                    buffer.flip();
                    while (buffer.hasRemaining()) {
                        System.out.print((char) buffer.get());
                    }
                    System.out.println();
                    buffer.clear();
                    len = socketChannel.read(buffer);
                }
                if (len == -1) {
                    socketChannel.close();
                }
            }


        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (socketChannel != null) {
                    socketChannel.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}


public class NIOServer {


    public static void main(String[] args) {
        ServerSocketChannel serverSocketChannel = null;
        Selector selector = null;
        try {
            // 初始化一个 serverSocketChannel
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.bind(new InetSocketAddress(8000));

            // 设置serverSocketChannel为非阻塞模式
            // 即 select()会立即得到返回
            serverSocketChannel.configureBlocking(false);

            // 初始化一个 selector
            selector = Selector.open();

            // 将 serverSocketChannel 与 selector绑定
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

            while (true) {
                // 通过操作系统监听变化的socket个数
                // 在windows平台通过selector监听(轮询所有的socket进行判断，效率低)
                // 在Linux2.6之后通过epool监听(事件驱动方式，效率高)
                int count = selector.select(3000);
                if (count > 0) {
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        SelectionKey key = iterator.next();

                        if (key.isAcceptable()) {
                            handleAccept(key);
                        }
                        if (key.isReadable()) {
                            handleRead(key);
                        }
                        if (key.isWritable() && key.isValid()) {
                            handleWrite(key);
                        }
                        if (key.isConnectable()) {
                            System.out.println("isConnectable = true");
                        }
                        iterator.remove();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (serverSocketChannel != null) {
                    serverSocketChannel.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            try {
                if (selector != null) {
                    selector.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private static void handleWrite(SelectionKey key) {
        // 获取 client 的 socket
        SocketChannel clientChannel = (SocketChannel) key.channel();
        // 获取缓冲区
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        buffer.clear();
        buffer.put("Hi client,I am Server.".getBytes());
        buffer.flip();
        try {
            while (buffer.hasRemaining()) {
                clientChannel.write(buffer);
            }
            buffer.compact();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static void handleRead(SelectionKey key) {
        // 获取 readable 的客户端 socketChannel
        SocketChannel clientChannel = (SocketChannel) key.channel();
        // 读取客户端发送的消息信息,我们已经在 acceptable 中设置了缓冲区
        // 所以直接冲缓冲区读取信息
        ByteBuffer buffer = (ByteBuffer) key.attachment();

        // 获取 client 发送的消息
        try {
            int len = clientChannel.read(buffer);
            while (len > 0) {
                // 设置 limit 位置
                buffer.flip();
                // 开始读取数据
                while (buffer.hasRemaining()) {
                    byte b = buffer.get();
                    System.out.print((char) b);
                }
                System.out.println();
                // 清除 position 位置
                buffer.clear();
                // 从新读取 len
                len = clientChannel.read(buffer);
            }
            if (len == -1) {
                clientChannel.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }



    }

    private static void handleAccept(SelectionKey key) {
        // 获得 serverSocketChannel
        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
        try {
            // 获得 socketChannel,就是client的socket
            SocketChannel clientChannel = serverSocketChannel.accept();
            if (clientChannel == null) return;
            // 设置 socketChannel 为无阻塞模式
            clientChannel.configureBlocking(false);
            // 将其注册到 selector 中，设置监听其是否可读，并分配缓冲区
            clientChannel.register(key.selector(), SelectionKey.OP_READ, ByteBuffer.allocateDirect(512));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

java复制代码package com.github.jiangxch.jdk.demo.netio.aio;


import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.charset.Charset;

public class AioServer {

    public AsynchronousServerSocketChannel serverSocketChannel;

    public void listen() throws Exception {
        //打开一个服务端通道
        serverSocketChannel = AsynchronousServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress(9988));//监听9988端口
        //监听
        serverSocketChannel.accept(this, new CompletionHandler<AsynchronousSocketChannel, AioServer>() {
            @Override
            public void completed(AsynchronousSocketChannel client, AioServer attachment) {
                try {
                    if (client.isOpen()) {
                        System.out.println("接收到新的客户端的连接，地址：" + client.getRemoteAddress());
                        final ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                        //读取客户端发送的数据
                        client.read(byteBuffer, client, new CompletionHandler<Integer, AsynchronousSocketChannel>() {
                            @Override
                            public void completed(Integer result, AsynchronousSocketChannel attachment) {
                                try {
                                    //读取请求，处理客户端发送的数据
                                    byteBuffer.flip();
                                    String content = Charset.defaultCharset().newDecoder().decode(byteBuffer).toString();
                                    System.out.println("服务端接受到客户端发来的数据：" + content);
                                    //向客户端发送数据
                                    ByteBuffer writeBuffer = ByteBuffer.allocate(1024);
                                    writeBuffer.put("Server send".getBytes());
                                    writeBuffer.flip();
                                    attachment.write(writeBuffer).get();
                                } catch (Exception e) {
                                    e.printStackTrace();
                                }
                            }

                            @Override
                            public void failed(Throwable exc, AsynchronousSocketChannel attachment) {
                                try {
                                    exc.printStackTrace();
                                    attachment.close();
                                } catch (IOException e) {
                                    e.printStackTrace();
                                }
                            }
                        });
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    //当有新的客户端接入的时候，直接调用accept的方法，递归执行下去，这样可以保证多个客户端都可以阻塞
                    attachment.serverSocketChannel.accept(attachment, this);
                }
            }

            @Override
            public void failed(Throwable exc, AioServer attachment) {
                exc.printStackTrace();
            }
        });
    }


    public static void main(String[] args) throws Exception {
        new AioServer().listen();
        Thread.sleep(Integer.MAX_VALUE);
    }

}

java复制代码package com.github.jiangxch.jdk.demo.netio.aio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutionException;


public class AioClient {
    public static void main(String[] args) throws IOException, InterruptedException {
        //打开一个客户端通道
        AsynchronousSocketChannel channel = AsynchronousSocketChannel.open();
        //与服务端建立连接
        channel.connect(new InetSocketAddress("127.0.0.1", 9988));
        //睡眠一秒，等待与服务端的连接
        Thread.sleep(1000);

        try {
            //向服务端发送数据
            channel.write(ByteBuffer.wrap("Hello,我是客户端".getBytes())).get();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        try {
            //从服务端读取返回的数据
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            channel.read(byteBuffer).get();//将通道中的数据写入缓冲Buffer
            byteBuffer.flip();
            String result = Charset.defaultCharset().newDecoder().decode(byteBuffer).toString();
            System.out.println("客户端收到服务端返回的内容：" + result);//服务端返回的数据
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

}

c复制代码#include <sys/select.h>
int select(int maxfdpl,fd_set *readset,fd_set *writeset,fd_set *excepset,const struct timeval *timeout);

c复制代码struct {
    long tv_sec;// 等待秒
    long tv_usec; // microseconds
}

c复制代码typedef struct fd_set {

__int32_t fds_bits[__DARWIN_howmany(__DARWIN_FD_SETSIZE, __DARWIN_NFDBITS)];

} fd_set;

c复制代码while(1) {
    select(fdlist);// 监听文件描述符
    for(fd : fdlist) {
        if(fd.can_read()) {
               
        }
        if(fd.can_write()) {
        
        }
    }

}

c复制代码int epoll_create(int size);

csharp复制代码int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

arduino复制代码int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);

c复制代码struct epoll_event {
    __unit32_t events;// 表示监听什么操作，读？写？
    epoll_data_t data;// 用户数据变量

}

bash复制代码GET /book/_search
​
{
    "aggs": {
        "avg_bucket": {
            "avg": {
                "field": "price"
            }
        }
    }
}

json复制代码{
    "took": 62,
    "timed_out": false,
    "_shards": {
        "total": 3,
        "successful": 3,
        "skipped": 0,
        "failed": 0
    },
    "hits": {
        ...
    },
    "aggregations": {
        "avg_bucket": {
            "value": 65.64846185537485
        }
    }
}

bash复制代码GET /book/_search
{
    "size": 0,
    "aggs": {
        "avg_bucket": {
            "avg": {
                "field": "price"
            }
        }
    }
}

json复制代码{
    "took": 43,
    "timed_out": false,
    "_shards": {
        "total": 3,
        "successful": 3,
        "skipped": 0,
        "failed": 0
    },
    "hits": {
        "total": {
            "value": 13,
            "relation": "eq"
        },
        "max_score": null,
        "hits": []
    },
    "aggregations": {
        "avg_bucket": {
            "value": 65.64846185537485
        }
    }
}

json复制代码GET /book/_search
{
    "size": 0,
    "aggs": {
        "group": {
            "terms": {
                "field": "categoryId"
            },
            "aggs": {
                "price_top_1": {
                    "top_hits": {
                        "size": 1,
                        "_source": {
                            "includes": [ // 注：为了节省返回数据，这里我们定义只返回这些字段
                                "id",
                                "bookName",
                                "categoryId",
                                "categoryName",
                                "price"
                            ]
                        },
                        "sort": [
                            {
                                "price": {
                                    "order": "desc"
                                }
                            }
                        ]
                    }
                }
            }
        }
    }
}

ruby复制代码{
    "took":23,
    "timed_out":false,
    "_shards":{
        "total":3,
        "successful":3,
        "skipped":0,
        "failed":0
    },
    "hits":{
        "total":{
            "value":13,   // 文档总数
            "relation":"eq"
        },
        "max_score":null,
        "hits":[]
    },
    "aggregations":{
        "group":{
            "buckets":[ // 桶数组，下面即为有多少分组
                {
                    "key":1,  // 这里就是分组的值，即为 categoryId
                    "doc_count":6, // 这是表示这里命中了6个文档
                    "price_top_1":{ // 自定义的top1聚合结果
                        "hits":{
                            "total":{
                                "value":6, // 文档数
                                "relation":"eq"
                            },
                            "hits":[ // 这里就是拿到的 类目1 下的价格最高的图书
                                {
                                    "_index":"book",
                                    "_type":"_doc",
                                    "_id":"3",
                                    "_score":null,
                                    "_source":{
                                        "price":88,
                                        "id":3,
                                        "bookName":"Java编程思想",
                                        "categoryName":"教科书",
                                        "categoryId":1
                                    }
                                }
                            ]
                        }
                    }
                },
                {
                    "key":3,
                    "doc_count":4,
                    "price_top_1":{
                        "hits":{
                            "total":{
                                "value":4,
                                "relation":"eq"
                            },
                            "max_score":null,
                            "hits":[
                                {
                                    "_index":"book",
                                    "_type":"_doc",
                                    "_id":"12",
                                    "_source":{
                                        "price":99.68,
                                        "id":12,
                                        "bookName":"周恩来传",
                                        "categoryName":"人物传记",
                                        "categoryId":3
                                    }
                                }
                            ]
                        }
                    }
                },
                {
                    "key":2,
                    "doc_count":3,
                    "price_top_1":{
                        "hits":{
                            "total":{
                                "value":3,
                                "relation":"eq"
                            },
                            "max_score":null,
                            "hits":[
                                {
                                    "_index":"book",
                                    "_type":"_doc",
                                    "_id":"8",
                                    "_source":{
                                        "price":40,
                                        "id":8,
                                        "bookName":"神雕侠侣",
                                        "categoryName":"小说",
                                        "categoryId":2
                                    }
                                }
                            ]
                        }
                    }
                }
            ]
        }
    }
}

java复制代码    private SearchResponse searchGroupData() {
        // 构建搜索请求
        SearchRequest searchRequest = new SearchRequest(EsConstant.BOOK_INDEX_NAME);
​
        AggregationBuilder groupAggregation = AggregationBuilders
                // 这里是定义的最外层聚合的名字
                .terms("group")
                // 这是返回的分组的数量，默认是10
                // 大家可以根据自己的需要改返回的总数，最大不能超过设置的 max_result_window
                .size(10)
                // 要分组的字段
                .field("categoryId");
​
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        // 不需要返回hits中的数据，设置 size 为 0
        searchSourceBuilder.size(0);
​
        // TOP hits 取图书的 top1
        // 定义返回的字段
        String[] includes = {"id", "bookName", "categoryId", "categoryName", "price"};
        String[] excludes = {};
        AggregationBuilder topHits = AggregationBuilders
                // 自定义的topHits名称
                .topHits("price_top_1")
                // 按照价格倒序
                .sort(SortBuilders.fieldSort("price").order(SortOrder.DESC))
                .fetchSource(includes, excludes)
                // 只取1条
                .size(1);
        // 将两个聚合结合起来，topHits 作为分组的子聚合
        groupAggregation.subAggregation(topHits);
​
        // 求分组之后的总数
        CardinalityAggregationBuilder totalAggregation = AggregationBuilders.cardinality("total")
                .field("categoryId");
        // 把聚合放在搜索构造器里
        searchSourceBuilder.aggregation(groupAggregation);
        searchSourceBuilder.aggregation(totalAggregation);
​
        searchRequest.source(searchSourceBuilder);
        try {
            return client.search(searchRequest, COMMON_OPTIONS);
        } catch (IOException e) {
            log.error("Failed to aggregations search data", e);
        }
        return null;
    }

ini复制代码    @Override
    public List<CategoryGroup> categoryGroup() {
        SearchResponse groupSearchResponse = searchGroupData();
        if (Objects.isNull(groupSearchResponse)) {
            return Collections.emptyList();
        }
        Aggregations aggregations = groupSearchResponse.getAggregations();
        List<CategoryGroup> groupList = new ArrayList<>();
        if (Objects.nonNull(aggregations)) {
            // 获取最外层的分组
            // 这里我们分组的条件是categoryId，ES类型为long，所以需要使用 ParsedLongTerms 来接收结果
            ParsedLongTerms group = aggregations.get("group");
            // 获取总数的 aggregation
            ParsedCardinality totalCount = aggregations.get("total");
            // 这里的总数我们暂时不返回，只是打印
            log.info("得到分组的总数：{}", totalCount);
            // 遍历得到的桶
            for (Terms.Bucket bucket : group.getBuckets()) {
                // 获取每个桶中的top1数据
                ParsedTopHits topHits = bucket.getAggregations().get("price_top_1");
                SearchHits groupHits = topHits.getHits();
                SearchHit[] hits = groupHits.getHits();
                List<Book> books = new ArrayList<>();
                Arrays.stream(hits).forEach(hit -> {
                    Map<String, Object> sourceAsMap = hit.getSourceAsMap();
                    Book book = BeanUtil.mapToBean(sourceAsMap, Book.class, false, new CopyOptions());
                    if (Objects.nonNull(book.getId())) {
                        books.add(book);
                    }
                });
                CategoryGroup categoryGroup = new CategoryGroup();
                // 键值即为类目ID
                categoryGroup.setCategoryId(Integer.valueOf(bucket.getKeyAsString()));
                categoryGroup.setBooks(books);
                groupList.add(categoryGroup);
            }
        }
​
        return groupList;
    }