使用多线程查询百万条用户数据将汉字转化成拼音

现在有一个需求：用户表里面有将近200万条数据，查询时需要按用户名字的汉语拼音按从a-z排序。有两种解决方案：1.查询时使用数据库自带的CONVERT()函数进行转化，按拼音首字母排序;2.新加一个拼音字段（spell_name），用户注册时，将用户名字的汉语拼音也一并插入数据库。权衡之后我采用了第二种，因为用户量还会持续增长，使用数据库自带的函数会拖慢查询速度，索引也会失效，如果采用第二种加一个拼音字段，就涉及到对原来老数据拼音字段的一个初始化，数据量比较多，使用多线程，在这里记录一下。

1.使用jpinyin和emoji-java将汉字转化成拼音

引入相关jar

<!--汉字转拼音jar-->
		<dependency>
			<groupId>com.github.stuxuhai</groupId>
			<artifactId>jpinyin</artifactId>
			<version>1.0</version>
		</dependency>
		<!--java操作emoji的jar-->
		<dependency>
			<groupId>com.vdurmont</groupId>
			<artifactId>emoji-java</artifactId>
			<version>4.0.0</version>
		</dependency>
复制代码

emoji表情处理 工具 类

public class EmojiDealUtil extends EmojiParser {
    /**
     * 获取非表情字符串
     * @param input
     * @return
     */
    public static String getNonEmojiString(String input) {
        int prev = 0;
        StringBuilder sb = new StringBuilder();
        List<UnicodeCandidate> replacements = getUnicodeCandidates(input);
        for (UnicodeCandidate candidate : replacements) {
            sb.append(input.substring(prev, candidate.getEmojiStartIndex()));
            prev = candidate.getFitzpatrickEndIndex();
        }
        return sb.append(input.substring(prev)).toString();
    }

    /**
     * 获取表情字符串
     * @param input
     * @return
     */
    public static String getEmojiUnicodeString(String input){
        EmojiTransformer  transformer = new EmojiTransformer() {
            public String transform(UnicodeCandidate unicodeCandidate) {
                return unicodeCandidate.getEmoji().getHtmlHexadecimal();
            }
        };
        StringBuilder sb = new StringBuilder();
        List<UnicodeCandidate> replacements = getUnicodeCandidates(input);
        for (UnicodeCandidate candidate : replacements) {
            sb.append(transformer.transform(candidate));
        }
        return  parseToUnicode(sb.toString());
    }

    public static String getUnicode(String source){
        String returnUniCode=null;
        String uniCodeTemp=null;
        for(int i=0;i<source.length();i++){
            uniCodeTemp = "\\u"+Integer.toHexString((int)source.charAt(i));
            returnUniCode=returnUniCode==null?uniCodeTemp:returnUniCode+uniCodeTemp;
        }
        return returnUniCode;
    }
}
复制代码

汉字转换成拼音的工具类

public class ChineseToPinYinUtil {

    /**
     * 转换为不带音调的拼音字符串
     * @param pinYinStr 需转换的汉字
     * @return 拼音字符串
     */
    public static String changeToTonePinYin(String pinYinStr) {
        String tempStr = null;
        try {
            tempStr = PinyinHelper.convertToPinyinString(pinYinStr, " ", PinyinFormat.WITHOUT_TONE);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return tempStr;
    }
}
复制代码

拼音转换不做为本文点，网上搜下有很多库和代码解决，主要说下面的多线程处理。

2.使用多线程查询并更新数据库

创建初始容量为5的线程池，每次每个线程查询500条记录并处理，加个同步锁，分配给每个线程它查询的起始记录，查询出记录之后调用上面的汉字转换成拼音方法处理，完毕之后更新到数据库。

2.1接收请求方法

//每个线程每次查询的条数
    private static final Integer LIMIT = 500;
    //起的线程数
    private static final Integer THREAD_NUM = 5;
    ThreadPoolExecutor pool = new ThreadPoolExecutor(THREAD_NUM, THREAD_NUM*2,0,TimeUnit.SECONDS,new LinkedBlockingQueue<>(100));
    
    @GetMapping("/chineseToSpellName")
    public void execute(){
        //计数器,一次转换只能一个请求调，不然会出错
        int count = 0;
        logger.info("trans start");
        //查询总记录数
        int total = userService.getTotalCount2();
        logger.info("total num:{}",total);
        int num = total/(LIMIT*THREAD_NUM) + 1;
        logger.info("要经过的轮数：{}",num);
        for(int j=0;j<num;j++){
            //起 THREAD_NUM 个线程并行查询更新库，加锁
            for(int i=0;i<THREAD_NUM;i++){
                synchronized(ChineseToPinYinController.class){
                    int start = count*LIMIT;
                    count++;
                    pool.submit(new TransTask(start,LIMIT));
                }
            }
        }
    }
复制代码

2.2多线程处理业务方法

class TransTask implements Runnable{
        int start;
        int limit;
        public TransTask(int start, int limit) {
            this.start = start;
            this.limit = limit;
        }

        @Override
        public void run() {
            //查询记录并更新数据库
            List<User> userList =  userService.getList2(start,limit);
            logger.info("更新记录起始位置：{}--{}",start,limit);
            if(!CollectionUtils.isEmpty(userList)){
                userList.stream().forEach(u -> {
                    u.setSpellName(ChineseToPinYinUtil.changeToTonePinYin(EmojiDealUtil.getNonEmojiString(u.getName())).trim());
                    userService.updateUser2(u);
                }
             );
            }
        }
    }
复制代码

3.不使用传统的limit分页查询数据

userService.getList2(start,num)是根据起始位置和查询条数查询记录，以前我们写的分页查询一般是这样写的：select * from table limit start,num（如：select * from user limit 0,20）。这种查询在数据量小时没有问题，但是数据量大的时候查询会非常慢，因为它走的不是索引，而是全表扫描，数据量越大，越到后面速度越慢。对于id是自增长的查询可以采用另一种查询方式，select * from table where id>start limit num(如：select * from user where id>1000 limit 20)，从指定id查询num条记录。这种查询即使到百万级数据量，查询速度也不会明显变慢，因为走的是主键索引，而不是全表扫描。

4.优化后记

代码写完之后在实际使用中，数据初始化到70多万条的时候，数据库连接数开的太多，将数据库里面的全部占满了，考虑再次优化，采用分段的方式，传入两个参数，初始化记录和初始化条数。比如第一次初始化0-10万条记录，第二次初始化10到20万条记录，依次类推，这样的好处是可以人工干预，即使出错，也可以只运行出错的这部分区间数据，代码如下：

//每个线程每次查询的条数
    private static final Integer LIMIT = 500;
    //起的线程数
    private static final Integer THREAD_NUM = 5;
    ThreadPoolExecutor pool = new ThreadPoolExecutor(THREAD_NUM,Integer.MAX_VALUE,0,TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(10));

    @GetMapping("/chineseToSpellName")
    public void execute(@RequestParam("startId") Integer startId,@RequestParam("total") Integer total){
        logger.info("trans start");
        int num = total/(LIMIT*THREAD_NUM) + 1;
        logger.info("要经过的轮数：{}",num);
        for(int j=0;j<num;j++){
            //起 THREAD_NUM 个线程并行查询更新库，加锁
            for(int i=0;i<THREAD_NUM;i++){
                synchronized(ChineseToPinYinController.class){
                    pool.submit(new TransTask(startId,LIMIT));
                    startId+=LIMIT;
                }
            }
        }
    }

    class TransTask implements Runnable{
        int start;
        int limit;
        public TransTask(int start, int limit) {
            this.start = start;
            this.limit = limit;
        }

        @Override
        public void run() {
            //查询记录并更新数据库
            List<User> userList =  userService.getList2(start,limit);
            logger.info("更新记录起始位置：{}--{}",start,limit);
            if(!CollectionUtils.isEmpty(userList)){
                userList.stream().forEach(u -> {
                    u.setSpellName(ChineseToPinYinUtil.changeToTonePinYin(EmojiDealUtil.getNonEmojiString(u.getName())).trim());
                    userService.updateUser2(u);
                }
             );
            }
        }
    }
复制代码