задний план
Понимание многопоточности не очень глубокое, и возможностей использовать многопоточный код в работе не так много.Не так давно я столкнулся со сценарием использования.После реализации его через кодинг у меня более глубокое понимание многопоточности нарезка и применение. Сценарий следующий: Необходимо отправить исследование продукта пользователям Коллега по эксплуатации принес файл Excel и попросил отправить текстовые сообщения исследования примерно на 60 000 номеров мобильных телефонов в Excel.
Проще всего сделать цикл, а затем отправить его последовательно в одном потоке, но основная проблема заключается в том, что время отклика интерфейса для отправки SMS-сообщений операторам SMS велико. занимает 60 000 * 0,1 секунды для однопоточной передачи 6000 секунд. Очевидно, что это время неприемлемо, мы не можем оптимизировать отправляющий интерфейс системы оператора, только наращивая собственные возможности отправки и обработки, мы можем выполнить задачу в кратчайшие сроки.
Массовая рассылка СМС
Чтение информации в Excel
зависимости пакета
Код класса инструмента, Maven представляет следующие два пакета
<dependency>
<groupId>org.apache.poi</groupId>
<artifactId>poi-ooxml</artifactId>
<version>3.17</version>
</dependency>
<dependency>
<groupId>org.apache.xmlbeans</groupId>
<artifactId>xmlbeans</artifactId>
<version>2.6.0</version>
</dependency>
Код инструмента для чтения Excel
/**
* 读取Excel的文件信息
*
* @param fileName
*/
public static void readFromExcel(String fileName) {
InputStream is = null;
try {
is = new FileInputStream(fileName);
XSSFWorkbook workbook = new XSSFWorkbook(is);
XSSFSheet sheet = workbook.getSheetAt(0);
int num = 0;
// 循环行Row
for (int rowNum = 0, lastNum = sheet.getLastRowNum(); rowNum <= lastNum; rowNum++) {
XSSFRow row = sheet.getRow(rowNum);
String phoneNumber = getStringValueFromCell(row.getCell(0)).trim();
phoneList.add(phoneNumber);
}
System.out.println(num);
} catch (FileNotFoundException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 读取Excel里面Cell内容
*
* @param cell
* @return
*/
private static String getStringValueFromCell(XSSFCell cell) {
if (cell == null) {
return null;
}
// 单元格内的时间格式
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
// 单元格内的数字类型
DecimalFormat decimalFormat = new DecimalFormat("#.#####");
// 单元格默认为空
String cellValue = "";
// 按类型读取
if (cell.getCellType() == XSSFCell.CELL_TYPE_STRING) {
cellValue = cell.getStringCellValue();
} else if (cell.getCellType() == XSSFCell.CELL_TYPE_NUMERIC) {
// 日期转为时间形式
if (DateUtil.isCellDateFormatted(cell)) {
double d = cell.getNumericCellValue();
Date date = DateUtil.getJavaDate(d);
cellValue = dateFormat.format(date);
} else {
// 其他转为数字
cellValue = decimalFormat.format((cell.getNumericCellValue()));
}
} else if (cell.getCellType() == XSSFCell.CELL_TYPE_BLANK) {
cellValue = "";
} else if (cell.getCellType() == XSSFCell.CELL_TYPE_BOOLEAN) {
cellValue = String.valueOf(cell.getBooleanCellValue());
} else if (cell.getCellType() == XSSFCell.CELL_TYPE_ERROR) {
cellValue = "";
} else if (cell.getCellType() == XSSFCell.CELL_TYPE_FORMULA) {
cellValue = cell.getCellFormula().toString();
}
return cellValue;
}  
Смоделируйте способ отправки оператором СМС
/**
* 外部接口耗时长，通过多线程增强
*
* @param userPhone
*/
public void sendMsgToPhone(String userPhone) {
try {
Thread.sleep(SEND_COST_TIME);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("send message to : " + userPhone);
}
Многопоточные текстовые сообщения
Простая однопоточная отправка
/**
* 单线程发送
*
* @param phoneList
* @return
*/
private long singleThread(List<String> phoneList) {
long start = System.currentTimeMillis();
/*// 直接主线程执行
for (String phoneNumber : phoneList) {
threadOperation.sendMsgToPhone(phoneNumber);
}*/
SendMsgExtendThread smet = threadOperation.new SendMsgExtendThread(phoneList);
smet.start();
long totalTime = System.currentTimeMillis() - start;
System.out.println("单线程发送总时间：" + totalTime);
return totalTime;
}
Для сценария массовых текстовых сообщений, если один поток используется для отправки всех 1000 номеров, это займет около 103132 мс, что показывает низкую эффективность и занимает много времени.
Одним из основных моментов многопоточной отправки SMS является разделение всех номеров мобильных телефонов на несколько групп и назначение их каждому потоку для выполнения.
Пример двух потоков
/**
* 两个线程发送
*
* @param phoneList
* @return
*/
private long twoThreads(List<String> phoneList) {
long start = System.currentTimeMillis();
List<String> list1 = phoneList.subList(0, phoneList.size() / 2);
List<String> list2 = phoneList.subList(phoneList.size() / 2, phoneList.size());
SendMsgExtendThread smet = threadOperation.new SendMsgExtendThread(list1);
smet.start();
SendMsgExtendThread smet1 = threadOperation.new SendMsgExtendThread(list2);
smet1.start();
return 0;
}
Другой способ группировки данных
/**
* 另外一种分配方式
*
* @param phoneList
*/
private void otherThread(List<String> phoneList) {
for (int threadNo = 0; threadNo < 10; threadNo++) {
int numbersPerThread = 10;
List<String> list = phoneList.subList(threadNo * numbersPerThread, (threadNo * numbersPerThread) + 10);
SendMsgExtendThread smet = threadOperation.new SendMsgExtendThread(list);
smet.start();
if (list.size() < numbersPerThread) {
break;
}
}
}
отправка пула потоков
/**
* 线程池发送
*
* @param phoneList
* @return
*/
private void threadPool(List<String> phoneList) {
for (int threadNo = 0; threadNo < THREAD_POOL_SIZE; threadNo++) {
int numbersPerThread = 10;
List<String> list = phoneList.subList(threadNo * numbersPerThread, (threadNo * numbersPerThread) + 10);
threadOperation.executorService.execute(threadOperation.new SendMsgExtendThread(list));
}
threadOperation.executorService.shutdown();
}
Отправить с помощью Callable
/**
* 多线程发送
*
* @param phoneList
* @return
*/
private void multiThreadSend(List<String> phoneList) {
List<Future<Long>> futures = new ArrayList<>();
for (int threadNo = 0; threadNo < THREAD_POOL_SIZE; threadNo++) {
int numbersPerThread = 100;
List<String> list = phoneList.subList(threadNo * numbersPerThread, (threadNo * numbersPerThread) + 100);
Future<Long> future = threadOperation.executorService.submit(threadOperation.new SendMsgImplCallable(list, String.valueOf(threadNo)));
futures.add(future);
}
for (Future<Long> future : futures) {
try {
System.out.println(future.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
threadOperation.executorService.shutdown();
}
Используя многопоточную отправку, задача отправки делится, а затем назначается каждому потоку для выполнения.Для завершения выполнения требуется 10266 мс.Видно, что эффективность выполнения значительно повышается, а время потребления значительно сокращается.
полный код
package com.lingyejun.tick.authenticator;
import org.apache.poi.ss.usermodel.DateUtil;
import org.apache.poi.xssf.usermodel.XSSFCell;
import org.apache.poi.xssf.usermodel.XSSFRow;
import org.apache.poi.xssf.usermodel.XSSFSheet;
import org.apache.poi.xssf.usermodel.XSSFWorkbook;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.text.DecimalFormat;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.*;
public class ThreadOperation {
// 发短信的同步等待时间
private static final long SEND_COST_TIME = 100L;
// 手机号文件
private static final String FILE_NAME = "/Users/lingye/Downloads/phone_number.xlsx";
// 手机号列表
private static List<String> phoneList = new ArrayList<>();
// 单例对象
private static volatile ThreadOperation threadOperation;
// 线程个数
private static final int THREAD_POOL_SIZE = 10;
// 初始化线程池
private ExecutorService executorService = new ThreadPoolExecutor(THREAD_POOL_SIZE, THREAD_POOL_SIZE,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
public ThreadOperation() {
// 从本地文件中读取手机号码
readFromExcel(FILE_NAME);
}
public static void main(String[] args) {
ThreadOperation threadOperation = getInstance();
//threadOperation.singleThread(phoneList);
threadOperation.multiThreadSend(phoneList);
}
/**
* 单例获取对象
*
* @return
*/
public static ThreadOperation getInstance() {
if (threadOperation == null) {
synchronized (ThreadOperation.class) {
if (threadOperation == null) {
threadOperation = new ThreadOperation();
}
}
}
return threadOperation;
}
/**
* 读取Excel的文件信息
*
* @param fileName
*/
public static void readFromExcel(String fileName) {
InputStream is = null;
try {
is = new FileInputStream(fileName);
XSSFWorkbook workbook = new XSSFWorkbook(is);
XSSFSheet sheet = workbook.getSheetAt(0);
int num = 0;
// 循环行Row
for (int rowNum = 0, lastNum = sheet.getLastRowNum(); rowNum <= lastNum; rowNum++) {
XSSFRow row = sheet.getRow(rowNum);
String phoneNumber = getStringValueFromCell(row.getCell(0)).trim();
phoneList.add(phoneNumber);
}
System.out.println(num);
} catch (FileNotFoundException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 读取Excel里面Cell内容
*
* @param cell
* @return
*/
private static String getStringValueFromCell(XSSFCell cell) {
if (cell == null) {
return null;
}
// 单元格内的时间格式
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
// 单元格内的数字类型
DecimalFormat decimalFormat = new DecimalFormat("#.#####");
// 单元格默认为空
String cellValue = "";
// 按类型读取
if (cell.getCellType() == XSSFCell.CELL_TYPE_STRING) {
cellValue = cell.getStringCellValue();
} else if (cell.getCellType() == XSSFCell.CELL_TYPE_NUMERIC) {
// 日期转为时间形式
if (DateUtil.isCellDateFormatted(cell)) {
double d = cell.getNumericCellValue();
Date date = DateUtil.getJavaDate(d);
cellValue = dateFormat.format(date);
} else {
// 其他转为数字
cellValue = decimalFormat.format((cell.getNumericCellValue()));
}
} else if (cell.getCellType() == XSSFCell.CELL_TYPE_BLANK) {
cellValue = "";
} else if (cell.getCellType() == XSSFCell.CELL_TYPE_BOOLEAN) {
cellValue = String.valueOf(cell.getBooleanCellValue());
} else if (cell.getCellType() == XSSFCell.CELL_TYPE_ERROR) {
cellValue = "";
} else if (cell.getCellType() == XSSFCell.CELL_TYPE_FORMULA) {
cellValue = cell.getCellFormula().toString();
}
return cellValue;
}
/**
* 外部接口耗时长，通过多线程增强
*
* @param userPhone
*/
public void sendMsgToPhone(String userPhone) {
try {
Thread.sleep(SEND_COST_TIME);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("send message to : " + userPhone);
}
/**
* 单线程发送
*
* @param phoneList
* @return
*/
private long singleThread(List<String> phoneList) {
long start = System.currentTimeMillis();
/*// 直接主线程执行
for (String phoneNumber : phoneList) {
threadOperation.sendMsgToPhone(phoneNumber);
}*/
SendMsgExtendThread smet = threadOperation.new SendMsgExtendThread(phoneList);
smet.start();
long totalTime = System.currentTimeMillis() - start;
System.out.println("单线程发送总时间：" + totalTime);
return totalTime;
}
/**
* 另外一种分配方式
*
* @param phoneList
*/
private void otherThread(List<String> phoneList) {
for (int threadNo = 0; threadNo < 10; threadNo++) {
int numbersPerThread = 10;
List<String> list = phoneList.subList(threadNo * numbersPerThread, (threadNo * numbersPerThread) + 10);
SendMsgExtendThread smet = threadOperation.new SendMsgExtendThread(list);
smet.start();
if (list.size() < numbersPerThread) {
break;
}
}
}
/**
* 两个线程发送
*
* @param phoneList
* @return
*/
private long twoThreads(List<String> phoneList) {
long start = System.currentTimeMillis();
List<String> list1 = phoneList.subList(0, phoneList.size() / 2);
List<String> list2 = phoneList.subList(phoneList.size() / 2, phoneList.size());
SendMsgExtendThread smet = threadOperation.new SendMsgExtendThread(list1);
smet.start();
SendMsgExtendThread smet1 = threadOperation.new SendMsgExtendThread(list2);
smet1.start();
return 0;
}
/**
* 线程池发送
*
* @param phoneList
* @return
*/
private void threadPool(List<String> phoneList) {
for (int threadNo = 0; threadNo < THREAD_POOL_SIZE; threadNo++) {
int numbersPerThread = 10;
List<String> list = phoneList.subList(threadNo * numbersPerThread, (threadNo * numbersPerThread) + 10);
threadOperation.executorService.execute(threadOperation.new SendMsgExtendThread(list));
}
threadOperation.executorService.shutdown();
}
/**
* 多线程发送
*
* @param phoneList
* @return
*/
private void multiThreadSend(List<String> phoneList) {
List<Future<Long>> futures = new ArrayList<>();
for (int threadNo = 0; threadNo < THREAD_POOL_SIZE; threadNo++) {
int numbersPerThread = 100;
List<String> list = phoneList.subList(threadNo * numbersPerThread, (threadNo * numbersPerThread) + 100);
Future<Long> future = threadOperation.executorService.submit(threadOperation.new SendMsgImplCallable(list, String.valueOf(threadNo)));
futures.add(future);
}
for (Future<Long> future : futures) {
try {
System.out.println(future.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
threadOperation.executorService.shutdown();
}
public class SendMsgExtendThread extends Thread {
private List<String> numberListByThread;
public SendMsgExtendThread(List<String> numberList) {
numberListByThread = numberList;
}
@Override
public void run() {
long startTime = System.currentTimeMillis();
for (int i = 0; i < numberListByThread.size(); i++) {
System.out.print("no." + (i + 1));
sendMsgToPhone(numberListByThread.get(i));
}
System.out.println("== single thread send " + numberListByThread.size() + "execute time:" + (System.currentTimeMillis() - startTime) + " ms");
}
}
public class SendMsgImplCallable implements Callable<Long> {
private List<String> numberListByThread;
private String threadName;
public SendMsgImplCallable(List<String> numberList, String threadName) {
numberListByThread = numberList;
this.threadName = threadName;
}
@Override
public Long call() throws Exception {
Long startMills = System.currentTimeMillis();
for (String number : numberListByThread) {
sendMsgToPhone(number);
}
Long endMills = System.currentTimeMillis();
return endMills - startMills;
}
}
}