Java 多线程分批同步数据

 
已识乾坤大，犹怜草木青。
         ——旷怡亭口占

前言

最近接到一个任务，将mysql中的数据同步到elasticsearch中，要求异步执行，接口不必返回结果，接到请求后后台默默执行就行了。这种情况下其实单线程一页一页去读取写入就可以了，因为不必立刻返回请求结果后台执行5分钟还是十分钟只要能把数据加入到es中就可以了。

但是，身为一个优秀的程序员 怎么只能考虑功能的实现不考虑效率问题呢😎😎😎（时间允许的情况下考虑效率对已完成功能的代码做优化），这里就用到了多线程分批导入。注意这里一定要分批，不要所有线程一块去执行 要按批次。例如有100万数据，2000条执行一次，每批次五个线程去执行，等待每批次完成后再开启下一次，否则100万数据一上来500个线程同时去执行，假如线程池设置过小就会导致部分请求直接拒绝，最后实际执行不到500次，多线程也不是越多越好，线程数需要结合实际情况测试反复推理选择最合适的并发数，另一方面一台服务器也肯定不是就跑你一个程序，操作数据量太大的话，一上来你就并发几百个线程去执行也不太友好（数据库连接数也是个问题）。所以最后采用分批次并发一定数量去执行。

设计思路

• 多线程读取，写入（对应生成、消费）
• 分批次执行（每批次多个线程去执行）
• 一批次内所有线程执行完成后，再开启下一批次执行

描述： 先查询需要同步的数据量，根据总数据量和每次执行数据量（2000条一次）计算出总共需要执行多少次，通过总共执行多少次和每批次几个线程去执行，就可以知道当前完成任务需要跑多少批次（一批次多个线程）

代码实践

定义线程池

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

    // 设置核心线程数
    executor.setCorePoolSize(50);
    // 设置最大线程数
    executor.setMaxPoolSize(200);
    // 设置队列容量
    executor.setQueueCapacity(200);
    // 设置线程活跃时间（秒）
    executor.setKeepAliveSeconds(800);
    // 设置默认线程名称
    executor.setThreadNamePrefix("task-");
    // 设置拒绝策略
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    // 等待所有任务结束后再关闭线程池
    executor.setWaitForTasksToCompleteOnShutdown(true);
    return executor;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

业务代码实现

import org.elasticsearch.client.RestHighLevelClient;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

@Service
@Slf4j
public class BookInfoServiceImpl extends ServiceImpl<BookInfoMapper, BookInfo> implements BookInfoService {
    @Autowired
    private RestHighLevelClient client; //spring-data-es中的类，配置好es连接可以直接使用
    @Resource
    private TaskExecutor taskExecutor;

     /**
     * 一次同步多少条数据(分页大小)
     * 每批次执行大小和并发线程数根据实际情况修改
     * @author yh
     * @date 2022/8/11
     */
    private final Integer batchSize = 2000;
    /**
     * 每批次执行多少线程(几个线程去执行)
     *
     * @date 2022/8/11
     */
    private final Integer threadCount = 5;

    /**
     * 不同系统的换行符，es _bulk批量操作时需要拼装ndjson格式数据时用到
     * @date 2022/8/11
     */
    private String newLine = System.getProperty("line.separator");

    /**
     * 根据业务id同步数据
     * 

     *
     * @param libraryId 业务id
     * @author yh
     * @date 2022/8/11
     */
    @Async
    @Override
    public void syncLibraryBooks(Long libraryId) {
        if (null == libraryId) {
            return;
        }
        try {
            // 查询需要同步的数据量
            LambdaQueryWrapper<BookInfo> wrapper = new LambdaQueryWrapper<>();
            wrapper.eq(BookInfo::getLibraryId, libraryId)
                    .isNotNull(BookInfo::getMeta)
                    .isNotNull(BookInfo::getBookId);
            // 同步数据总数据量
            Integer dataCount = baseMapper.selectCount(wrapper);
            if (dataCount < 1) {
                log.error("同步数据：libraryId：{}，没有需要同步的数据, 结束", libraryId);
                return;
            }

            //---------------------开始同步数据------------
            // 数据操作索引
            String indexName = libraryId + IndexConst.SCHOOL_SUPPLEMENT_INDEX_SUFFIX;
            // 验证索引是否存在，不存在：创建，存在：删除后重新创建
            verifyIndexExist(indexName);

            // 当前需要执行多少次（根据当前的数据量和每次执行多少条数据计算得出多少次）
            int execNext = 0;

            if (dataCount % batchSize == 0) {
                execNext = dataCount / batchSize;
            } else {
                execNext = (dataCount / batchSize) + 1;
            }

            long startTime = System.currentTimeMillis();

            // 并发线程下执行的任务批次
            int taskNext = 0;
            if (execNext % threadCount == 0) {
                taskNext = execNext / threadCount;
            } else {
                taskNext = (execNext / threadCount) + 1;
            }
            log.info("start------开始执行同步数据，业务id: {}, 数据总量: {}, 每次请求数据量：{}， 需要发送请求次数: {}。 单次并发线程数: {}, 执行批次：{}",
                    libraryId, dataCount, batchSize, execNext, threadCount, taskNext);
            // 代表页数
            int pageIndex = 1;
            for (int i = 0; i < taskNext; i++) {
                // 本次需要开几个线程，默认五个，如果当前执行次数 乘 并发线程数 大于总共执行的次数，说明当前这次不够threadCount的数量，取余数就可以了
                int threadNumber = threadCount;
                if ((i + 1) * threadCount > execNext) {
                    threadNumber = execNext % threadCount;
                }
                log.info("当前执行次数：{}, 本次执行线程数：{}", i + 1, threadNumber);

                CountDownLatch count = new CountDownLatch(threadNumber);
                for (int j = 0; j < threadNumber; j++) {
                    AtomicInteger currPage = new AtomicInteger(pageIndex);
                    taskExecutor.execute(() -> {
                        try {
                            // 生产
                            List<BookInfo> bookInfoList = producer(libraryId, currPage.get());
                            // 消费
                            consumer(bookInfoList, indexName);
                            log.info("同步成功：第 {} 页, 数据量: {}条", currPage.get(), bookInfoList.size());
                        } finally {
                            count.countDown();
                        }
                    });
                    pageIndex++;
                }
                // 阻塞当前执行任务中的所有线程，等待本次线程全部执行完再开启下一次
                count.await();
            }

            long endTime = System.currentTimeMillis();
            log.info("end------成功：执行同步数据结束，业务id: {}，数据量: {},执行时间：{}毫秒", libraryId, dataCount, endTime - startTime);
        } catch (Exception e) {
            log.error("同步数据错误：libraryId：{}，error content: {}", libraryId, e.getMessage());
        }
    }


    /**
     * 生产数据
     *
     * @param libraryId  业务id
     * @param pageNumber 当前页
     * @return List
     * @author yh
     * @date 2022/8/11
     */
    private List<BookInfo> producer(Long libraryId, int pageNumber) {
        log.info("同步页数：{}", pageNumber);
        LambdaQueryWrapper<BookInfo> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(BookInfo::getLibraryId, libraryId)
                .isNotNull(BookInfo::getMeta)
                .isNotNull(BookInfo::getBookId);
        //这里需要用到哪些字段就查询哪些字段，因为同步一般数据量大，所以不要查询多余的冗余字段
        wrapper.select(BookInfo::getMeta, BookInfo::getBookId);

        Page<BookInfo> page = new Page<>(pageNumber, batchSize, false);
        Page<BookInfo> bookInfoPage = baseMapper.selectPage(page, wrapper);
        return bookInfoPage.getRecords();
    }

     /**
     * 消费数据
     *
     * @param bookInfoList 图书数据
     * @param indexName    索引名称
     * @author yh
     * @date 2022/8/11
     */
    private void consumer(List<BookInfo> bookInfoList, String indexName) {
        // 这里可以捕获一下异常，保存失败时把当前批次数据的id记录下来
        StringBuilder sb = new StringBuilder();

        for (BookInfo info : bookInfoList) {
            String meta = info.getMeta();
            if (StringUtils.isNotBlank(meta)) {
                String bookId = info.getBookId();
                
                // 具体数据格式，查看es bulk请求
                sb.append("{\"index\": {\"_index\": \"")
                        .append(indexName)
                        .append("\", \"_type\": \"")
                        .append(IndexConst.DEFAULT_TYPE_NAME)
                        .append("\", \"_id\": \"")
                        .append(bookId)
                        .append("\"}}")
                        .append(newLine);

                //这里meta是es中doc对应的json字符串
                sb.append(meta).append(newLine); //ndjson格式以换行符分割
            }
        }
        String url = RestTemplateUtil.BULK_URL.replace("{esIndex}", indexName);
        RestTemplateUtil.postForNdjson(url, sb.toString());
    }


     /**
     * 验证es索引是否存在，不存在就创建，存在删除重新创建
     *
     * @param indexName indexName
     * @author yh
     * @date 2022/8/5
     */
    private void verifyIndexExist(String indexName) throws IOException {
        Request request = new Request(HttpHead.METHOD_NAME, indexName);
        Response response = client.getLowLevelClient().performRequest(request);

        // 404 索引不存在，创建
        // 注意：这里需要设置RestTemplate 404异常不要拦截，否则404会直接抛异常
        if (404 == response.getStatusLine().getStatusCode()) {
            log.info("索引: {} 不存在 开始创建", indexName);
            String url = RestTemplateUtil.CREATE_INDEX.replace("{esIndex}", indexName);
            //ConfigConst.CREATE_INDEX_ENTITY：就是创建索引字段映射关系，这里就不贴代码了
            RestTemplateUtil.putForVoid(url, ConfigConst.CREATE_INDEX_ENTITY);
            return;
        }
        // 存在删除，重新创建
        log.info("索引: {} 存在 删除重新创建", indexName);
        String deleteIndex = RestTemplateUtil.DELETE_INDEX.replace("{esIndex}", indexName);
        RestTemplateUtil.deleteForVoid(deleteIndex);
        String url = RestTemplateUtil.CREATE_INDEX.replace("{esIndex}", indexName);
        RestTemplateUtil.putForVoid(url, ConfigConst.CREATE_INDEX_ENTITY);
    }
}
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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210

RestTemplateUtil

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.util.MultiValueMap;
import org.springframework.web.client.RestTemplate;

import javax.annotation.PostConstruct;
import java.util.Base64;

/**
 * @author: yh
 * @Description: restTemplate工具类
 * @date: 2022/8/4 15:43
 */
@Component
public class RestTemplateUtil {
    @Value("${spring.es.ip}")
    private String ip;

    @Value("${spring.es.prot}")
    private Integer port;

    @Value("${spring.es.username}")
    private String userName;

    @Value("${spring.es.password}")
    private String passWord;

    /**
     * 认证的Authorization
     */
    private static String authentication;
    /**
     * 查询ES POST-url
     */
    public static String SEARCH_INDEX_URL = null;
    /**
     * 多个搜索API-url
     */
    public static String MSEARCH_INDEX_URL = null;

    /**
     * 查询文档-url
     */
    public static String GET_DOC_ID = null;

    /**
     * 新建文档-url
     */
    public static String CREATE_DOC = null;

    /**
     * 更新文档 POST-url
     */
    public static String UPDATE_INDEX_DOC = null;

    /**
     * 按查询更新API-url
     */
    public static String UPDATE_BY_QUERY = null;

    /**
     * 创建ES索引，PUT请求-url
     */
    public static String CREATE_INDEX = null;

    /**
     * 批量增删改API  POST-url
     */
    public static String BULK_URL = null;

    /**
     * 根据文档id查询文档，也可以判断文档是否存在-url
     */
    public static String QUERY_INDEX_DOC = null;

    /**
     * Multi get (mget) API-url
     */
    public static String INDEX_MGET = null;

    /**
     * 删除索引（危险） delete请求
     */
    public static String DELETE_INDEX = null;

    @PostConstruct
    public void initProperty() {
        //{esIndex}：操作索引名称，{id}：操作文档id
        SEARCH_INDEX_URL = "http://" + ip + ":" + port + "/{esIndex}/_search";
        MSEARCH_INDEX_URL = "http://" + ip + ":" + port + "/{esIndex}/_msearch";
        GET_DOC_ID = "http://" + ip + ":" + port + "/{esIndex}/_doc/{id}";
        CREATE_DOC = "http://" + ip + ":" + port + "/{esIndex}/_create/{id}";
        UPDATE_INDEX_DOC = "http://" + ip + ":" + port + "/{esIndex}/_doc/{id}/_update";
        UPDATE_BY_QUERY = "http://" + ip + ":" + port + "/{esIndex}/_update_by_query";
        CREATE_INDEX = "http://" + ip + ":" + port + "/{esIndex}";
        BULK_URL = "http://" + ip + ":" + port + "/{esIndex}/_bulk";
        QUERY_INDEX_DOC = "http://" + ip + ":" + port + "/{esIndex}/_doc/{id}";
        INDEX_MGET = "http://" + ip + ":" + port + "/{esIndex}/_mget";
        DELETE_INDEX = "http://" + ip + ":" + port + "/{esIndex}";

        authentication = "Basic " + Base64.getEncoder().encodeToString((userName + ":" + passWord).getBytes());
    }

    private static RestTemplate restTemplate;

    @Autowired
    public void setRestTemplate(RestTemplate restTemplate) {
        RestTemplateUtil.restTemplate = restTemplate;
    }

    /**
     * POST请求
     *
     * @param url     请求路径
     * @param jsonStr 参数
     * @return String
     * @author yh
     * @date 2022/8/4
     */
    public static String postForObject(String url, String jsonStr) {
        //设置header信息
        HttpHeaders requestHeaders = new HttpHeaders();
        requestHeaders.setContentType(MediaType.APPLICATION_JSON);
        requestHeaders.set("authorization", "Basic " + authentication);

        HttpEntity<String> requestEntity = new HttpEntity<>(jsonStr, requestHeaders);

        return restTemplate.postForObject(url, requestEntity, String.class);
    }

    /**
     * ndjson格式数据请求
     *
     * @param ndjson 一种数据格式
     * @param url    url
     * @return String
     * @author yh
     * @date 2022/8/5
     */
    public static String postForNdjson(String url, String ndjson) {
        if (StringUtils.isBlank(ndjson)) {
            return "{}";
        }
        HttpHeaders headers = new HttpHeaders();
        headers.add("Content-Type", "application/x-ndjson");
        headers.set("authorization", "Basic " + authentication);
        HttpEntity<String> request = new HttpEntity<>(ndjson, headers);

        return restTemplate.postForObject(url, request, String.class);
    }

    /**
     * get查询es
     *
     * @param url 请求路径
     * @return String
     * @author yh
     * @date 2022/8/5
     */
    public static String getForString(String url) {
        if (StringUtils.isBlank(url)) {
            return "{}";
        }
        HttpHeaders headers = new HttpHeaders();
        headers.set("authorization", "Basic " + authentication);
        HttpEntity<MultiValueMap<String, Object>> request = new HttpEntity(null, headers);

        return restTemplate.exchange(url, HttpMethod.GET, request, String.class).getBody();
//        return restTemplate.getForObject(url, String.class);
    }

    /**
     * PUT请求
     *
     * @param url     请求路径
     * @param jsonStr body参数
     * @author yh
     * @date 2022/8/5
     */
    public static void putForVoid(String url, String jsonStr) {
        //设置header信息
        HttpHeaders requestHeaders = new HttpHeaders();
        requestHeaders.setContentType(MediaType.APPLICATION_JSON);
        requestHeaders.set("authorization", "Basic " + authentication);

        HttpEntity<String> requestEntity = new HttpEntity<>(jsonStr, requestHeaders);
        restTemplate.exchange(url, HttpMethod.PUT, requestEntity, String.class);
//        restTemplate.put(url, JSONObject.parse(json));
    }

    /**
     * delete请求
     *
     * @param url 请求路径
     * @author yh
     * @date 2022/8/11
     */
    public static void deleteForVoid(String url) {
        restTemplate.delete(url);
    }
}

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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208

yml配置

spring:
  es:
    ip: 127.0.0.1
    prot: 9200
    username: elastic
    password: 123456
1
2
3
4
5
6

执行结果：

最后，不建议使用原生ES API去进行操作，毕竟需要在代码中拼接一些JSON字符串，建议按照ORM对象关系映射的方式去操作数据，例如结合spring-data-elasticsearch去操作。😋😋😋

以后有新的优化会持续更新🐒🐒🐒