场景
Java中ExecutorService线程池的使用(Runnable和Callable多线程实现):
https://blog.csdn.net/BADAO_LIUMANG_QIZHI/article/details/126242904
SpringBoot+Lombok+Builder实现任意个数属性的对象构造:
https://blog.csdn.net/BADAO_LIUMANG_QIZHI/article/details/126384244
上面讲了自定义线程池的使用以及Builder构造者模式的一个简单应用。
是否可以将两者结合使用,Hutool中已经实现了这一点。
文档说明
自定义线程池ExecutorBuilder
在JDK中,提供了Executors用于创建自定义的线程池对象ExecutorService,但是考虑到线程池中存在众多概念,这些概念通过不同的搭配实现灵活的线程管理策略,单独使用Executors无法满足需求,构建了ExecutorBuilder。
概念
corePoolSize 初始池大小
maxPoolSize
最大池大小(允许同时执行的最大线程数)
workQueue 队列,用于存在未执行的线程
handler
当线程阻塞(block)时的异常处理器,所谓线程阻塞即线程池和等待队列已满,无法处理线程时采取的策略
线程池对待线程的策略
1、如果池中任务数<corePoolSize,放入立即执行。
2、如果池中任务数>corePoolSize,放入队列等待。
3、队列满,新建线程立即执行。
4、执行中的线程>maxPoolSize,触发handler(RejectedExecutionHandler)异常。
workQuene线程池策略
SynchronousQueue它将任务直接提交给线程而不保持他们。当运行线程小于maxPoolSize时会创建新线程,否则触发异常策略。
LinkedBlockingQueue默认无界队列,当运行线程大于corePoolSize时始终放入此队列,此时maxPoolSize无效。当构造LinkedBlockingQuene
对象时传入参数,变成有界队列,队列满时,运行线程小于maxPoolSize时会创建新线程,否则触发异常策略。
ArrayBlockingQuene有界队列,相对无界队列有利于控制队列大小,队列满时,运行线程小于maxPoolSize时会创建新线程,否则触发异常策略。
注:
博客:
https://blog.csdn.net/badao_liumang_qizhi
关注公众号
霸道的程序猿
获取编程相关电子书、教程推送与免费下载。
实现
1、使用示例-默认线程池
ExecutorService executor = ExecutorBuilder builder = ExecutorBuilder.create()..build();
初始线程数为corePoolSize指定的大小
没有最大线程数限制
默认使用LinkedBlockingQueue,默认队列大小为1024(最大等待数1024)
当运行线程大于corePoolSize放入队列,队列满后抛出异常
2、使用示例-单线程线程池
ExecutorService executor = ExecutorBuilder.create()// .setCorePoolSize(1)// .setMaxPoolSize(1)// .setKeepAliveTime(0)// .build();
初始线程数为 1
最大线程数为 1
默认使用LinkedBlockingQueue,默认队列大小为1024
同时只允许一个线程工作,剩余放入队列等待,等待数超过1024报错
3、使用示例-更多选项
ExecutorService executor = ExecutorBuilder.create() .setCorePoolSize(5) .setMaxPoolSize(10) .setWorkQueue(new LinkedBlockingQueue<>(100)) .build();
初始5个线程
最大10个线程
有界等待队列,最大等待数是100
4、使用示例-特殊策略
ExecutorService executor = ExecutorBuilder.create() .setCorePoolSize(5) .setMaxPoolSize(10) .useSynchronousQueue() .build();
初始5个线程
最大10个线程
它将任务直接提交给线程而不保持它们。当运行线程小于maxPoolSize时会创建新线程,否则触发异常策略
5、业务中使用示例
引入项目依赖
<dependency> <groupId>cn.hutool</groupId> <artifactId>hutool-all</artifactId> <version>5.8.3</version> </dependency>
使用示例
private static ExecutorService pool = ExecutorBuilder.create() .setCorePoolSize(20)//初始20个线程 .setMaxPoolSize(40)//最大40个线程 .setWorkQueue(new LinkedBlockingQueue<>(60))//有界等待队列,最大等待数是60 .setThreadFactory(ThreadFactoryBuilder.create().setNamePrefix("IM-Pool-").build())//设置线程前缀 .build();
完整使用示例
package com.chrisf.imextend; import cn.hutool.core.thread.ExecutorBuilder; import cn.hutool.core.thread.ThreadFactoryBuilder; import com.chrisf.sdk.protocal.Protocal; import com.chrisf.sdk.utils.LocalSendHelper; import io.netty.channel.Channel; import java.util.Collection; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.LinkedBlockingQueue; /** * im 客户端用户集 * * @author yansh */ public class ImUsers { private static ExecutorService pool = ExecutorBuilder.create() .setCorePoolSize(20)//初始20个线程 .setMaxPoolSize(40)//最大40个线程 .setWorkQueue(new LinkedBlockingQueue<>(60))//有界等待队列,最大等待数是60 .setThreadFactory(ThreadFactoryBuilder.create().setNamePrefix("IM-Pool-").build())//设置线程前缀 .build(); /** * 用户集 */ private static Map<String, Channel> USERS = new ConcurrentHashMap<String, Channel>(); /** * 存储用户 * * @param key 唯一键 * @param session 用户信息 */ public static void put(String key, Channel session) { USERS.put(key, session); } /** * 移除用户 * * @param session 用户信息 * @return 移除结果 */ public static boolean remove(Channel session) { String key = null; boolean flag = USERS.containsValue(session); if (flag) { Set<Map.Entry<String, Channel>> entries = USERS.entrySet(); for (Map.Entry<String, Channel> entry : entries) { Channel value = entry.getValue(); if (value.equals(session)) { key = entry.getKey(); break; } } } else { return true; } return remove(key); } /** * 移出用户 * * @param key 键 */ public static boolean remove(String key) { Channel remove = USERS.remove(key); if (remove != null) { boolean containsValue = USERS.containsValue(remove); return containsValue; } else { return true; } } /** * 获取在线用户列表 * * @return 返回用户集合 */ public static Map<String, Channel> getUsers() { return USERS; } /** * 群发消息文本消息 * * @param protocal 消息内容 */ public static void sendMessageToUsersByText(Protocal protocal) { Collection<Channel> values = USERS.values(); for (Channel value : values) { pool.submit(() -> sendMessageToUserByText(value, protocal)); } } /** * 发送消息 * * @param session * @param protocal */ public static void sendMessageToUserByText(Channel session, Protocal protocal) { if (session != null) { synchronized (session) { try { LocalSendHelper.sendData(session,protocal,null); } catch (Exception e) { e.printStackTrace(); } } } else { } } }
标签:Java,自定义,队列,Hutool,static,session,线程,ExecutorBuilder,import From: https://www.cnblogs.com/badaoliumangqizhi/p/16719023.html