java多线程采集，多线程数据同步的问题。 怎么做同步啊。。

Posted 2023-04-25

参考技术A 源代码如下：
1、线程类
package com.yjf.util; import java.util.Date; import java.util.List; public class GetWebThread extends Thread /** * 线程 */ public void run() try while (true) int day = 0; long time1 = new Date().getTime(); //用来同步抓取线程 synchronized("searchthead") Main.thisdaycount++; if(Main.thisdaycount>Main.daycount) break; System.out.println("开始查询第"+(Main.thisdaycount)+"天"); Thread.sleep(133); day = Main.thisdaycount-1; //获取抓取的时间 String datetext = TimeUtil.date.format(TimeUtil.addDaysForDate(day)); String[] txt =FileUtil.getCityByTxtFile(); for(int t=0;t<txt.length;t++) String[] way = txt[t].split("\|"); String start = way[0]; String end = way[1]; //抓取到的数据列表 List<DataBean> datalist = Main.getDataList(datetext, start, end); if(datalist!=null) Main.isadsl = 0; CheckAdsl.adsllasttime = new Date().getTime(); FileUtil.addDataToFileCsv(datalist); Main.log.printLog("===="+datetext+"="+start+"="+end+"="+t+"=数据总数："+datalist.size()); else Thread.sleep(11); AdslThead.isadsl = true; Thread.sleep(11); //判断是否正在拨号 并暂停线程 while (AdslThead.isadsl) Thread.sleep(5000); t--; long time2 = new Date().getTime(); Main.log.printLog(datetext+"==查询完毕=========耗时"+(time2-time1)); catch (Exception e) Main.log.printLog(e.getMessage()); e.printStackTrace(); 第二步， 准备线程启动。
控制线程数量。 http://www.qi788.com
查询数据总天数。 设置拨号状态等。
2、线程启动类。 包括启动，控制，停止
package com.yjf.util; import java.io.BufferedInputStream; import java.io.FileInputStream; import java.io.InputStream; import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.Properties; import java.util.Timer; public class Main private static boolean isRunGrabThread = true; //抓取线程是否执行完毕 public static int threadCount = 3; //线程数量 public static int daycount = 30; //查询总天数 public static int thisdaycount = 0; public static int isadsl = 0; public static int adslcount = 1; public static void main(String[] args) try startThead();//启动抓取数据主线程 catch (Exception e) e.printStackTrace(); public static void startThead() Thread[] grabThreads= new Thread[threadCount]; try //开启-数据抓取子线程 for(int i=0;i<grabThreads.length;i++) Thread searchThread=new GetWebThread(); grabThreads[i] = searchThread; grabThreads[i].setDaemon(true); grabThreads[i].start(); isRunGrabThread = true; //监控子线程,全部完成程序退出
WhileLoop: while(true) //拨号策略控制 //reconnectControl(); //判断子线程是否运行完成 for(int i=0;i<threadCount;i++) if(grabThreads[i].isAlive()) Thread.sleep(3*1000); continue WhileLoop; //所有抓取线程执行完毕 isRunGrabThread = false; //子线程执行完毕则退出 break; catch (Exception e) System.out.println("main主线程--系统异常!");
文章来源注明： http://qi788.com/info/11.html

Java多线程——线程同步

在之前，已经学习到了线程的创建和状态控制，但是每个线程之间几乎都没有什么太大的联系。可是有的时候，可能存在多个线程多同一个数据进行操作，这样，可能就会引用各种奇怪的问题。现在就来学习多线程对数据访问的控制吧。

 

 由于同一进程的多个线程共享同一片存储空间，在带来方便的同时，也带来了访问冲突这个严重的问题。Java语言提供了专门机制以解决这种冲突，有效避免了同一个数据对象被多个线程同时访问。

 

一、多线程引起的数据访问安全问题

下面看一个经典的问题，银行取钱的问题：

1）、你有一张银行卡，里面有5000块钱，然后你到取款机取款，取出3000，当正在取的时候，取款机已经查询到你有5000块钱，然后正准备减去300块钱的时候

2）、你的老婆拿着那张银行卡对应的存折到银行取钱，也要取3000.然后银行的系统查询，存折账户里还有6000（因为上面钱还没扣），所以它也准备减去3000，

3）、你的卡里面减去3000，5000-3000=2000，并且你老婆的存折也是5000-3000=2000。

4）、结果，你们一共取了6000，但是卡里还剩下2000。

下面看程序的模拟过程：

 

[java] view plain copy

1. package com.tao.test;  
2.   
3. public class GetMoneyTest {  
4.     public static void main(String[] args) {  
5.         Account account = new Account(5000);  
6.         GetMoneyRun runnable = new GetMoneyRun(account);  
7.         new Thread(runnable, "你").start();  
8.         new Thread(runnable, "你老婆").start();  
9.     }  
10. }  
11.   
12. // 账户Mode  
13. class Account {  
14.     private int money;  
15.   
16.     public Account(int money) {  
17.         super();  
18.         this.money = money;  
19.     }  
20.   
21.     public int getMoney() {  
22.         return money;  
23.     }  
24.   
25.     public void setMoney(int money) {  
26.         this.money = money;  
27.     }  
28.   
29. }  
30. //runnable类  
31. class GetMoneyRun implements Runnable {  
32.     private Account account;  
33.   
34.     public GetMoneyRun(Account account) {  
35.         this.account = account;  
36.     }  
37.   
38.     @Override  
39.     public void run() {  
40.         if (account.getMoney() > 3000) {  
41.             System.out.println(Thread.currentThread().getName() + "的账户有"  
42.                     + account.getMoney() + "元");  
43.             try {  
44.                 Thread.sleep(10);  
45.             } catch (InterruptedException e) {  
46.                 e.printStackTrace();  
47.             }  
48.             int lasetMoney=account.getMoney() - 3000;  
49.             account.setMoney(lasetMoney);  
50.             System.out.println(Thread.currentThread().getName() + "取出来了3000元"  
51.                     + Thread.currentThread().getName() + "的账户还有"  
52.                     + account.getMoney() + "元");  
53.   
54.         } else {  
55.             System.out.println("余额不足3000" + Thread.currentThread().getName()  
56.                     + "的账户只有" + account.getMoney() + "元");  
57.         }  
58.   
59.     }  
60.   
61. }  

多次运行程序，可以看到有多种不同的结果，下面是其中的三种：

 

 

[java] view plain copy

1. 你的账户有5000元  
2. 你老婆的账户有5000元  
3. 你老婆取出来了3000元你老婆的账户还有2000元  
4. 你取出来了3000元你的账户还有-1000元  

 

[java] view plain copy

1. 你的账户有5000元  
2. 你老婆的账户有5000元  
3. 你老婆取出来了3000元你老婆的账户还有-1000元  
4. 你取出来了3000元你的账户还有-1000元  

 

[java] view plain copy

1. 你的账户有5000元  
2. 你老婆的账户有5000元  
3. 你老婆取出来了3000元你老婆的账户还有2000元  
4. 你取出来了3000元你的账户还有2000元  

可以看到，由于有两个线程同时访问这个account对象，导致取钱发生的账户发生问题。当多个线程访问同一个数据的时候，非常容易引发问题。为了避免这样的事情发生，我们要保证线程同步互斥，所谓同步互斥就是：并发执行的多个线程在某一时间内只允许一个线程在执行以访问共享数据。

 

 

二、同步互斥锁

同步锁的原理：Java中每个对象都有一个内置同步锁。Java中可以使用synchronized关键字来取得一个对象的同步锁。synchronized的使用方式，是在一段代码块中，加上synchronized(object){ ... }

例如，有一个show方法，里面有synchronized的代码段：

 

[java] view plain copy

1. public void show() {  
2.     synchronized(object){  
3.        ......  
4.     }  
5. }  

 

这其中的object可以使任何对象，表示当前线程取得该对象的锁。一个对象只有一个锁，所以其他任何线程都不能访问该对象的所有由synchronized包括的代码段，直到该线程释放掉这个对象的同步锁（释放锁是指持锁线程退出了synchronized同步方法或代码块）。

注意：synchronized使用方式有几个要注意的地方（还是以上面的show方法举例）：

①、取得同步锁的对象为this，即当前类对象，这是使用的最多的一种方式

 

[java] view plain copy

1. public void show() {  
2.     synchronized(this){  
3.        ......  
4.     }  
5. }  

 

 

②、将synchronized加到方法上，这叫做同步方法，相当于第一种方式的缩写

[java] view plain copy

1. public synchronized void show() {  
2.      
3. }  

 

③、静态方法的同步

[java] view plain copy

1. public static synchronized void show() {  
2.      
3. }  

相当于

 

[java] view plain copy

1. public static void show() {  
2.    synchronized(当前类名.class)   
3. }  

相当于取得类对象的同步锁，注意它和取得一个对象的同步锁不一样

 

 

明白了同步锁的原理和synchronized关键字的使用，那么解决上面的取钱问题就很简单了，我们只要对run方法里面加上synchronized关键字就没有问题了，如下：

[java] view plain copy

1. @Override  
2.     public void run() {  
3.         synchronized (account) {  
4.             if (account.getMoney() > 3000) {  
5.                 System.out.println(Thread.currentThread().getName() + "的账户有"  
6.                         + account.getMoney() + "元");  
7.                 try {  
8.                     Thread.sleep(10);  
9.                 } catch (InterruptedException e) {  
10.                     e.printStackTrace();  
11.                 }  
12.                 int lasetMoney = account.getMoney() - 3000;  
13.                 account.setMoney(lasetMoney);  
14.                 System.out.println(Thread.currentThread().getName()  
15.                         + "取出来了3000元" + Thread.currentThread().getName()  
16.                         + "的账户还有" + account.getMoney() + "元");  
17.   
18.             } else {  
19.                 System.out.println("余额不足3000"  
20.                         + Thread.currentThread().getName() + "的账户只有"  
21.                         + account.getMoney() + "元");  
22.             }  
23.   
24.         }  
25.     }  

当甲线程执行run方法的时候，它使用synchronized (account)取得了account对象的同步锁，那么只要它没释放掉这个锁，那么当乙线程执行到run方法的时候，它就不能获得继续执行的锁，所以只能等甲线程执行完，然后释放掉锁，乙线程才能继续执行。

 

 

synchronized关键字使用要注意以下几点：

1）、只能同步方法和代码块，而不能同步变量和类。只要保护好类中数据的安全访问和设置就可以了，不需要对类使用synchronized关键字，所以Java不允许这么做。并且想要同步数据，只需要对成员变量私有化，然后同步方法即可，不需要对成员变量使用synchronized，java也禁止这么做。

2）、每个对象只有一个同步锁；当提到同步时，应该清楚在什么上同步？也就是说，在哪个对象上同步？上面的代码中run方法使用synchronized (account)代码块，因为两个线程访问的都是同一个Account对象，所以能够锁定。但是如果是其他的一个无关的对象，就没用了。比如说synchronized (new Date())代码块，一样没有效果。

3）、不必同步类中所有的方法，类可以同时拥有同步和非同步方法。

 

4）、如果两个线程要执行一个类中的synchronized方法，并且两个线程使用相同的实例来调用方法，那么一次只能有一个线程能够执行方法，另一个需要等待，直到锁被释放。也就是说：如果一个线程在对象上获得一个锁，就没有任何其他线程可以进入（该对象的）类中的任何一个同步方法。

5）、如果线程拥有同步和非同步方法，则非同步方法可以被多个线程自由访问而不受锁的限制。

6）、线程睡眠时，它所持的任何同步锁都不会释放。

7）、线程可以获得多个同步锁。比如，在一个对象的同步方法里面调用另外一个对象的同步方法，则获取了两个对象的同步同步锁。

8）、同步损害并发性，应该尽可能缩小同步范围。同步不但可以同步整个方法，还可以同步方法中一部分代码块。

 

9）、编写线程安全的代码会使系统的总体效率会降低，要适量使用

 

 

一个线程取得了同步锁，那么在什么时候才会释放掉呢？

1、同步方法或代码块正常结束

2、使用return或 break终止了执行，或者跑出了未处理的异常。

3、当线程执行同步方法或代码块时，程序执行了同步锁对象的wait()方法。

 

三、死锁

死锁：多个线程同时被阻塞，它们中的一个或者全部都在等待某个资源被释放。由于线程被无限期地阻塞，因此程序不能正常运行。简单的说就是:线程死锁时，第一个线程等待第二个线程释放资源，而同时第二个线程又在等待第一个线程释放资源。这里举一个通俗的例子：如在人行道上两个人迎面相遇，为了给对方让道，两人同时向一侧迈出一步，双方无法通过，又同时向另一侧迈出一步，这样还是无法通过。假设这种情况一直持续下去，这样就会发生死锁现象。

    导致死锁的根源在于不适当地运用“synchronized”关键词来管理线程对特定对象的访问。“synchronized”关键词的作用是，确保在某个时刻只有一个线程被允许执行特定的代码块，因此，被允许执行的线程首先必须拥有对变量或对象的排他性访问权。当线程访问对象时，线程会给对象加锁，而这个锁导致其它也想访问同一对象的线程被阻塞，直至第一个线程释放它加在对象上的锁。

一个死锁的造成很简单，比如有两个对象A 和 B 。第一个线程锁住了A，然后休眠1秒，轮到第二个线程执行，第二个线程锁住了B，然后也休眠1秒，然后有轮到第一个线程执行。第一个线程又企图锁住B，可是B已经被第二个线程锁定了，所以第一个线程进入阻塞状态，又切换到第二个线程执行。第二个线程又企图锁住A，可是A已经被第一个线程锁定了，所以第二个线程也进入阻塞状态。就这样，死锁造成了。

举个例子：

 

[java] view plain copy

1. package com.tao.test;  
2.   
3. public class DeadLock2 {  
4.     public static void main(String[] args) {  
5.         Object object1=new Object();  
6.         Object object2=new Object();  
7.         new Thread(new T(object1,object2)).start();  
8.         new Thread(new T(object2,object1)).start();  
9.     }  
10. }  
11. class T implements Runnable{  
12.     private Object object1;  
13.     private Object object2;  
14.     public T(Object object1,Object object2) {  
15.         this.object1=object1;  
16.         this.object2=object2;  
17.     }  
18.     public void run() {  
19.         synchronized (object1) {  
20.             try {  
21.                 Thread.sleep(1000);  
22.             } catch (InterruptedException e) {  
23.                 e.printStackTrace();  
24.             }  
25.             synchronized (object2) {  
26.                 System.out.println("无法执行到这一步");  
27.             }  
28.         }  
29.     };  
30. }  

上面的就是个死锁。

 

第一个线程首先锁住了object1，然后休眠。接着第二个线程锁住了object2，然后休眠。在第一个线程企图在锁住object2，进入阻塞。然后第二个线程企图在锁住object1，进入阻塞。死锁了。

 

 

 

四、线程的协调运行

关于线程的协调运行，经典的例子就是生产者和消费者的问题。比如有生产者不断的生产馒头，放入一个篮子里，而消费者不断的从篮子里拿馒头吃。并且，当篮子满的时候，生产者通知消费者来吃馒头，并且自己等待不在生产馒头。当篮子没满的的时候，由消费者通知生产者生产馒头。这样不断的循环。

要完成上面的功能，光靠我们前面的同步等知识，是不能完成的。而是要用到线程间的协调运行。顶级父类Object中有3种方法来控制线程的协调运行。

notify、notifyAll、wait。其中wait有3个重载的方法。

这三个方法必须由同步监视器对象（即线程获得的锁对象）来调用，这可分为两种情况：

1、对于使用synchronized修饰的同步代码块，因为当前的类对象（this）就是同步监视器，所以可以再同步方法中直接调用这三个方法。

2、对于使用synchronized修饰的同步代码块，同步监视器是synchronized后括号的对象，所以必须使用该对象调用这三个方法。

wait()： 导致当前线程等待，直到其他线程调用该同步监视器的notify()方法或notifyAll()方法来唤醒该线程。wait()方法有三种形式：无时间参数的wait(一直等待，直到其他线程通知)，带毫秒参数的wait和带毫秒、微秒参数的wait（这两种方法都是等待指定时间后自动苏醒）。调用wait()方法的当前线程会释放对该同步监视器的锁定。

notify()： 唤醒在此同步监视器上等待的单个线程。如果所有线程都在此同步监视器上等待，则会选择幻想其中一个线程。选择是任意性。只有当前线程放弃对该同步监视器的锁定后(使用wait()方法)，才可以执行被唤醒的其他线程。

notifyAll()：唤醒在此同步监视器上等待的所有线程。只有当前线程放弃对该同步监视器的锁定后，才可以执行被唤醒的线程。

 

因为使用wait、notify和notifyAll三个方法一定是在同步代码块中使用的，所以一定要明白下面几点：

1、如果两个线程是因为都要得到同一个对象的锁，而导致其中一个线程进入阻塞状态。那么只有等获得锁的线程执行完毕，或者它执行了该锁对象的wait方法，阻塞的线程才会有机会得到锁，继续执行同步代码块。

2、使用wait方法进入等待状态的线程，会释放掉锁。并且只有其他线程调用notify或者notifyAll方法，才会被唤醒。要明白，线程因为锁阻塞和等待是不同的，因为锁进入阻塞状态，会在其他线程释放锁的时候，得到锁在执行。而等待状态必须要靠别人唤醒，并且唤醒了也不一定会立刻执行，有可能因为notifyAll方法使得很多线程被唤醒，多个线程等待同一个锁，而进入阻塞状态。还可能是调用notify的线程依然没有释放掉锁，只有等他执行完了，其他线程才能去争夺这个锁。

看下面的例子：

[java] view plain copy

1. package com.tao.test;  
2.   
3. public class ThreadA {  
4.     public static void main(String[] args) {  
5.         RunnableTest myRunnanle=new RunnableTest();  
6.        new Thread(myRunnanle).start();  
7.         synchronized (myRunnanle) {  
8.             try {  
9.                 System.out.println("第一步");  
10.                 myRunnanle.wait();  
11.             } catch (InterruptedException e) {  
12.                 e.printStackTrace();  
13.             }  
14.             System.out.println("第四步");  
15.         }  
16.     }  
17. }  
18.  class RunnableTest implements Runnable {  
19.     public void run() {  
20.         try {  
21.             Thread.sleep(1);  
22.         } catch (InterruptedException e) {  
23.             e.printStackTrace();  
24.         }  
25.         synchronized (this) {  
26.             System.out.println("第二步");  
27.             notify();  
28.             System.out.println("第三步");  
29.         }  
30.     }  
31. }  

有两个线程，主线程和我们自己新建的子线程。一步步的分析程序的执行：

1、因为子线程启动后，调用了sleep，所以主线程先进入同步代码块，而子线程之后因为没有锁，会进入阻塞状态。

2、主线程的同步代码块执行，打印第一句话，然后调用wait方法，进入等待状态。因为进入了等待状态，所以释放掉了锁，所以子线程可以获得锁，开始执行。

3、子线程执行，打印第二句话，然后调用notify方法，将主线程唤醒。可是子线程并没有结束，依然持有锁，所以主线程不得不进入阻塞状态，等待这个锁。

4、子线程打印第三句话，然后线程正常运行结束，释放掉锁。然后主线程得到了锁，从阻塞进入运行状态，打印第四句话。

5、完毕

 

在看一个关于上面提到的生产者和消费者的例子：

首先，是生产物品的Mode，这里以馒头举例：

 

[java] view plain copy

1. // 馒头的实例  
2. class ManTou {  
3.     private int id;// 馒头的id  
4.   
5.     public ManTou(int id) {  
6.         this.id = id;  
7.     }  
8.     public String toString(){  
9.         return "ManTou"+id;  
10.     }  
11. }  

共享对象，生产者生产的馒头放入其中，消费者从里面拿出馒头，这里以篮子举例：

 

 

[java] view plain copy

1. // 篮子的实例，用来放馒头  
2. class BasketBall {  
3.     private int index = 0;// 表示装到第几个了馒头  
4.     private ManTou[] manTous = new ManTou[6];// 可以放6个馒头  
5.   
6.     // 放进去一个馒头  
7.     public synchronized void push(ManTou manTou) {  
8.         while(index==manTous.length){  
9.             try {  
10.                 System.out.println("篮子满了！");  
11.                 this.wait();  
12.             } catch (InterruptedException e) {  
13.                 e.printStackTrace();  
14.             }  
15.         }  
16.         System.out.println(Thread.currentThread().getName()+"生产"+manTou.toString());  
17.         this.notify();  
18.         manTous[index] = manTou;  
19.         index++;  
20.     }  
21.   
22.     // 拿一个馒头  
23.     public synchronized ManTou pop() {  
24.         while (index==0) {  
25.             try {  
26.                 System.out.println("篮子空了！");  
27.                 this.wait();  
28.             } catch (InterruptedException e) {  
29.                 e.printStackTrace();  
30.             }  
31.         }  
32.         ManTou manTou=manTous[--index];  
33.         System.out.println(Thread.currentThread().getName()+"吃了"+manTou.toString());  
34.         this.notify();  
35.         return manTou;  
36.     }  
37. }  

生产者：

 

 

[java] view plain copy

1. // 生产者，生产馒头  
2. class Producer implements Runnable {  
3.     private BasketBall basketBall;  
4.   
5.     public Producer(BasketBall basketBall) {  
6.         this.basketBall = basketBall;  
7.     }  
8.   
9.     @Override  
10.     public void run() {  
11.         for (int i = 0; i < 20; i++) {  
12.             ManTou manTou = new ManTou(i);// 生产馒头  
13.             basketBall.push(manTou);  
14.             try {  
15.                 Thread.sleep(500);  
16.             } catch (InterruptedException e) {  
17.                 e.printStackTrace();  
18.             }  
19.         }  
20.     }  
21. }  

 

 

[java] view plain copy