多线程和同步队列

淡淡的烟草味﹌ 2021-12-05 13:25 312阅读 0赞

最近要割接个项目，要把另外一个公司的数据库里的一张表倒到我们库里，数据有一亿三千多万吧。正号也符合生产者和消费者的状况。以前用过点线程池和同步队列，，写个例子，让大家拍砖。不多说了，直接上代码，

1. 线程池

public class ThreadPool \{

private ExecutorService exe = null;// 线程池

private int pool\_size;

private  Service  service ;

public ThreadPool(int pool\_size, Service  service   \{

this.pool\_size = pool\_size;

this. service  = service  ;

exe = Executors.newFixedThreadPool(pool\_size);// 创建线程池

System.out.println("the server is ready");

/\*\*

\* 运行循环实例线程，根据要实例的线程个数，传入条件ID

\* @param worknum

\*/

public void server() \{

int i = 0;

while (i < pool\_size) \{

// 实例指定个线程

InsertData t = new InsertData( service  );

exe.execute(t);// 放入线程池

i++;

2.线程

public class InsertData extends Thread \{

private Service  service ;

public InsertData( Service  service  )\{

this. service  = service  ;

@Override

public void run() \{

SyncDataQueue syncDataQueue=SyncDataQueue.getInstance();

while (true) \{

List< Mode > list=new ArrayList<Mode>();

//移除队列里的所有实体，并批量添加

syncDataQueue.getBlockingQueue().drainTo(list);

if (list!=null) \{

service.batchSvae(list);

3.实现同步队列

public class SyncDataQueue \{

private static SyncDataQueue instance = null;

private BlockingQueue< Mode > blockingQueue = null;

public SyncDataQueue() \{

//队列容量

blockingQueue = new ArrayBlockingQueue< Mode >(40000);

public static SyncDataQueue getInstance() \{

if (instance == null) \{

instance = new SyncDataQueue();

return instance;

public BlockingQueue< Mode > getBlockingQueue() \{

return blockingQueue;

4.导入文件，并启动线程池

public class FileThread extends Thread \{

private String path;

public FileThread(String path) \{

this.path = path;

@Override

public void run() \{

if (StringUtils.isNotBlank(path)) \{

try \{

getSmsListByFileThread(new File(path));

\} catch (FileNotFoundException e) \{

// TODO Auto-generated catch block

e.printStackTrace();

\} catch (IOException e) \{

// TODO Auto-generated catch block

e.printStackTrace();

\} catch (InterruptedException e) \{

// TODO Auto-generated catch block

e.printStackTrace();

public void getSmsListByFileThread(File file)

throws FileNotFoundException, IOException, InterruptedException \{

//初始化同步队列

SyncDataQueue syncDataQueue = SyncDataQueue.getInstance();

BlockingQueue< Mode > blockingQueue = syncDataQueue.getBlockingQueue();

// 缓冲大小为10M

final int BUFFER\_SIZE = 10 \* 1024 \* 1024;

BufferedInputStream fis = new BufferedInputStream(new FileInputStream(file));

// 用10M的缓冲读取文本文件

BufferedReader reader = new BufferedReader(new InputStreamReader(fis,"utf-8"), BUFFER\_SIZE);

String line = "";

int i=0;

while ((line = reader.readLine()) != null) \{

String\[\] data = line.split(",");

Mode mode=new Mode();

//填充实体

//放进队列 让 线程去抢队列的里实体

blockingQueue.put(mode);

public String getPath() \{

return path;

public void setPath(String path) \{

this.path = path;

/\*\*

\*  不加索引每分钟百万上下， 加了索引有点惨不忍睹。

\*/

public static void main(String\[\] args) \{

Service service=(Service)Syscontext.getBean(" service ");

//初始化线程池，并启动

ThreadPool threadPool = new ThreadPool(30,smsHouseSmsService);

threadPool.server();

//每个文件都是1G的

new FileThread("C:/Users/bjz/Desktop/mcexport/Msg\_20120322104001\_1\_1.txt").start();

........

new FileThread("C:/Users/bjz/Desktop/mcexport/Msg\_20120322104001\_1\_15.txt").start();

转载于:https://www.cnblogs.com/1q2w3e/archive/2012/06/11/2544544.html

多线程和同步队列

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 Java多线程03——线程安全和线程同步

相关 Java线程同步队列SynchronousQueue

相关 java多线程：线程池和阻塞队列

相关 Java多线程（九）线程同步和锁

相关 JUC多线程：AQS抽象队列同步器原理

相关 [Win32] 多线程和线程同步

相关【多线程-线程同步】

相关多线程同步

相关多线程和同步队列

相关多线程（2）- 线程同步