跳到主要内容

ThreadLocal用法与问题

使用场景​

线程绑定连接实现线程隔离​

在数据库操作中，事务的控制通常依赖于连接。为了实现线程内的事务隔离，我们可以使用 Thread Local 为每个线程绑定一个连接。

String outUser = "zhangan";
String inUser = "lisi";
int money = 10086;
boolean result = accountService.transfer(outUser, inUser, money);

//转账
public boolean transfer(String outUser, String inUser, int money) {
    Connection conn = null;
    try {
        //1. 开启事务
        conn = jdbcUtils.getConnection();
        conn.setAutoCommit(false);
        // 转出
        accountDao.out(outUser, money);

        //算术异常: 模拟转出成功,转入失败

        int i = 1 / 0;
        // 转入
        accountDao.in(inUser, money);
        //2. 成功提交
        jdbcUtils.commitAndClose(conn);
    } catch (Exception e) {
        e.printStackTrace();
        //2. 或者失败回滚
        jdbcUtils.rollbackAndClose(conn);
        return false;
    }
    return true;
}

JDBCUtils:

我们使用 Thread Local 来存储每个线程的连接。在需要进行数据库操作时，首先从 Thread Local 中获取连接，如果没有则从连接池中获取一个连接并绑定到当前线程。这样可以确保每个线程都有自己独立的连接，从而更好地控制事务。

@Component
public class JdbcUtils {

    static ThreadLocal<Connection> tl = new ThreadLocal<>();

    @Autowired
    private DruidDataSource dataSource;

    // 获取连接
    /*
    * 原本: 直接从连接池中获取连接
    * 现在:
    *       1. 直接获取当前线程绑定的连接对象
    *       2. 如果连接对象是空的
    *           2.1 再去连接池中获取连接
    *           2.2 将此连接对象跟当前线程进行绑定
    * */
    public  Connection getConnection() throws SQLException {
        Connection conn = tl.get();
        if(conn == null){
            conn = dataSource.getConnection();
            tl.set(conn);
        }
        return conn;
    }
    //释放资源
    public  void release(AutoCloseable... ios){
        for (AutoCloseable io : ios) {
            if(io != null){
                try {
                    io.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public  void commitAndClose(Connection conn) {
        try {
            if(conn != null){
                //提交事务
                conn.commit();
                //解绑当前线程绑定的连接对象
                tl.remove();
                //释放连接
                conn.close();
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    public  void rollbackAndClose(Connection conn) {
        try {
            if(conn != null){
                //回滚事务
                conn.rollback();
                //解绑当前线程绑定的连接对象
                tl.remove();
                //释放连接
                conn.close();
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}

线程内保存全局变量​

在 Web 应用中，我们经常需要在拦截器中获取用户信息，并在后续的方法中使用。使用 Thread Local 可以避免参数传递的麻烦。

@Component
public class LoginUserInterceptor implements HandlerInterceptor {
    public static ThreadLocal<User> loginUser = new ThreadLocal<>();
    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {

        //获取登录的用户信息
        User attribute = (User) request.getSession().getAttribute(AuthServerConstant.LOGIN_USER);

        if (attribute != null) {
            //把登录后用户的信息放在ThreadLocal里面进行保存
            loginUser.set(attribute);
            return true;
        } else {
            //未登录，返回登录页面
            return false;
        }
    }
    @Override
    public void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler, ModelAndView modelAndView) throws Exception {
        loginUser.remove();
    }
}

代替参数传递​

在方法调用的过程中，如果有一个参数需要在多个方法之间传递，并且中间有很多复杂的逻辑处理，直接使用参数传递会造成代码冗余。这时可以使用 Thread Local 来代替参数传递。

    static final ThreadLocal<String> threadLocal = new ThreadLocal();

    @Autowired
    private AService aService;

    public void dealTransfer() {
        String address = "nanjing";
        threadLocal.set(address);
        A();
    }

    private void A() {
        B();
    }

    private void B() {
        aService.C();
    }

们在方法 A 中设置了一个值到 ThreadLocal 中，然后在方法 B 中调用了 AService 的方法 C。在方法 C 中，可以直接从 Thread Local 中获取这个值，而不需要通过参数传递。

每个线程独享工具类对象​

在多线程环境下，一些工具类可能不是线程安全的。例如，SimpleDateFormat在多线程环境下使用可能会出现问题。我们可以使用 ThreadLocal 来为每个线程提供一个独享的工具类对象，每个线程都有自己独立的对象副本，避免了线程不安全的问题。

public class ThreadSafeFormatterUtil {
    public static ThreadLocal<SimpleDateFormat> dateFormatThreadLocal = new ThreadLocal<SimpleDateFormat>() {
        //创建一份 SimpleDateFormat 对象
        @Override
        protected SimpleDateFormat initialValue() {
            return new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
        }
    };
}
for (int i = 0; i < 30; i++) {
    new Thread(new Runnable() {
        @Override
        public void run() {
            SimpleDateFormat simpleDateFormat = ThreadSafeFormatterUtil.dateFormatThreadLocal.get();
            String date = simpleDateFormat.format(new Date());
            System.out.println(date);
        }
    }).start();
}

子线程如何获取父线程ThreadLocal的值?​

代码演示

/**
 * @author houyunfei
 * 子线程如何访问父线程的ThreadLocal
 */
public class ThreadLocalDemo {
    public static void main(String[] args) {
        Thread parentThread = new Thread(() -> {
            ThreadLocal<String> threadLocal = new ThreadLocal<>();
            threadLocal.set("父线程的ThreadLocal");
            InheritableThreadLocal<String> inheritableThreadLocal = new InheritableThreadLocal<>();
            inheritableThreadLocal.set("父线程的InheritableThreadLocal");
            Thread childThread = new Thread(() -> {
                System.out.println("子线程获取父线程的ThreadLocal：" + threadLocal.get());
                System.out.println("子线程获取父线程的InheritableThreadLocal：" + inheritableThreadLocal.get());
            }, "子线程");
            childThread.start();
        }, "父线程");
        parentThread.start();
    }
}

这样有问题，子线程拿不到父线程的ThreadLocal，但是我们可以使用InheritableThreadLocal来解决这个问题

image-20240908230339035

异步线程间的数据传递​

原来的问题：

public class UserUtils {
    private static final ThreadLocal<String> userLocal = new ThreadLocal<>();
    public static String getUserId() {
        return userLocal.get();
    }
    public static void setUserId(String userId) {
        userLocal.set(userId);
    }
    public static void clear() {
        userLocal.remove();
    }
}

我们每次创建线程，都需要手动设置，来传递值

public void handlerAsync() {
    //1. 获取父线程的userId
    UserUtils.setUserId("232");
    String userId = UserUtils.getUserId();
    log.info("父线程的值：{}", userId);
    CompletableFuture.runAsync(() -> {
        //2. 设置子线程的值，复用
        UserUtils.setUserId(userId);
        log.info("子线程的值：{}", UserUtils.getUserId());
    });
}

定义一个装饰器：

public class CustomTaskDecorator implements TaskDecorator {
    @Override
    public Runnable decorate(Runnable runnable) {
        String robotId = UserUtils.getUserId();
        System.out.println(robotId);
        return () -> {
            try {
                // 将主线程的请求信息，设置到子线程中
                UserUtils.setUserId(robotId);
                // 执行子线程，这一步不要忘了
                runnable.run();
            } finally {
                // 线程结束，清空这些信息，否则可能造成内存泄漏
                UserUtils.clear();
            }
        };
    }
}

设置给线程池：

@Configuration
@EnableAsync
@Slf4j
public class AsyncTaskPoolConfig {
    @Bean("taskExecutor")
    public ThreadPoolExecutor taskExecutor() {
        int i = Runtime.getRuntime().availableProcessors();
        System.out.println("系统最大线程数：" + i);
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(i,
                i+1, 5, TimeUnit.SECONDS, new LinkedBlockingDeque<>(200),new NamedThreadFactory("execl导出线程池"));
        System.out.println("execl导出线程池初始化完毕-------------");
        return threadPoolExecutor;
    }

    @Bean(name = "asyncServiceExecutor")
    public ThreadPoolTaskExecutor asyncServiceExecutor() {
        log.info("start asyncServiceExecutor----------------");
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //配置核心线程数
        executor.setCorePoolSize(10);
        //配置最大线程数
        executor.setMaxPoolSize(50);
        //配置队列大小
        executor.setQueueCapacity(200);
        // rejection-policy：当pool已经达到max size的时候，如何处理新任务
        // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //增加线程池修饰类
        executor.setTaskDecorator(new CustomTaskDecorator());
        executor.initialize();
        log.info("end asyncServiceExecutor------------");
        return executor;
    }
}

使用，指定线程池：

/**
 * 线程池设置TaskDecorator
 */
@Test
public void handlerAsync2() {
    UserUtils.setUserId("232");
    log.info("父线程的用户信息：{}", UserUtils.getUserId());
    //执行异步任务，需要指定的线程池
    CompletableFuture.runAsync(() ->
                    log.info("子线程的用户信息：{}", UserUtils.getUserId()),
            threadPoolTaskExecutor);
}

第二种办法：

使用阿里的工具，可以解决InheritableThreadLocal只能在new Thread的时候传递本地变量的问题，无法用到线程池的问题

<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>transmittable-thread-local</artifactId>
    <version>2.14.2</version>
</dependency>

改造工具类：

public class UserUtilsTran {
    private static final TransmittableThreadLocal<String> threadLocal = new TransmittableThreadLocal<>();
    public static String get() {
        return threadLocal.get();
    }
    public static void set(String userId) {
        threadLocal.set(userId);
    }
    public static void clear() {
        threadLocal.remove();
    }
}

使用：

/**
 * TransmittableThreadLocal
 */

@Autowired
private ThreadPoolExecutor threadPoolExecutor;

@Test
public void handlerAsync3() {
    UserUtilsTran.set("213");
    log.info("父线程的用户信息：{}", UserUtilsTran.get());
    CompletableFuture.runAsync(()->
                    log.info("子线程的用户信息：{}", UserUtilsTran.get()),
            threadPoolExecutor);
}
0%