DoneHub
0%

发表于 更新于 分类于 Valine：

一、背景介绍

ThreadLocal是线程提供的本地变量，因其线程特殊属性，被经常用于存储与线程相关的信息，如保存登录用户信息、数据库连接配置等。但我们忽略了一个关键点：ThreadLocal 只能用在同步线程中，而在异步线程或线程池中则不起作用。因此，本博文主要探讨如何在异步线程和线程池中传递 ThreadLocal 上下文。

二、问题描述

发版之后，爆出线上问题：“用户信息获取失败”。为了更好地解释这个问题，先介绍下公司的系统架构：用户登录时，后台系统通过用户 key 获取个人信息并保存在 ThreadLocal 静态对象中，以供后续使用。简要实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
/**
* 拦截器
*/
@Component
public class HandlerAccessInterceptor implements HandlerInterceptor {
    
    @Override
    public boolean preHandle(HttpServletRequest httpServletRequest,
                             HttpServletResponse httpServletResponse, Object o) throws Exception {
        // 增加允许跨域的返回信息
        httpServletResponse.setHeader("Access-Control-Allow-Origin", "*");
        httpServletResponse.setHeader("Access-Control-Allow-Headers", "Content-Type,Content-Length, Authorization, Accept,X-Requested-With");
        httpServletResponse.setHeader("Access-Control-Allow-Methods", "PUT,POST,GET,DELETE,OPTIONS");
        
        // 获取并保存用户信息
        if (httpServletRequest.getCookies() != null) {
            for (Cookie cookie : httpServletRequest.getCookies()) {
                if ("vkey".equals(cookie.getName())) {
                    UserContextUtil.setUser(cookie.getValue());
                }
            }
        }
        
        return true;
    }
    
    @Override
    public void postHandle(HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse, Object o, ModelAndView modelAndView) throws Exception {

    }
    
    // 清除用户信息
    @Override
    public void afterCompletion(HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse, Object o, Exception e) throws Exception {
        UserContextUtil.remove();
    }
}

/**
* 用户信息工具
*/
@Component
public class UserContextUtil {
    private static ThreadLocal<SyUser> threadLocal = new ThreadLocal<>();
    private static final String TOKEN_BEARER = "Bearer ";

    /**
     * 获取当前线程用户信息
     *
     * @return 前线程用户信息
     */
    public synchronized static SyUser getUser() {
        SyUser syUser = threadLocal.get();
        return syUser;
    }

    /**
    * 直接保存用户信息
    */
    public synchronized static boolean setSyUser(SyUser syUser) {

        if (syUser == null) {
            return false;
        }

        threadLocal.set(syUser);

        return true;
    }

    /**
     * 转换登录用户信息为当期系统用户信息
     *
     * @param cookies
     */
    public synchronized static boolean setUser(Cookie[] cookies) {
        if (cookies == null) {
            return false;
        }
        for (Cookie cookie : cookies) {
            String name = cookie.getName().toLowerCase();
            if ("vkey".equals(name)) {
                String value = cookie.getValue();
                if (StrUtil.isEmpty(value)) {
                    return false;
                }
                LoginUser loginUser = new LoginUser(value);
                if (loginUser.getId() == null) {
                    return false;
                }
                SyUser syUser = new SyUser();
                syUser.setId(loginUser.getId().intValue());
                syUser.setUserName(loginUser.getMobile());
                syUser.setTrueName(loginUser.getName());
                return setSyUser(syUser);
            }
        }
        return false;
    }

    /**
     * 转换登录用户信息为当期系统用户信息
     *
     * @param headerValue 头部 vkey 值
     */
    public synchronized static boolean setUser(String headerValue) {
        if (StrUtil.isEmpty(headerValue)) {
            return false;
        }
        if (headerValue.startsWith(TOKEN_BEARER)) {
            headerValue = headerValue.substring(TOKEN_BEARER.length());
        }
        LoginUser loginUser = new LoginUser(headerValue);
        if (loginUser.getId() == null) {
            return false;
        }
        SyUser syUser = new SyUser();
        syUser.setId(loginUser.getId().intValue());
        syUser.setUserName(loginUser.getMobile());
        syUser.setTrueName(loginUser.getName());
        syUser.setPhone(loginUser.getMobile());
        return setSyUser(syUser);
    }
    
    /**
     * 获取登录用户姓名
     */
    public synchronized static String getUserName() {
        SyUser syUser = threadLocal.get();
        if (syUser == null) {
            return null;
        }
        return syUser.getTrueName();
    }

    /**
     * 移除线程中用户信息
     */
    public synchronized static void remove() {
        threadLocal.remove();
    }
}

问题代码定位：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@RestController
@RequestMapping("/web/file")
@Slf4j
public class ManageFileController {
    
    @Resource
    private ManageFileService manageFileService;
    
	/**
     * 【加载文件】
     * 下载文件的数据准备阶段可能非常长, 前台操作人员无法得知进度, 且无法做其他操作
     * 故将数据准备与文件下载解耦, 解决交互上不舒服的问题
     *
     * @param loadFileReqDTO 加载文件请求
     * @return 调用是否成功
     */
    @PostMapping("/loadFile")
    public RespDTO<String> loadFile(@Valid @RequestBody LoadFileReqDTO loadFileReqDTO) {

        // 初始化加载文件信息
        AsyncLoadFile asyncLoadFile = manageFileService.initLoadFileInfo(loadFileReqDTO);

        // 执行文件加载
        manageFileService manageFileService.loadFile(loadFileReqDTO, inheritedSyUser);

        return RespDTO.success();
    }
}

/**
 * @author zourongsheng
 * @version 1.0
 * @date 2020/10/21 13:55
 */
@Service
@Slf4j
public class ManageFileService {
    
    @Resource
    private ManageFileHelper manageFileHelper;
    
    @Resource
    private AsyncLoadFileMapper asyncLoadFileMapper;
    
    /**
     * 【加载文件】
     *
     * @param loadFileReqDTO 加载文件请求
     * @param asyncLoadFile  初始化加载文件信息
     * @param syUser         用户信息
     */
    @Async(TASK_EXECUTOR)
    public void loadFile(LoadFileReqDTO loadFileReqDTO, AsyncLoadFile asyncLoadFile, SyUser syUser) {

        try {

            log.info("异步加载文件开始; userName: {}", UserContextUtil.getUserName());

            // 路由加载服务
            LoadFileService loadFileService = manageFileHelper.router(loadFileReqDTO.getLoadFileType());

            // 开启文件加载过程
            LoadFileReqBO loadFileReqBO = new LoadFileReqBO();
            BeanUtils.copyProperties(loadFileReqDTO, loadFileReqBO);

            LoadFileRespBO loadFileRespBO = loadFileService.executeLoadFile(loadFileReqBO);

            // 备份文件至影像件系统
            String fileKey = FileUtil.uploadFile(loadFileRespBO.getFilePathUrl(), FileTypeEnum.XLSX, STORE_FILE_KEY);

            CuiShouAssert.notEmpty(fileKey, "调用影像件系统异常!");

            asyncLoadFile.setFileName(loadFileRespBO.getFileName());
            asyncLoadFile.setFileKey(fileKey);
            asyncLoadFile.setLoadStatus(LoadStatusEnum.SUCCESS.name());
        } catch (Exception e) {
            log.info("加载文件异常; errMsg: {}", e.getMessage(), e);
            asyncLoadFile.setLoadStatus(LoadStatusEnum.FAILURE.name());
            asyncLoadFile.setRemark(e.getMessage());
        }

        asyncLoadFileMapper.update(asyncLoadFile);
        
        log.info("异步加载文件结束; userName: {}", UserContextUtil.getUserName());
    }
}
1
2
运行结果...
2020-12-14 17:51:15.235  INFO [-,d11235d6f6eda8d0,8c023b1e45855f97,false] 19444 --- [common-async-executor-1] c.v.c.o.service.file.ManageFileService   : 异步加载文件开始; userName: null

问题很好定位，@Async 异步线程池开启子线程处理任务后，父线程提供的本地线程变量就销毁了。

三、解决方法

  • 将用户信息当作参数传入异步方法，再在异步方法中重新设置本地变量；
  • 采用 InheritedThreadLocal 实现子线程上下文的传递。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
/**
 * 【加载文件-方法一】
 * 下载文件的数据准备阶段可能非常长, 前台操作人员无法得知进度, 且无法做其他操作
 * 故将数据准备与文件下载解耦, 解决交互上不舒服的问题
 *
 * @param loadFileReqDTO 加载文件请求
 * @return 调用是否成功
 */
@PostMapping("/loadFile")
public RespDTO<String> loadFile(@Valid @RequestBody LoadFileReqDTO loadFileReqDTO) {

    // 初始化加载文件信息
    AsyncLoadFile asyncLoadFile = manageFileService.initLoadFileInfo(loadFileReqDTO);
    
    // #loadFile 为异步方法, 这里将操作人信息继承到子线程中
    SyUser syUser = UserContextUtil.getUser();

    SyUser inheritedSyUser = new SyUser();

    BeanUtils.copyProperties(syUser, inheritedSyUser);

    // 执行文件加载
    manageFileService manageFileService.loadFile(loadFileReqDTO, inheritedSyUser);

    return RespDTO.success();
}

/**
* 【加载文件】
* 下载文件的数据准备阶段可能非常长, 前台操作人员无法得知进度, 且无法做其他操作
* 故将数据准备与文件下载解耦, 解决交互上不舒服的问题
*
* @param loadFileReqDTO 加载文件请求
* @param asyncLoadFile  初始化加载文件信息
* @param syUser         用户信息
*/
@Async(TASK_EXECUTOR)
public void loadFile(LoadFileReqDTO loadFileReqDTO, AsyncLoadFile asyncLoadFile, SyUser syUser) {
    
    UserContextUtil.setSyUser(syUser);
    log.info("异步加载文件开始; UserName: {}", UserContextUtil.getRealName());
    UserContextUtil.remove();
    log.info("异步加载文件结束; UserName: {}", UserContextUtil.getRealName());
}
1
2
3
方法一运行结果...
2020-12-14 18:03:34.947  INFO [-,480f9c55e785ec6f,c8de532979871193,false] 19444 --- [common-async-executor-2] c.v.c.o.service.file.ManageFileService   : 异步加载文件开始; UserName: 邹荣升
2020-12-14 18:03:35.235  INFO [-,480f9c55e785ec6f,c8de532979871193,false] 19444 --- [common-async-executor-2] c.v.c.o.service.file.ManageFileService   : 异步加载文件结束; UserName: null

方法一可以说是傻瓜操作，治标不治本，后来的同事稍不注意还会掉进坑里。我们来实现方法二：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
private static ThreadLocal<SyUser> threadLocal = new InheritableThreadLocal<>();

/**
* 【加载文件-异步线程】
* 下载文件的数据准备阶段可能非常长, 前台操作人员无法得知进度, 且无法做其他操作
* 故将数据准备与文件下载解耦, 解决交互上不舒服的问题
*
* @param loadFileReqDTO 加载文件请求
* @param asyncLoadFile  初始化加载文件信息
* @param syUser         用户信息
*/
@Async
public void loadFile(LoadFileReqDTO loadFileReqDTO, AsyncLoadFile asyncLoadFile, SyUser syUser) {
    
    UserContextUtil.setSyUser(syUser);
    log.info("异步加载文件开始; UserName: {}", UserContextUtil.getRealName());
    UserContextUtil.remove();
    log.info("异步加载文件结束; UserName: {}", UserContextUtil.getRealName());
}

/**
* 【加载文件-异步线程池】
* 下载文件的数据准备阶段可能非常长, 前台操作人员无法得知进度, 且无法做其他操作
* 故将数据准备与文件下载解耦, 解决交互上不舒服的问题
*
* @param loadFileReqDTO 加载文件请求
* @param asyncLoadFile  初始化加载文件信息
* @param syUser         用户信息
*/
@Async(TASK_EXECUTOR)
public void loadFile(LoadFileReqDTO loadFileReqDTO, AsyncLoadFile asyncLoadFile, SyUser syUser) {
    
    UserContextUtil.setSyUser(syUser);
    log.info("异步加载文件开始; UserName: {}", UserContextUtil.getRealName());
    UserContextUtil.remove();
    log.info("异步加载文件结束; UserName: {}", UserContextUtil.getRealName());
}

异步线程和异步线程池两种情况下的运行结果：

1
2
3
4
运行结果...
2020-12-14 18:36:41.560  INFO [-,8658c694de3e9b7a,ccc7a782204c91b0,false] 19876 --- [common-async-executor-1] c.v.c.o.service.file.ManageFileService   : 异步加载文件开始; UserName: 邹荣升

2020-12-14 18:41:17.735  INFO [-,67b4725bcaacfc64,36ee2acec232be32,false] 21600 --- [common-async-executor-1] c.v.c.o.service.file.ManageFileService   : 异步加载文件开始; UserName: 邹荣升

三、总结

虽然 ThreadLocal 的线程相关属性提供了很多便利性，但在高并发系统中，直接使用原生静态对象并不合适。因此，如使用 Redis、MQ 一样，深入了解常用工具的特性，是系统设计的必备要素。

发表于 更新于 分类于 Valine：

最近在忙很多事，工作和生活都乱糟糟的，博客也停更了。写技术还是美食呢，当然是美食啦。

今天做萝卜丸子。边做边吃，忽然想起读初二的一天，同学们凑钱买了五斤萝卜丸子，然后在英语课上偷吃的故事。十五年过去了，但历历在目。

一、食材准备

青萝卜、胡萝卜、红薯淀粉、面粉、鸡蛋、辣椒、胡椒、葱姜蒜、生抽、老抽、蚝油、冰糖。

DOPsrd.jpg

二、调料子

首先将萝卜刨丝。

DOPjRU.jpg

然后要杀掉萝卜的水气和臭味。比较喜欢看老饭骨，大伯给的方子是在沸水中放入冰糖，再倒入萝卜丝。今天试了下，效果确实很棒，建议民间推广。

DOPgat.jpg

接下来调糊糊。用筷子顺着一个方向搅拌，直到可以用筷子提起来。这步也叫上劲，是关键一环。

DOiNLj.jpg

调糊糊前，可以先把油热上。

DOiXTI.jpg

待油七分热，将糊糊团成丸子，依次放入。油温太高会让丸子糊掉，太低又会导致丸子吸油太多。七分热怎么看呢？可以先放一小块糊糊进去，剧烈翻滚但短时间内又不会泛黄就可以了。

DOFTNq.jpg

文火慢炸20分钟，表面金黄时捞出。

DOFxb9.jpg

丸子放入次序不一样，可能炸出来的颜色也不一样。强迫症患者可以将丸子下入油中复炸一次，让颜色均匀。

萝卜丸子可以当零食吃，也可以做汤。

发表于 更新于 分类于 Valine：

一、InnoDB 索引

索引是数据库应用中非常重要的组成部分，可以提升数据检索效率，减轻服务引擎的压力。InnoDB 服务引擎，支持哈希索引和B+Tree索引。其中，B+Tree 是默认选择，也是最常用的索引结构。普通索引和唯一索引是最常用的两种索引类型，本文详细介绍一下两者的区别。

二、普通索引与唯一索引的作用

普通索引，规律分布在叶子节点上，唯一的目的就是方便随机查找和范围查找，提升检索效率。

唯一索引，不仅要实现普通索引的功能，还要保证值的唯一性。

三、普通索引与唯一索引的区别

change buffer

探讨两者的区别，需要从它们的作用机制入手。对于普通索引，change buffer 的作用至关重要，这也是其与唯一索引区别最大的地方。至于 change buffer 是什么，要从 MySQL 的数据管理机制说起。

为了减少 IO 读的次数，MySQL 将数据分别存在磁盘和内存中，磁盘中是原始数据，内存（buffer pool）中是缓存的数据页和索引页。可以说，buffer pool 是缓解磁盘读的一种手段。当然，MySQL 不会单单研究出来一个缓解磁盘读的机制，毕竟磁盘读和写都是十分消耗的操作，而 change buffer 就是缓解磁盘写的一种手段。

当需要更新一个数据页时，如果数据页在 buffer pool 中，则直接更新；如果不在缓存中，InnoDB 就会把更新操作缓存在 change buffer 中，而不会从磁盘中读取数据页到 buffer pool。在合适的时机，如再次访问这个数据页时，InnoDB 便会将数据页读入内存，然后执行 change buffer 中的相关操作。总之，在不影响数据逻辑正确性的前提下，减少了磁盘写的次数。

对于 InnoDB 数据引擎，将 change buffer 的操作应用到数据页的过程，叫 merge。以下场景会触发 merge 操作：

  • 访问这个数据页；
  • 系统后台线程定期 merge；
  • change buffer 容量不够用时；
  • 数据库正常关闭时；
  • redo log 写满时（固定大小，循环写）；

此外，虽然 change buffer 看起来像是一个临时缓存，但它却是可持久化的数据，不仅存于内存，也被写到磁盘上。这样的设计才是合理的，当出现突发状况时，可以通过磁盘上的 change buffer 恢复逻辑操作，保证数据的正确性和系统的容灾能力。

3.1 在查询方面的差异

id 是聚簇索引，k 是二级索引。可以看出，这是一个典型的索引覆盖场景。

1
select id from T where k = 10;

如果给 k 加上普通索引，那么随机查询将根据二分法找到第一个目标数据，然后再沿着叶子节点查找下一个数据，判断是否为 10，如果不是，则返回查询结果。

如果给 k 加上唯一索引，那么随机查询将根据二分法找到第一个目标数据，然后返回查询结果，因为这个目标数据一定是唯一的。

所以，在查询数据方面，普通索引比唯一索引多一个向下一个节点查找的步骤。如果下一个节点在同一个数据页，通过指针直接就可以拿到；如果下一个节点不在当前数据页，那么将会发生一次 IO 读。就是说，普通索引的查询性能弱于唯一索引。

3.2 在更新方面的差异

从 change buffer 的介绍可知，当数据页不在内存中，使用普通索引会将更新操作缓存在 change buffer 中，然后在适当的时机 merge 到原数据页中，从而减少 IO 次数；而使用唯一索引，为了保证数据的唯一性，需要将数据页读到内存中来，进行更新操作，再写回磁盘。就是说，普通索引的更新性能强于唯一索引。

发表于 更新于 分类于 Valine：

一、@Retention 简介

@Retention 是用来定义注解的注解，这类注解也叫 元注解。@Retention 定义了注解的生命周期，生命周期的长短取决于 RetentionPolicy 属性值。

1
2
3
4
5
6
7
8
9
10
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.ANNOTATION_TYPE)
public @interface Retention {
    /**
     * Returns the retention policy.
     * @return the retention policy
     */
    RetentionPolicy value();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public enum RetentionPolicy {
    /**
     * Annotations are to be discarded by the compiler.
     */
    SOURCE,

    /**
     * Annotations are to be recorded in the class file by the compiler
     * but need not be retained by the VM at run time.  This is the default
     * behavior.
     */
    CLASS,

    /**
     * Annotations are to be recorded in the class file by the compiler and
     * retained by the VM at run time, so they may be read reflectively.
     *
     * @see java.lang.reflect.AnnotatedElement
     */
    RUNTIME
}

归纳如下：

属性值 描述 使用场景
RetentionPolicy.SOURCE 注解只保留在源文件，当被编译成class文件，注解就会消失 代码预检，如 @Override、@SuppressWarnings
RetentionPolicy.CLASS 注解被保留到class文件，但当jvm加载class文件时被遗弃 编译时进行一些预处理操作，如生成辅助信息（META-INF/services）
RetentionPolicy.RUNTIME 注解不仅被保存到class文件中，jvm加载class文件之后，仍然存在 运行时动态获取注解信息，如 @Autowired、@Required

二、通过字节码认识 @Retention

定义三种生命周期的注解：SourcePolicy、ClassPolicy、RuntimePolicy
1
2
3
4
5
6
7
8
9
10
11
12
13
14
@Retention(RetentionPolicy.SOURCE)
@Target(ElementType.METHOD)
public @interface SourcePolicy {
}

@Retention(RetentionPolicy.CLASS)
@Target(ElementType.METHOD)
public @interface ClassPolicy {
}

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface RuntimePolicy {
}

使用以上注解：

1
2
3
4
5
6
7
8
9
10
11
12
13
public class RetentionPolicyTest {
    @SourcePolicy
    public void sourcePolicy() {
    }

    @ClassPolicy
    public void classPolicy() {
    }

    @RuntimePolicy
    public void runtimePolicy() {
    }
}

javap -v RetentionPolicyTest 查看字节码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
{
  public RetentionPolicyTest();
    flags: ACC_PUBLIC
    Code:
      stack=1, locals=1, args_size=1
         0: aload_0
         1: invokespecial #1                  // Method java/lang/Object."<init>":()V
         4: return
      LineNumberTable:
        line 3: 0

  public void sourcePolicy();
    flags: ACC_PUBLIC
    Code:
      stack=0, locals=1, args_size=1
         0: return
      LineNumberTable:
        line 7: 0

  public void classPolicy();
    flags: ACC_PUBLIC
    Code:
      stack=0, locals=1, args_size=1
         0: return
      LineNumberTable:
        line 11: 0
    RuntimeInvisibleAnnotations:
      0: #11()

  public void runtimePolicy();
    flags: ACC_PUBLIC
    Code:
      stack=0, locals=1, args_size=1
         0: return
      LineNumberTable:
        line 15: 0
    RuntimeVisibleAnnotations:
      0: #14()
}

可以看到：

  • 编译之后，@SourcePolicy 信息不存在；
  • 编译之后，@ClassPolicy 具有属性：RuntimeInvisibleAnnotations；
  • 编译之后，@RuntimePolicy 具有属性：RuntimeVisibleAnnotations；

发表于 更新于 分类于 Valine：

自从国内猪肉紧俏以来，市面上就经常买到进口猪肉。外国人养猪比较人性化，不给小猪做阉割，所以猪肉又腥又骚。今天抱着踩雷的心态，买了一份肉，并在叮咚上特意备注道：“只要中国猪肉！”。

今天来做个狮子头。

一、食材准备

五花肉（肥瘦比: 6:4），山药（马蹄/萝卜）、老豆腐、红薯淀粉、辣椒、葱姜蒜、生抽、老抽、蚝油、料酒、八角、鸡蛋。

wyDNmd.jpg

辅料我用的是山药，大家可以根据自己的口味调整。辅料添加要适量，否则可能团不成肉丸子。

二、制作流程
1
2
3
4
5
graph LR
A[拌料] --> B(成型)
B --> C(煎炸)
C --> D(红烧)
D --> E(勾芡)
三、操作步骤

  1. 将五花肉，山药、老豆腐、葱姜蒜、生抽、老抽、蚝油、料酒、鸡蛋依次放入碗中，用手沿着一个方向搅拌上劲，待拌料可以黏在手上，加入少量淀粉后继续搅拌摔打；

    wyD8SO.jpg

  2. 将拌料团成你喜欢的大小；

    wyDJ6e.jpg

  3. 热锅冷油，煎炸二十分钟；油热6分可入锅，全程小火；这一步的主要目的是定型，口味清淡的朋友也可以用沸水来操作；

    wyDGlD.jpg

  4. 将辣椒、葱姜蒜、八角炒香，再加入生抽、老抽、蚝油、料酒，添2/3锅水，加入肉丸子小火慢煮30分钟；

    wyDYOH.jpg

  5. 勾芡 ；待锅内汤汁不多时，盛出肉丸子，然后加入水淀粉熬制挂锅，然后将红烧汁浇在狮子头上；

    wygNnS.jpg

    wygwkj.jpg

四、多说一句

做狮子头比较耗时耗力，只能在周末享用。相比来说，我更喜欢老家的肉丸子。材料比较简单，猪肉、馒头渣、黑胡椒。团成丸子，再兑上半锅水，炖煮一个小时。味浓又不油腻！

发表于 更新于 分类于 Valine：

一、应用背景

在支付系统中，线程池在批量支付中有着十分重要的作用。随着业务量的增多，支付接口的处理能力越来越弱，甚至出现一个支付批次的处理时间超过5分钟的情况。这对于调用方来说，是无法容忍的。最简单粗暴的解决办法，便是增加线程数和阻塞队列容量。但至于怎么调整，才能做到既保证接口对支付请求的处理能力，又不浪费系统资源，好像只能凭感觉来了。Prometheus + Grafana 是目前比较流行的监控体系，在管理线程池方面的表现也十分出色。

二、线程池配置

采用 Spring 提供的 ThreadPoolTaskExecutor 来完成线程池的配置。根据业务需要，创建两个线程池：同步线程池和异步线程池。具体配置如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@Configuration
public class ExecutorPoolConfig {
    
    /**
    * 线程池参数配置
    */
    @Resource
    private ExecutorProperties executorProperties;
    
    private static final String ASYNC_EXECUTOR = "asyncExecutor";
    private static final String SYNC_EXECUTOR = "syncExecutor";
    
    /**
    * @return 同步线程池
    */
    @RefreshScope
    @Bean(SYNC_EXECUTOR)
    public ThreadPoolTaskExecutor serviceExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(executorProperties.getSync().getCorePoolSize());
        executor.setMaxPoolSize(executorProperties.getSync().getMaxPoolSize());
        executor.setQueueCapacity(executorProperties.getSync().getQueueCapacity());
        executor.setKeepAliveSeconds((int)executorProperties.getSync()
                                     .getKeepAlive().getSeconds());
        executor.setThreadNamePrefix("executor-sync-service-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(executorProperties.getSync()
                                            .getAwaitTerminationSeconds());
        return executor;
    }
    
    /**
     * @Async 注解默认走该线程池
     * @return 异步处理线程池
     */
    @RefreshScope
    @Bean(ASYNC_EXECUTOR)
	public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(executorProperties.getAsync().getCorePoolSize());
        executor.setMaxPoolSize(executorProperties.getAsync().getMaxPoolSize());
        executor.setQueueCapacity(executorProperties.getAsync().getQueueCapacity());
        executor.setKeepAliveSeconds((int)executorProperties.getAsync()
                                     .getKeepAlive().getSeconds());
        executor.setThreadNamePrefix("common-async-executor-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(executorProperties.getAsync()
                                            .getAwaitTerminationSeconds());
        return executor;
    }
}

三、监控指标

综合线程池的核心要素和生产业务的关键要素，提出以下几种监控指标:

  • 核心线程数;
  • 最大线程数；
  • 活跃线程数；
  • 当前线程数；
  • 队列中任务数；

  • 已完成任务数； 

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    import io.micrometer.core.instrument.Gauge;
    import io.micrometer.core.instrument.MeterRegistry;
    import io.micrometer.core.instrument.Metrics;

    @Component
    public class ExecutorMetricsSupport implements InitializingBean {
        
        /**
        * Prometheus 数据采集中心
        */
        @Resource
        private MeterRegistry meterRegistry;
        
        @Autowired
        @Qualifier(SYNC_EXECUTOR)
        private ThreadPoolTaskExecutor syncExecutor;
        
        @Autowired
        @Qualifier(ASYNC_EXECUTOR)
        private ThreadPoolTaskExecutor asyncExecutor;
        
        @Override
        public void afterPropertiesSet() throws Exception {
            initServiceExecutorMetrics(syncExecutor, "executor.sync");
            initServiceExecutorMetrics(asyncExecutor, "executor.async");
        }

        /**
         * 线程池metrics指标监控
         * @param serviceExecutor 线程池
         * @param namePrefix 指标名称前缀
         */
        private void initServiceExecutorMetrics(ThreadPoolTaskExecutor serviceExecutor, String namePrefix) {
            Gauge
                .builder(namePrefix.concat(".active"),
                         serviceExecutor, ThreadPoolTaskExecutor::getActiveCount)
                .register(meterRegistry);
            
            Gauge
                .builder(namePrefix.concat(".core"),
                         serviceExecutor, ThreadPoolTaskExecutor::getCorePoolSize)
                .register(meterRegistry);
            
            Gauge
                .builder(namePrefix.concat(".max"),
                          serviceExecutor, ThreadPoolTaskExecutor::getMaxPoolSize)
                .register(meterRegistry);
            
            Gauge
                .builder(namePrefix.concat(".pool"),
                         serviceExecutor, ThreadPoolTaskExecutor::getPoolSize)
                .register(meterRegistry);
            
            Gauge
                .builder(namePrefix.concat(".queue"), serviceExecutor,
                         executor -> executor.getThreadPoolExecutor().getQueue().size())
                .register(meterRegistry);
            
            Gauge
                .builder(namePrefix.concat(".completetask"), serviceExecutor,
                         executor -> executor.getThreadPoolExecutor().getCompletedTaskCount())
                .register(meterRegistry);
        }
    }

四、指标分析

通过 PSQL 将以上各项指标展示在 Grafana 中。 Prometheus 默认 15s 拉取一次数据，对于线程池这种波定性较大的指标，建议将拉取时间调整至 10s，以便灵活且准确地反应线程池的运行情况。

  • 异步线程池

dDNE9g.png

  • 同步线程池

dDN6vd.png

由监控结果可以作出以下分析：

  • 异步线程池的负担非常小，目前还没出现过队列积压的情况，可以适当减少核心线程和最大线程数；
  • 同步线程池配置较为妥当，峰值期间未启用非核心线程，队列积压任务量在合理区间，请求处理效率非常高。

发表于 更新于 分类于 Valine：

一、需求背景

本周负责做调账线上化，其中一个业务需求比较棘手：提交调账申请后，需在数据表记录对应的调账类型；而在调账复核页面，列表筛选项-调账类型，需支持多选。简单枚举如下:

1
2
3
4
5
6
7
8
9
10
11
public enum AdjustTypeEnum {
    /**
     * 调账类型枚举; 一次调账申请可能对应多种调账类型，在数据表中枚举值之间以英文 , 分隔
     */
    FINE(0, "应还罚息"),
    TOTAL_REPAYMENT(1, "应还总额"),
    FACT_REPAYMENT(2, "实还总额"),
    FACT_PAYMENT_DATE(3, "实还日期"),
    OVERDUE_DAYS(4, "逾期天数"),
    PAYMENT_STATUS(5, "还款状态");
}

假如调账申请A的调账类型包括应还罚息和应还总额，则数据表相应字段值为0,1。在复核页面，若调账类型复选项中包含应还罚息或应还总额，则需查询到该申请。此时，我们发现SQL内置函数FIND_IN_SET只支持单选项查询，而无法灵活拆分筛选条件和存储数据。因此，有必要调整一下数据的存储规则和查询规则。

二、位与运算的妙用

首先，我们需要进一步抽象：一方面，为了不浪费存储空间，科学地设计数据表，我们需要将目标数据列表整合存储在一个字段中；另一方面，数据库引擎需要灵活使用整合后的字段，以执行相应的查询语句。说白了，就是看起来像做了整合，实际上还是原始的目标数据列表。

事实证明，上面提出的方案根本满足不了业务需求，是一个非常差的设计。除了以逗号分隔枚举值的存储方式，还有一种更为科学的存储方式——二进制。简单做几个位与运算：

1
2
3
4
5
6
7
8
// 存储值: 3; 查询条件: 1
// 3 == 2 + 1
// 1 & 3 == 1
0001
&
0011
=
0001
1
2
3
4
5
6
7
8
// 存储值: 7; 查询条件: 1
// 7 == 1 + 2 + 4
// 1 & 7 == 1
0111
&
0001
=
0001
1
2
3
4
5
6
7
8
9
// 存储值: 7; 查询条件: 3
// 7 == 1 + 2 + 4
// 3 == 1 + 2
// 3 & 7 == 3
0111
&
0011
=
0011
1
2
3
4
5
6
7
8
9
// 存储值: 7; 查询条件: 15
// 7 == 1 + 2 + 4
// 15 == 1 + 2 + 4 + 8
// 15 & 7 == 7
0111
&
1111
=
0111

从以上位与运算实例，可以发现一个有趣的现象：若将枚举值设置为2的n次幂，则只需将目标数据列表加合，即可保证每一个目标数据对应到二进制中的一位。

此时，对于任何一个查询条件，只需要将其与整合值进行位与运算，即可得到匹配的值。

由此，我们调整枚举为：

1
2
3
4
5
6
7
8
9
10
11
public enum AdjustTypeEnum {
    /**
     * 调账类型枚举; 一次调账申请可能对应多种调账类型，在数据表中存储各值之和
     */
    FINE(1, "应还罚息"),
    TOTAL_REPAYMENT(2, "应还总额"),
    FACT_REPAYMENT(4, "实还总额"),
    FACT_PAYMENT_DATE(8, "实还日期"),
    OVERDUE_DAYS(16, "逾期天数"),
    PAYMENT_STATUS(32, "还款状态");
}

调整SQL为：

1
2
3
4
5
SELECT
*
FROM `boss-account`.acct_adjust_account_record
WHERE
adjust_type_sum & #{调账类型筛选条件值之和} > 0

发表于 更新于 分类于 Valine：

一、技术背景

实现发布订阅的中间件有许多，包括时下最热的Kafka、RabbitMQ、ActiveMQ，不太常见的Guava下的EventBus，以及Redis等。关于MQ之流的分析已经够多了，今天详细介绍下Redis的发布订阅机制及其实现。事实上，Redis的发布订阅机制在具体项目中的应用非常少，主要有两方面原因：其一，Redis的可靠性比较差，一旦出现断网等情况，则发布的消息便全部丢失；其二，Redis的消息处理方式是通过单线程循环遍历实现的，若存在大量的消息发布，则可能导致输出缓冲区膨胀，甚至服务崩溃。但对数据安全性和稳定性要求不高的场景来说，Redis不失为最佳的选择。

二、基本操作

Redis 的发布订阅模式包括普通订阅，普通订阅取消，模式订阅，模式订阅取消这四个场景。用命令实现如下：

  • 启动 Redis 服务（以 Windows 平台为例）

    1
    2
    3
    C:\Tools\Redis>redis-server.exe redis.windows.conf
    ...
    The server is now ready to accept connections on port 6379

  • 创建客户端client1，并以普通订阅渠道channel1

    1
    2
    3
    4
    5
    6
    redis-cli.exe -h 127.0.0.1 -p 6379
    127.0.0.1:6379> SUBSCRIBE channel1
    Reading messages... (press Ctrl-C to quit)
    1) "subscribe"
    2) "channel1"
    3) (integer) 1

  • 创建客户端client2，并普通订阅渠道channel2

    1
    2
    3
    4
    5
    6
    redis-cli.exe -h 127.0.0.1 -p 6379
    127.0.0.1:6379> SUBSCRIBE channel2
    Reading messages... (press Ctrl-C to quit)
    1) "subscribe"
    2) "channel2"
    3) (integer) 1

  • 创建客户端client3，并模式订阅渠道channel*

    1
    2
    3
    4
    5
    6
    redis-cli.exe -h 127.0.0.1 -p 6379
    127.0.0.1:6379> PSUBSCRIBE channel*
    Reading messages... (press Ctrl-C to quit)
    1) "psubscribe"
    2) "channel*"
    3) (integer) 1

  • 向渠道channel2发布一条消息

    1
    2
    3
    4
    redis-cli.exe -h 127.0.0.1 -p 6379
    127.0.0.1:6379> PUBLISH channel2 "msg from channel2"
    (integer) 2
    127.0.0.1:6379>

  • client2 、client3 接收到消息

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    # ----------client2---------
    1) "message"
    2) "channel2"
    3) "msg from channel2"

    # ----------client3---------
    1) "pmessage"
    2) "channel*"
    3) "channel2"
    4) "msg from channel2"

因此，我们可以猜测，消息发布者与消息订阅者之间是通过渠道连接的，包括精准匹配（普通订阅）和模糊匹配（模式订阅）。经过分析其结构设计，可表示为用例：

三、原理分析

Redis发布订阅的核心实现在pubsub.c文件。从头文件server.h中可以读取相关函数声明：

1
2
3
4
5
6
void subscribeCommand(client *c);      /* 普通订阅 */
void unsubscribeCommand(client *c);    /* 普通订阅取消 */
void psubscribeCommand(client *c);     /* 模式订阅 */
void punsubscribeCommand(client *c);   /* 模式订阅取消 */
void publishCommand(client *c);
void pubsubCommand(client *c);
  • 普通订阅模式：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
#define CLIENT_PUBSUB (1<<18)      /* Client is in Pub/Sub mode. */
/*-----------------------------------------------------------------------------
 * Pubsub commands implementation
 *----------------------------------------------------------------------------*/

void subscribeCommand(client *c) {
    int j;
	/* 分析client结构，c->argv[0]为命令本身，第2位开始为渠道参数 */
    for (j = 1; j < c->argc; j++)
        /* 订阅一个渠道 */
        pubsubSubscribeChannel(c,c->argv[j]);
    /* 设置客户端为发布订阅模式，此模式在processCommand中为校验值 */
    c->flags |= CLIENT_PUBSUB;
}

/* 一个客户端订阅一个渠道。若返回 1，则订阅成功；若返回 0，则该客户端已订阅该渠道 */
int pubsubSubscribeChannel(client *c, robj *channel) {
    dictEntry *de;
    list *clients = NULL;
    int retval = 0;

    /* 将该渠道添加到该客户端的pubsub_channels哈希表中； key: channel，value: NULL*/
    if (dictAdd(c->pubsub_channels,channel,NULL) == DICT_OK) {
        retval = 1;
        /* 引用自增 1 */
        incrRefCount(channel);
        /* 在服务端pubsub_channels中查找该channel */
        de = dictFind(server.pubsub_channels,channel);
        /* 若该channel不存在，则新建 */
        if (de == NULL) {
            clients = listCreate();
            /* 将该客户端添加到该服务端的pubsub_channels哈希表中; key: channel，value: client*/
            dictAdd(server.pubsub_channels,channel,clients);
             /* 引用自增 1 */
            incrRefCount(channel);
        } else {
            clients = dictGetVal(de);
        }
         /* 将客户端c记录在订阅客户端列表 */
        listAddNodeTail(clients,c);
    }
    /* 通知客户端订阅结果 */
    addReplyPubsubSubscribed(c,channel);
    return retval;
}

普通订阅模式主要做了两件事：将订阅渠道添加至客户端pubsub_channels哈希表中；将订阅客户端添加至服务端的pubsub_channels哈希表中。可如下表示：

发表于 更新于 分类于 Valine：

今天特别想吃我妈做的油馍，也就是大家常说的葱油饼。虽然出租屋的条件有限，但我打算尝试一次。

一、原料

小麦粉，小香葱，香油，花生油，食盐，佐料（看个人喜好选择添加）。

二、步骤

1. 和面

加入适量面粉（尽量选高筋粉），边用筷子搅拌边兑温水，待面结成絮状，手动和成面团。建议面团和软一点，保证很好的延展性。和好后，套上保鲜膜，醒面20分钟。

N1pDh9.jpg

2. 擀面

面醒好后，揉捏成条状。

N19cCj.jpg

葱油饼是多层叠加起来的，所以需要尽可能地将面擀薄。

N1C3Mq.jpg

3. 加料

将适量香油滴在面皮上，然后不断对折面皮，直至香油全部铺开。

N1PkY4.jpg

然后，均匀撒上葱花，食盐少量。

N1P4NF.jpg

4. 成型

将铺满葱花的面从一边角叠起。

N1ZtPK.jpg

将面擀平实。

N1VM7t.jpg

5. 煎饼

注意控制油温。家用燃气灶，小火慢热即可。

N1VaBn.jpg

N1VUns.jpg

N1VtXj.jpg

发表于 更新于 分类于 Valine：

一、线程的使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public static void main(String[] args) {
    Thread thread1 = new Thread(new Runnable() {
        @Override
        public void run() {
            System.out.println("执行线程1");
        }
    });
    thread1.start();

    Thread thread2 = new Thread(new Runnable() {
        @Override
        public void run() {
            System.out.println("执行线程2");
        }
    });
    thread2.run();
}

创建以上两个线程，执行结果：

1
2
执行线程1
执行线程2

线程1与线程2都执行了 run()，但线程1是通过 start() 执行的，线程2是直接调用 run() 执行的。

二、start() 与 run() 的区别

为了更精准地找到两种调用之间的区别，需要打印线程相关的信息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public static void main(String[] args) {
    Thread thread1 = new Thread(new Runnable() {
        @Override
        public void run() {
            Thread currentThread = Thread.currentThread();
            System.out.println("执行线程1; 线程名称：" + currentThread.getName());
        }
    });
    thread1.start();

    Thread thread2 = new Thread(new Runnable() {
        @Override
        public void run() {
            Thread currentThread = Thread.currentThread();
            System.out.println("执行线程2; 线程名称：" + currentThread.getName());
        }
    });
    thread2.run();
}

执行结果：

1
2
执行线程1; 线程名称：Thread-0
执行线程2; 线程名称：main

从执行结果来看，start() 调用，是通过开启一个新线程来执行 run() 的；而直接调用 run()，是主线程直接执行方法内容的。

此外，我们还可以做另一种尝试：多次执行 run() 方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public static void main(String[] args) {
    Thread thread1 = new Thread(new Runnable() {
        @Override
        public void run() {
            Thread currentThread = Thread.currentThread();
            System.out.println("执行线程1; 线程名称：" + currentThread.getName());
        }
    });
    thread1.run();
    thread1.run();

    Thread thread2 = new Thread(new Runnable() {
        @Override
        public void run() {
            Thread currentThread = Thread.currentThread();
            System.out.println("执行线程2; 线程名称：" + currentThread.getName());
        }
    });

    thread2.start();
    thread2.start();
}

执行结果：

1
2
3
4
5
6
执行线程1; 线程名称：main
执行线程1; 线程名称：main
执行线程2; 线程名称：Thread-1
Exception in thread "main" java.lang.IllegalThreadStateException
	at java.lang.Thread.start(Thread.java:708)
	at com.example.springbatchdemo.job.Test.main(Test.java:29)

从执行结果可知，可以多次调用 run()，而 start() 只能被执行一次，这是为什么呢？我们看下 start 方法说明：

1
2
3
4
5
6
7
8
9
10
public synchronized void start() {
    /**
    * This method is not invoked for the main method thread or "system"
    * group threads created/set up by the VM. Any new functionality added
    * to this method in the future may have to also be added to the VM.
    *
    * A zero status value corresponds to state "NEW".
    */
    if (threadStatus != 0)
        throw new IllegalThreadStateException();

调用 start()，首先需要判断当前线程的状态是否为 0 （NEW），如果不是，则抛异常。当第一次调用 start() 时，线程状态从 NEW （新建）变为 RUNNABLE（就绪），当第二次调用 start() 时，线程状态肯定不是 NEW。

三、总结

Thread 的 start() 与 run() 都可以完成任务执行，主要区别在：

  • 作用机制不同：直接调用 run()，相当于在当前线程中执行普通方法；调用 start()，是开启一个新的线程来执行 run()；
  • 执行速度不同：调用 run() 会立即执行任务，调用 start() 是将线程的状态改为就绪状态，等待空闲 CPU，不会立即执行；
  • 调用次数不同：run() 可以被重复调用，而 start() 只能被调用一次。