import java.io.BufferedReader;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.concurrent.*;
public class ceshi2 {
public static void main(String[] args) throws InterruptedException, ExecutionException {
LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>(100);
ExecutorService executor = Executors.newFixedThreadPool(10); // 创建一个固定大小的线程池,最多同时执行10个线程
// 将读取文件任务提交给线程池
Future<?> readerFuture = executor.submit(new ReaderTask(queue, "/Users/flyme/data/input-50w-ex.txt"));
// 创建10个处理任务,将它们都提交给线程池
Future<?>[] workerFutures = new Future[10];
for (int i = 0; i < 10; i++) {
workerFutures[i] = executor.submit(new WorkerTask(queue));
}
// 等待读取任务完成
readerFuture.get();
// 等待处理任务完成
for (Future<?> workerFuture : workerFutures) {
workerFuture.get();
}
// 关闭线程池
executor.shutdown();
System.out.println("所有任务完成");
}
}
class ReaderTask implements Runnable {
private LinkedBlockingQueue<String> queue;
private String filePath;
public ReaderTask(LinkedBlockingQueue<String> queue, String filePath) {
this.queue = queue;
this.filePath = filePath;
}
@Override
public void run() {
try (BufferedReader reader = new BufferedReader(new FileReader(filePath))) {
String line;
while ((line = reader.readLine()) != null) {
queue.put(line); // 将每行内容放入队列
// System.out.println(Thread.currentThread().getName() + ": 发送到队列中:" + line);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
queue.put("EOF"); // 放置一个结束标记
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
class WorkerTask implements Runnable {
private LinkedBlockingQueue<String> queue;
public WorkerTask(LinkedBlockingQueue<String> queue) {
this.queue = queue;
}
@Override
public void run() {
try {
String line;
while ((line = queue.poll(1, TimeUnit.SECONDS)) != null) { // 从队列中取出一行数据
if ("EOF".equals(line)) {
queue.put(line); // 将结束标记放回队列
break;
}
System.out.println(Thread.currentThread().getName() + ": 队列中取到:" + line);
String[] parts = line.split(",");
String id = parts[0];
String address = parts[1];
// 向API接口发送请求,并获取返回结果
String result = sendRequest(address);
// 解析返回结果,提取商品信息
String[] coordinates = parseCoordinates(result);
// 拼接id、商品id、商品信息成一行记录
String record = id + "," + address + "," + coordinates[0] + "," + coordinates[1];
// 将记录写入输出文件
writeRecordToFile(record);
}
} catch (InterruptedException | IOException e) {
e.printStackTrace();
}
}
private String sendRequest(String address) {
// 发送http请求代码省略
return "名称,销量"; // 模拟返回结果
}
private String[] parseCoordinates(String result) {
// 解析商品信息代码省略
return new String[]{"名称", "销量"}; // 模拟解析结果
}
private void writeRecordToFile(String record) throws IOException {
try (FileWriter writer = new FileWriter("/Users/flyme/data/output-50w-ex.txt", true)) { // 追加模式写入
writer.write(record + System.lineSeparator());
}
}
}
标签:String,queue,线程,new,line,多线程,public,BlockingQueue
From: https://www.cnblogs.com/flyme6/p/17300144.html