语源科技BlogJava-JAVA

JDK21 CompletableFuture 链式多线程处理任务方式

为自己代言 — Mon, 08 Dec 2025 08:41:00 GMT

springboot 3 web工程多线程查询任务处理方式：

@Configuration
public class AsyncThreadPoolConfig {
    // 获取 CPU 核心数
 private static final int CPU_CORES = Runtime.getRuntime().availableProcessors();

 /**
     * 自定义异步线程池 springboot 推荐返回 ThreadPoolTaskExecutor ，也可以返回JDK原生：Executor，ExcecutorService


 */
 @Bean(name = "asyncExecutor") // 线程池名称，用于 @Async 注解指定
 public Executor asyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

 // 核心线程数（默认线程数）：线程池创建时初始化的线程数量
 executor.setCorePoolSize(CPU_CORES);
 // 最大线程数：线程池最大可创建的线程数量
 executor.setMaxPoolSize(CPU_CORES * 2);
 // 队列容量：任务队列的最大长度，超过核心线程数的任务会先进入队列等待
 executor.setQueueCapacity(10000);
 // 线程空闲时间：当线程数超过核心线程数时，多余线程的最大空闲时间（单位：秒）
 executor.setKeepAliveSeconds(60);
 // 线程名称前缀：方便日志排查
 executor.setThreadNamePrefix("async-task-");
 // 拒绝策略：当任务队列满且线程数达到最大线程数时，如何处理新任务
 // ThreadPoolExecutor.CALLER_RUNS_POLICY：由提交任务的线程执行（避免任务丢失，适合非核心任务）
 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
 // 初始化线程池
 executor.initialize();
 return executor;
 }


List>> futureList = new ArrayList<>();
for (int i = 0; i < userIdList.size(); i += 500) {
    int end = Math.min(i + 500, userIdList.size());
 List batchIds = new ArrayList<>(userIdList.subList(i, end));
 // 指定自定义线程池
 CompletableFuture> future = CompletableFuture.supplyAsync(
            () -> gameUserTelegramInfoMapper.getBotNameByUserIds(batchIds,YesOrNoEnum.YES.getCode()), asyncExecutor)//使用自定义线程池
 .orTimeout(600, MILLISECONDS)//单次超时600ms
 .exceptionally(e -> {
                log.error("多线程批量查询用户信息完成!", e);
                return new ArrayList<>();
 });
 futureList.add(future);
}
//2: 等待所有任务完成并汇总结果
CompletableFuture allFuture = CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0]));
//3: 汇总结果
List userViews =allFuture.thenApply(v ->
                futureList.stream()
                        //单个任务异常兜底（exceptionally）
 .map(f -> f.exceptionally(e -> new ArrayList<>()).join())
                        .filter(result -> !result.isEmpty())
                        .flatMap(List::stream)
                        .collect(Collectors.toList())
        ).orTimeout(5, SECONDS)//4:总超时兜底,任务层面超时，替代try-catch捕获TimeoutException
 .exceptionally(e ->{
            //5:超时/异常时自动取消所有未完成任务,防止资源耗尽。
 futureList.forEach(f -> {
                if(!f.isDone()){
                    boolean cancel = f.cancel(true);
 log.info("取消子任务：{}", cancel);
 }
            });
 log.error("多线程批量查询用户信息完成!",e);
            return new ArrayList<>();
 }).join();//6:最终无参join（已通过orTimeout+exceptionally兜底，无阻塞风险）
log.info("多线程批量查询用户信息完成!结果数:{}",userViews.size());
return userViews;

为自己代言 2025-12-08 16:41 发表评论

springboot3+springcloudstream4.x配置集成

为自己代言 — Thu, 31 Oct 2024 03:27:00 GMT

摘要: ringboot3+springcloudstream4.x配置集成springcloudstream3.X 后逐渐淘汰了 @input @output @EnableBinding 这些注解到4.X后这个注解都没有了，全部转向function 方式(关于function可以了解下)组件版本：springboot 3、 springcloud 2022.0.0、springc... 阅读全文

为自己代言 2024-10-31 11:27 发表评论

springboot3.X + nacos2.3 集成使用备忘

为自己代言 — Sun, 27 Oct 2024 14:16:00 GMT

1：nacos2.X 配置中心三个主要的参数 dataId、group、namespace

其中dataid 是指配置文件名称；group:相当于给配置文件分类；namespace:用来隔离环境的(例如：dev,test,product)

比较关注的点是 bootstrap.yml； application.yml 这两文件用途和加载顺序

bootstrap.yml 用于配置nacos地址，用户名，密码，命名空间(相当要先连接上nacos),然后才能拉去nacos上的配置文件信息;
application.yml 文件用于配置在本地的配置文件(其实这个也可以放到nacos上)
bootstrap配置文件是spring cloud新增的启动配置文件，需要引入spring-cloud-context依赖后，才会进行加载(看spring cloud 2022版本以以前)。
- bootstrap由父ApplicationContext加载，所以比application优先加载
- 因为bootstrap优先于application加载，所以不会被覆盖
- 使用配置中心spring cloud config时，需要在bootstarp中配置配置中心的地址，从而实现父ApplicationContext加载时，从配置中心拉去相应的配置到应用中。
它们加载顺序：bootstrap.yaml文件生效后，去nacos拉去完配置信息后，与本地的application.yaml配置信息进行合并，然后加载到spring容器中

在springboot工程中使用基础上使用两个注释：@RefreshScope 放在类上和@value 配置使用
要实现取值和动态刷新通过实验证明使用@NacosValue是不行了，因为使用了Spring Cloud的依赖包，所以根据官方文档显示，要通过 Spring Cloud 原生注解 @RefreshScope + @Value 来实现配置自动更新，我们可以用下面的方法达到同样的效。
nacos client 从服务端拉取配置信息会放到client 本地缓存起来默认目录：${user}\nacos\config 下(拉到调试时候有用)

详细文章：https://developer.aliyun.com/article/859891

为自己代言 2024-10-27 22:16 发表评论

软件微服务体系中的分层设计和领域划分

为自己代言 — Sun, 26 Sep 2021 10:09:00 GMT

摘要: 1：软件分层图：2：详细分层架构图：3：适用业务快速发展的分层图COLA 是 Clean Object-Oriented and Layered Architecture的缩写，代表“整洁面向对象分层架构”。目前COLA已经发展到COLA 4.0。COLA分为两个部分，COLA架构和COLA组件。https://github.com/alibaba/COLA?spm=at... 阅读全文

为自己代言 2021-09-26 18:09 发表评论

InnoDB的锁的机制探究

为自己代言 — Mon, 12 Jul 2021 07:38:00 GMT

InnoDB的锁

InnoDB的行锁：共享锁、排他锁、MDL锁

共享锁：又称读锁、S锁。一个事务获取一个数据行的共享锁，其他事务能获取该行对应的共享锁，但不能获得排他锁；即一个事务在读取一个数据行时，其他事务也可以读，但不能对数据进行增删改查。

应用：

1.自动提交模式下的select查询，不加任何锁，直接返回查询结果

2.通过select……lock in share mode在被读取的行记录或范围上加一个读锁，其他事务可以读，但是申请加写锁会被阻塞

排他锁：又称写锁、X锁。一个事务获取了一个数据行的写锁，其他事务就不能再获取该行的其他锁，写锁优先级最高。

应用：

1.一些DML操作会对行记录加写锁

2.select for update会对读取的行记录上加一个写锁，其他任何事务都不能对锁定的行加任何锁，否则会被阻塞

MDL锁：MySQL5.5引入，用于保证表中元数据的信息。在会话A中，表开启了查询事务后，会自动获得一个MDL锁，会话B就不能执行任何DDL语句的操作

行锁实现方式

InnoDB 行锁是通过给索引上的索引项加锁来实现的，这一点 MySQL 与 Oracle 不同，后者是通过在数据块中对相应数据行加锁来实现的。InnoDB 这种行锁实现特点意味着:只有通过索引条件检索数据，InnoDB 才使用行级锁，否则，InnoDB 将使用表锁! 在实际应用中，要特别注意 InnoDB 行锁的这一特性，不然的话，可能导致大量的锁冲突，从而影响并发性能。

行锁的三种算法

InnoDB 存储引擎有三种行锁的算法，其分别是：

Record Lock: 单个行记录上的锁
Gap Lock: 间隙锁，锁定一个范围，但不包含记录本身
Next-Key 锁: Gap Lock + Record Lock，锁定一个范围，并且会锁定记录本身

RC模式下只采用Record Lock，RR模式下采用了Next-Key

加锁场景分析

主键索引

如果我们加锁的行上存在主键索引，那么就会在这个主键索引上添加一个 Record Lock。

辅助索引

如果我们加锁的行上存在辅助索引，那么我们就会在这行的辅助索引上添加 Next-Key Lock，并在这行之后的辅助索引上添加一个 Gap Lock

辅助索引上的 Next-Key Lock 和 Gap Lock 都是针对 Repeatable Read 隔离模式存在的，这两种锁都是为了防止幻读现象的发生。

唯一的辅助索引

这里有一个特殊情况，如果辅助索引是唯一索引的话，MySQL 会将 Next-Key Lock 降级为 Record Lock，只会锁定当前记录的辅助索引。

如果唯一索引由多个列组成的，而我们只锁定其中一个列的话，那么此时并不会进行锁降级，还会添加 Next-Key Lock 和 Gap Lock。

Insert 语句

在 InnoDB 存储引擎中，对于 Insert 的操作，其会检查插入记录的下一条记录是否被锁定，若已经被锁定，则不允许查询。

意向锁

意向锁可以分为意向共享锁(Intention Shared Lock, IS)和意向排他锁(Intention eXclusive Lock, IX)。但它的锁定方式和共享锁和排他锁并不相同，意向锁上锁只是表示一种“意向”,并不会真的将对象锁住，让其他事物无法修改或访问。例如事物T1想要修改表test中的行r1，它会上两个锁:

在表test上意向排他锁
在行r1上排他锁

事物T1在test表上上了意向排他锁，并不代表其他事物无法访问test了，它上的锁只是表明一种意向，它将会在db中的test表中的某几行记录上上一个排他锁。

	意向共享锁	意向排他锁	共享锁	排他锁
意向共享锁	兼容	兼容	兼容	不兼容
意向排他锁	兼容	兼容	不兼容	不兼容
共享锁	兼容	不兼容	兼容	不兼容
排他锁	不兼容	不兼容	不兼容	不兼容

一致性非锁定读

一致性非锁定读是指 InnoDB 存储引擎通过行多版本控制(multi version)的方式来读取当前执行时间数据库中行的数据。具体来说就是如果一个事务读取的行正在被锁定，那么它就会去读取这行数据之前的快照数据，而不会等待这行数据上的锁释放。这个读取流程如图1所示:

行的快照数据是通过undo段来实现的，而undo段用来回滚事务，所以快照数据本身没有额外的开销。此外，读取快照数据时不需要上锁的，因为没有事务会对快照数据进行更改。

MySQL 中并不是每种隔离级别都采用非一致性非锁定读的读取模式，而且就算是采用了一致性非锁定读，不同隔离级别的表现也不相同。在 READ COMMITTED 和 REPEATABLE READ 这两种隔离级别下，InnoDB存储引擎都使用一致性非锁定读。但是对于快照数据，READ COMMITTED 隔离模式中的事务读取的是当前行最新的快照数据，而 REPEATABLE READ 隔离模式中的事务读取的是事务开始时的行数据版本。

一致性锁定读

在 InnoDB 存储引擎中，select语句默认采取的是一致性非锁定读的情况，但是有时候我们也有需求需要对某一行记录进行锁定再来读取，这就是一致性锁定读。

InnoDB 对于select语句支持以下两种锁定读:

select ... for update
select ... lock in share mode

select ... for update会对读取的记录加一个X锁，其他事务不能够再来为这些记录加锁。select ... lock in share mode会对读取的记录加一个S锁，其它事务能够再为这些记录加一个S锁，但不能加X锁。

对于一致性非锁定读，即使行记录上加了X锁，它也是能够读取的，因为它读取的是行记录的快照数据，并没有读取行记录本身。

select ... for update和select ... lock in share mode这两个语句必须在一个事务中，当事务提交了，锁也就释放了。因此在使用这两条语句之前必须先执行begin, start transaction，或者执行set autocommit = 0。

InnoDB 在不同隔离级别下的一致性读及锁的差异

consisten read //一致性读
share locks //共享锁
Exclusive locks //排他锁

		读未提交	读已提交	可重复读	串行化
SQL	条件
select	相等	None locks	Consisten read/None lock	Consisten read/None lock	Share locks
	范围	None locks	Consisten read/None lock	Consisten read/None lock	Share Next-Key
update	相等	Exclusive locks	Exclusive locks	Exclusive locks	Exclusive locks
	范围	Exclusive next-key	Exclusive next-key	Exclusive next-key	Exclusive next-key
Insert	N/A	Exclusive locks	Exclusive locks	Exclusive locks	Exclusive locks
Replace	无键冲突	Exclusive locks	Exclusive locks	Exclusive locks	Exclusive locks
	键冲突	Exclusive next-key	Exclusive next-key	Exclusive next-key	Exclusive next-key
delete	相等	Exclusive locks	Exclusive locks	Exclusive locks	Exclusive locks
	范围	Exclusive next-key	Exclusive next-key	Exclusive next-key	Exclusive next-key
Select … from … Lock in share mode	相等	Share locks	Share locks	Share locks	Share locks
	范围	Share locks	Share locks	Exclusive next-key	Exclusive next-key
*Select from … For update**	相等	Exclusive locks	Exclusive locks	Exclusive locks	Exclusive locks
	范围	Exclusive locks	Exclusive locks	Exclusive next-key	Exclusive next-key
Insert into … Select …	innodb_locks_ unsafe_for_bi nlog=off	Share Next-Key	Share Next-Key	Share Next-Key	Share Next-Key
(指源表锁)	innodb_locks_ unsafe_for_bi nlog=on	None locks	Consisten read/None lock	Consisten read/None lock	Share Next-Key
create table … Select …	innodb_locks_ unsafe_for_bi nlog=off	Share Next-Key	Share Next-Key	Share Next-Key	Share Next-Key
(指源表锁)	innodb_locks_ unsafe_for_bi nlog=on	None locks	Consisten read/None lock	Consisten read/None lock	Share Next-Key

在了解 InnoDB 锁特性后，用户可以通过设计和 SQL 调整等措施减少锁冲突和死锁，包括:

尽量使用较低的隔离级别;
精心设计索引，并尽量使用索引访问数据，使加锁更精确，从而减少锁冲突的机会;
选择合理的事务大小，小事务发生锁冲突的几率也更小;
给记录集显示加锁时，最好一次性请求足够级别的锁。比如要修改数据的话，最好直接申请排他锁，而不是先申请共享锁，修改时再请求排他锁，这样容易产生死锁;
不同的程序访问一组表时，应尽量约定以相同的顺序访问各表，对一个表而言，尽可能以固定的顺序存取表中的行。这样可以大大减少死锁的机会;
尽量用相等条件访问数据，这样可以避免间隙锁对并发插入的影响;
不要申请超过实际需要的锁级别;除非必须，查询时不要显示加锁;
对于一些特定的事务，可以使用表锁来提高处理速度或减少死锁的可能。

参考资料

1.https://dev.mysql.com/doc/refman/5.7/en/innodb-locking.html#innodb-intention-locks mysql官网开发手册

2.《MySQL 技术内幕 – InnoDB 存储引擎》

3.《深入浅出MySQL》

4.https://www.modb.pro/db/33873

为自己代言 2021-07-12 15:38 发表评论

大厂链路追踪架构图(鹰眼)

为自己代言 — Fri, 04 Jun 2021 07:36:00 GMT

整体架构

从日志生成到抓取、存储、分析、展现的多个系统间交互过程。

EagleEye 的核心

TraceId

在复杂的分布式系统环境下，EagleEye是一个有广泛用途的调用分析和问题排查工具。与一般的调用信息埋点日志相比，EagleEye埋点的一个显著的不同点在于它的每条日志都有与每次请求关联的上下文ID，我们称为TraceId。通过TraceId，后期的日志处理时可以把一次前端请求在不同服务器记录的调用日志关联起来，重新组合成当时这个请求的调用链。因此，EagleEye不仅可以分析到应用之间的直接调用关系，还可以得到他们的间接调用关系、以及上下游的业务处理信息；对于调用链的底层系统，可以追溯到它的最上层请求来源以及中间经过的所有节点；对于调用链的上层入口，可以收集到它的整棵调用树，从而定位下游系统的处理瓶颈，当下游某个应用有异常发生时，能迅速定位到问题发生的位置。

如上图所示，应用A是接受到来自用户浏览器的Web请求的前端服务器，它是一条调用链的开始端，在TBSession和EagleEyeFilter中都做了EagleEye上下文埋点。请求收到后它会先调用EagleEye StartTrace生成TraceId并放置在当前线程的ThreadLocal，日志埋点请求信息（如URL、SessionId、UserId等）。在请求处理完毕提交相应时，再调用EndTrace清理线程中的EagleEye信息。在应用A调用应用B、C的HSF服务，或者发送Notify消息时，TraceId被包含在EagleEye上下文中，随网络请求到达应用B、C、D、E之中，并放置在线程ThreadLocal内，因此后续调用到的这些系统都会有EagleEye这次请求的上下文。这些系统再发起网络请求时，也类似的携带了上下文信息的。

RpcId

为了区别同一个调用链下多个网络调用的顺序和嵌套层次，EagleEye还需要传输和记录RpcId。 RpcId用0.X1.X2.X3.....Xi来表示，Xi都是非负整数，根节点的RpcId固定从0开始，第一层网络调用的RpcId是0.X1，第二层的则为0.X1.X2，依次类推。*例如，从根节点发出的调用的RpcId是0.1、0.2、0.3，RpcId是0.1的节点发出的RpcId则为0.1.1、0.1.2、0.1.3。如下图所示

通过RpcId，可以准确的还原出调用链上每次调用的层次关系和兄弟调用之间的先后顺序。例如上图应用 G 的两次调用0.2.1.1和0.1.2.1，可以看出对 DB 的访问0.2.1.1源于 C 到 G 的调用0.2.1，对 Tair 的访问0.1.2.1源于B 到 G 的调用0.1.2。很多调用场景会比上面说的完全同步的调用更为复杂，比如会遇到异步、单向、广播、并发、批处理等等，这时候需要妥善处理好ThreadLocal上的调用上下文，避免调用上下文混乱和无法正确释放。另外，采用多级序号的RpcId设计方案会比单级序号递增更容易准确还原当时的调用情况。

为自己代言 2021-06-04 15:36 发表评论

分布式锁实现-redis,zk

为自己代言 — Wed, 24 Mar 2021 12:11:00 GMT

1:分布锁有好多实现方式

基于数据库实现

这个实现方式比较复杂,考虑因素比较多，比如：超时,非公平锁,非重入等会有各种各样的问题，在解决问题的过程中会使整个方案变得越来越复杂。操作数据库需要一定的开销，性能问题需要考虑

基于redis实现(这个对于不太敏感的场景可以使用，由于redis集群和单机，还有客户端，版本等多方面因素考虑情况比较多)

性能好。使用缓存实现分布式锁的缺点其数据库一样

基于zookeeper实现(这个是最终也是最好最可靠的)

创建临时节点，可以解决单机，锁无法释放，非阻塞，不可冲入，非公平的问题

总结

从理解的难易程度角度（从低到高）

数据库 > 缓存 > Zookeeper

从实现的复杂性角度（从低到高）

Zookeeper > 缓存 > 数据库

从性能角度（从高到低）

缓存 > Zookeeper >= 数据库

从可靠性角度（从高到低）

Zookeeper > 缓存 > 数据库

下面讲基于redis实现分布锁代码：RedisTemplate 客户端 lettuce

@Service
public class RedisDistributedLockUtils {

    @Autowired
    private RedisTemplate redisTemplate;

    private static final Long RELEASE_SUCCESS = 1L;

    private static final long DEFAULT_TIMEOUT = 1000 * 10;
    //因为要使用lua 脚本是因为 redis 执行lua脚本是原子操作
    private static final String UNLOCK_LUA= "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";

    /**
     * 实时获取锁
     *
     * 尝试获取分布式锁将redis版本升级到2.1以上(spring-boot-starter-data-redis 版本 2.X以上)，然后使用setIfAbsent 不存在
     * 当setIfAbsent成功之后断开连接，下面设置过期时间的代码 stringRedisTemplate.expire(key,timeout);是无法执行的，这时候就会有大量没有过期时间的数据存在数据库
     * @param lockKey    锁
     * @param requestId  请求标识
     * @param expireTime 超期时间
     * @return 是否获取成功
     */
    public boolean trySetDistributedLock(String lockKey, String requestId, long expireTime) {
        return redisTemplate.opsForValue().setIfAbsent(lockKey, requestId,0 == expireTime ? DEFAULT_TIMEOUT : expireTime, TimeUnit.MILLISECONDS);
    }

    /**
     * 以阻塞方式的获取锁
     * @param key
     * @param value
     * @param timeout
     * @return
     */
    public boolean setDistributedLock(String key, String value, long timeout) {
        Boolean lock = false;
        long start = System.currentTimeMillis();
        while (!lock && (System.currentTimeMillis() - start < timeout)) {
            //执行set命令
            lock = redisTemplate.opsForValue().setIfAbsent(key, value, timeout, TimeUnit.MILLISECONDS);
            //不频繁去获取锁
            try {
                if (!lock) {
                    Thread.sleep(60);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return lock;
    }

    public boolean releaseLock(String key, String value) {
        // 使用Lua脚本：先判断是否是自己设置的锁，再执行删除
        // 使用lua脚本删除redis中匹配value的key，可以避免由于方法执行时间过长而redis锁自动过期失效的时候误删其他线程的锁
        // spring自带的执行脚本方法中，集群模式直接抛出不支持执行脚本的异常EvalSha is not supported in cluster environment
        // 所以只能拿到原redis的connection来执行脚本

        List<String> keys = new ArrayList<>();
        keys.add(key);
        List<String> args = new ArrayList<>();
        args.add(value);
        Long result = (Long)redisTemplate.execute(new RedisCallback<Long>() {
            @Override
            public Long doInRedis(RedisConnection connection) throws DataAccessException {
                Object nativeConnection = connection.getNativeConnection();
                // 集群模式和单机模式虽然执行脚本的方法一样，但是没有共同的接口，所以只能分开执行
                // 集群模式
                if (nativeConnection instanceof JedisCluster) {
                    return (Long)((JedisCluster)nativeConnection).eval(UNLOCK_LUA, keys, args);
                }
                //客户端是Jedis时候(单机模式)
                else if (nativeConnection instanceof Jedis) {
                    return (Long)((Jedis)nativeConnection).eval(UNLOCK_LUA, keys, args);
                }
                //这里使用 redisTemplate 中lettuce 客户端
                else{
                    DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
                    redisScript.setScriptText(UNLOCK_LUA);
                    redisScript.setResultType(Long.class);
                    return (Long)redisTemplate.execute(redisScript, keys, value);
                }
            }
        });
        //返回最终结果
        return RELEASE_SUCCESS.equals(result);
    }
}

基于zookeeper实现下期补上：

介绍分布式锁文章写的比较详细：
https://blog.csdn.net/u010963948/article/details/79006572

为自己代言 2021-03-24 20:11 发表评论

JDK 8 函数式接口 Supplier、Function、Consumer、Predicate

为自己代言 — Fri, 24 Jul 2020 07:46:00 GMT

函数式接口的特征

1、三种方法

唯一的抽象方法
使用default定义普通方法（默认方法），通过对象调用。
使用static定义静态方法，通过接口名调用。

2、一个新注解@FunctionInterface

注解@FunctionalInterface告诉编译器这是一个函数式接口，明确这个函数中只有一个抽象方法，当你尝试在接口中编写多个抽象方法的时候编译器将不允许，但是可以有多个非抽象方法。
不过Object类的方法可以定义为抽象方法，因为接口的实现类一定是Object的子类
如果接口被标注了@FunctionalInterface，这个类就必须符合函数式接口的规范
即使一个接口没有标注@FunctionalInterface，如果这个接口满足函数式接口规则，依旧被当作函数式接口。

3、JDK 1.8 新增加的函数接口包：

    java.util.function.*
  java.util.function 它包含了很多接口，用来支持 Java的函数式编程，它们大致分为五类：



4、代码样例

/**
*JDK 8 函数式接口 Supplier、Function、Consumer、Predicate

     *
     * @param args
     * @throws Exception
     */

    public static void main(String[] args) throws Exception {
        ThreadPoolExecutor executor = (ThreadPoolExecutor)newFixedThreadPool(10);
        //1:JDK8以前，通过匿名内部类实现函数式接口
        executor.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("JDK8以前，通过匿名内部类实现函数式接口");
            }
        });
        //2:JDK8以后可以使用lambda 表达式来实现，lambda表达式就是为了优化匿名内部类而生(分开写效果)
        Thread thread = new Thread(() -> System.out.println("task running !"));
        Runnable r = () -> System.out.println("JDK8以后可以使用lambda 表达式来实现，lambda表达式就是为了优化匿名内部类而生(分开写效果)!");
        executor.submit(r);
        //3:合并起来效果
        executor.submit(() -> {
            System.out.println("JDK8以后可以使用lambda 表达式来实现，lambda表达式就是为了优化匿名内部类而生!");
        });

        //4:其它 Supplier、Function、Consumer、Predicate 都可以用lambda 表达式来实现
        Supplier<String> supplier = () -> "我是SuSupplier";
        Supplier<Integer> supplier2 = () -> new Integer(1);
        System.out.println("supplier=" + supplier.get() + ";supplier2=" + supplier2.get());

        //5: Function功能型函数式接口 Function 接受一个输入参数T，返回一个结果R
        Function<String,Integer> function=str -> Integer.parseInt(str);
        Function<Integer,String> fun2 = item -> item+"";
        System.out.println("输入字符型 1 返回int型结果："+function.apply("1"));
        System.out.println("输入整型 1 返回字符型结果："+fun2.apply(2));

        //6: Consumer 一个接受单个输入参数并且不返回结果的操作。与大多数其他函数接口不同， Consumer接口期望通过接受参数，改普通对象引用值(说明白点就是对原来的值进行加工，注意返回值 void)
        Consumer<StringBuffer> consumer= sb->sb.append("-yyy");
        StringBuffer sb1=new StringBuffer().append("111");
        consumer.accept(sb1);
        //改变sb的内部引用值
        System.out.println("=========s="+sb1.toString());

        //7: Predicate 断言型接口常用于集合的过滤，得到一个新的集合 Stream filter(Predicate predicate);
        Predicate<Integer> predicate = age -> age > 18;
        Predicate<String> predicate2 = str -> str != null;
        System.out.println(predicate.test(19));
        System.out.println(predicate2.test(null));
        //我们常用集合过滤类就是对这个接口实现类其中 filter(Predicate predicate) 用的就是这个接口
        List<String> list= Lists.newArrayList("1","2","2","3","4","4","8");
        list.stream().map(s -> Long.parseLong(s)).distinct().filter(s -> s < 10).collect(Collectors.toList()).forEach(u -> System.out.println(u));

        //总结，以上的例子其实都是JDK8 lambda 表达式简洁的写法，而且全是合并写的，并没有分开步骤写(所有函数性接口，都可以用lambda 表达式简洁写法)

        //关闭线程池
        executor.shutdownNow();
    }

为自己代言 2020-07-24 15:46 发表评论

JDK 8 CompletableFuture 详细使用介绍

为自己代言 — Thu, 23 Jul 2020 11:29:00 GMT


JDK 8 中 CompletableFuture 是对 Future 的增强 大大简化了异步编程步骤,在Spring 框架中配合@EnableAsync @Async 更加事办功倍。

1:在JDK 8 之前实现多线必需实现两个接口 Runnable 不带返回值，另一个Callable带返回值的接口,配合ThreadPoolTaskExecutor.submit(Callable callable) 返回一个Future对象。

使用Future获得异步执行结果时，要么调用阻塞方法get()，要么轮询看isDone()是否为true，这两种方法都不是很好，因为主线程也会被迫等待，而CompletableFuture出现改变了这个问题，而且提供更多并且强大的其它功能。

2：CompletableFuture简介

CompletableFuture implements Future, CompletionStage

其实CompletableFuture 除了实现原来的Future 接口外，其它大部分方法都是在CompletionStage中


大致介绍下completableFuture的命名规则：

1:按功能分类的话：
xxx()：表示该方法将继续在已有的线程中执行；

xxxAsync()：表示将异步在线程池中执行。

异步执行方法默认一个参数的话任务是在 ForkJoinPool.commonPool() 线程池中执行的，带executor 参数的使用 executor线程池异步执行。

2:按逻辑和组织方式来分话(completableFuture 中大约有50个来方法)
一种是 then 的逻辑，即前一个计算完成的时候调度后一个计算

一种是 both 的逻辑，即等待两个计算都完成之后执行下一个计算，只要能组合一个和另一个，我们就可以无限复用这个 +1 的逻辑组合任意多的计算

另一种是 either 的逻辑，即等待两个计算的其中一个完成之后执行下一个计算。注意这样的计算可以说是非确定性的。因为被组合的两个计算中先触发下一个计算执行的那个会被作为前一个计算，而这两个前置的计算到底哪一个先完成是不可预知的
3:从依赖关系和出入参数类型区别，基本分为三类：
apply 字样的方式意味着组合方式是 Function，即接受前一个计算的结果，应用函数之后返回一个新的结果

accept 字样的方式意味着组合方式是 Consumer，即接受前一个计算的结果，执行消费后不返回有意义的值

run 字样的方式意味着组合方式是 Runnable，即忽略前一个计算的结果，仅等待它完成后执行动作
其中出入参数主要有JDK8 Function，Consumer或Runnable三中函数型接口,每一种都决定了是怎么样一种依赖关系,我有一篇文章详细介绍了JDK8函数型接口用法，能有助理解completableFuture方法使用。
http://www.blogjava.net/zzzlyr/articles/435611.html
4:completableFuture 配合框架使用
  因为自从JDK8以后增强了多线程的使用便捷程度：
  1：JDk8 的函数式接口和lambda表过式
  2：completableFuture 对 Future 类的增强。
 这只是JDK 基础包中的功能，现在大部分开发都在使用框架 java 现在基本上都在使用spring框架，因为JDK基础包中的功能还是不如框架使用方便，下边文章详细介绍 springboot中对JDK基础包中多线程功能配置和使用。 
http://www.blogjava.net/zzzlyr/articles/435305.html

为自己代言 2020-07-23 19:29 发表评论

SpringBoot异步及线程池配置(jdk8新增加线程支持CompletableFuture)

为自己代言 — Wed, 25 Mar 2020 12:19:00 GMT

摘要: 如何在 Spring 使用@Async,@EnableAsync注释进行异步处理:异步处理适用那些与业务逻辑（横切关注点）不直接相关或者不作为其他业务逻辑输入的部分，也可在分布式系统中解耦。*译注：横切关注点（cross-cutting concerns）指一些具有横越多个模块的行为，使用传统的软件开发方法不能够达到有效模块化的一类特殊关注点。*Spring 中，`@Async`注解可以标记异步操... 阅读全文

为自己代言 2020-03-25 20:19 发表评论