多线程(二)线程基础、线程之间的共享和协作

多线程(二)线程基础、线程之间的共享和协作

1.线程间协作

如果要完成一个系统功能,同样需要各个线程的配合,这样就少不了线程之间的通信与协作。

1.1 等待和通知

wait()和notify/notifyAll()都是对象上的方法

1.1.1 等待和通知的标准范式

等待方:

  • 获取对象的锁;
  • 循环里判断条件是否满足,不满足调用wait方法。
  • 条件满足执行业务逻辑

通知方:

  • 获取对象的锁;
  • 改变条件
  • 通知所有等待在对象的线程

定义快递实体类,其中包含两个变量km和site。它们分别代表快递距离目的地的距离和快递现在所处的位置。

public class Express {
    public final static String CITY = "ShangHai";
    private int km;/*快递运输里程数*/
    private String site;/*快递到达地点*/

    public Express() {
    }

    public Express(int km, String site) {
        this.km = km;
        this.site = site;
    }

    /* 变化公里数,然后通知处于wait状态并需要处理公里数的线程进行业务处理*/
    public synchronized void changeKm(){
        this.km = 101;
        notifyAll();
        //其他的业务代码

    }

    /* 变化地点,然后通知处于wait状态并需要处理地点的线程进行业务处理*/
    public synchronized void changeSite(){
        this.site = "BeiJing";
        notifyAll();
    }

    public synchronized void waitKm(){
        while(this.km<=100) {
            try {
                wait();
                System.out.println("check km thread["+Thread.currentThread().getId()
                        +"] is be notifed.");
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
        System.out.println("the km is"+this.km+",I will change db.");

    }

    public synchronized void waitSite(){
        while(CITY.equals(this.site)) {
            try {
                wait();
                System.out.println("check site thread["+Thread.currentThread().getId()
                        +"] is be notifed.");
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
        System.out.println("the site is"+this.site+",I will call user.");
    }
}

测试类:

public class TestWN {
    private static Express express = new Express(0,Express.CITY);

    /*检查里程数变化的线程,不满足条件,线程一直等待*/
    private static class CheckKm extends Thread{
        @Override
        public void run() {
            express.waitKm();
        }
    }

    /*检查地点变化的线程,不满足条件,线程一直等待*/
    private static class CheckSite extends Thread{
        @Override
        public void run() {
        	express.waitSite();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        for(int i=0;i<3;i++){//三个线程
            new CheckSite().start();
        }
        for(int i=0;i<3;i++){//里程数的变化
            new CheckKm().start();
        }

        Thread.sleep(1000);
        express.changeKm();//快递距离变化
    }
}

输出结果:

check km thread[16] is be notifed.
the km is101,I will change db.
check km thread[15] is be notifed.
the km is101,I will change db.
check km thread[14] is be notifed.
the km is101,I will change db.
check site thread[13] is be notifed.
check site thread[12] is be notifed.
check site thread[11] is be notifed.

1.1.2 notify()和notifyAll()应该选择哪个方法更好?

更改changeKm()和changeSite()方法

 /* 变化公里数,然后通知处于wait状态并需要处理公里数的线程进行业务处理*/
    public synchronized void changeKm(){
        this.km = 101;
        notify();
        //其他的业务代码

    }

    /* 变化地点,然后通知处于wait状态并需要处理地点的线程进行业务处理*/
    public synchronized void changeSite(){
        this.site = "BeiJing";
        notify();
    }

测试结果:

check site thread[11] is be notifed.

notify只能唤醒处于等待队列中的一个线程。上面的结果显示notify唤醒了检查地点变化的线程 而我们希望它唤醒检查距离变化的线程 这就造成了信号丢失的现象。
使用notifyAll会唤醒所有的线程,这样就防止了信号丢失的发生。所以建议使用notifyAll()方法

1.1.3 等待超时模式

使用等待超时模式实现一个数据库连接池

public class SqlConnectImpl implements Connection{
	
	/*拿一个数据库连接*/
    public static final Connection fetchConnection(){
        return new SqlConnectImpl();
    }
}


public class DBPool {

    private static LinkedList<Connection> pool = new LinkedList<>();

    //initialSize为连接池初始化容量
    public DBPool(int initialSize) {
        for (int i = 0; i < initialSize; i++) {
            //取出数据库连接并且添加到连接池的末尾
            pool.addLast(new SqlConnectImpl().fetchConnection());
        }
    }

    //从连接池中获取连接的方法 mills为超时时长
    public Connection fetchConn(long mills) throws InterruptedException {
        synchronized (pool) {
            //如果传入的mills<0 则永不超时 会一直等待
            if (mills < 0) {
                //当连接池为空时则一直等待
                while (pool.isEmpty()){
                    pool.wait();
                }
                //当连接池不为空时则移除并返回第一个连接
                return pool.removeFirst();
            } else {
                //超时的时刻
                long overTime = System.currentTimeMillis() + mills;
                //剩余的超时时间
                long remain = mills;
                //如果连接池为空,并且未到达超时时刻 则继续等待 并且更新剩余的超时时间
                if (pool.isEmpty() && remain > 0) {
                    pool.wait(remain);
                    //更新剩余的超时时间
                    remain = overTime - System.currentTimeMillis();
                }
                //连接池为空并且已经到达超时时间 直接返回null
                Connection result = null;
                if (!pool.isEmpty()) {
                    //当连接池不为空时 从连接池中取出第一个连接 并且将其从池中移除
                    result = pool.removeFirst();
                }
                return result;
            }
        }
    }

    //释放数据库连接池的连接
    public void releaseConn(Connection conn) {
        if (conn != null) {
            synchronized (pool) {
                pool.addLast(conn);
                //通知所有的线程 连接池中有连接了
                pool.notifyAll();
            }
        }
    }
}


测试类:

public class DBPoolTest {
    static DBPool pool  = new DBPool(10);
    // 控制器:控制main线程将会等待所有Woker结束后才能继续执行
    static CountDownLatch end;

    public static void main(String[] args) throws Exception {
    	// 线程数量
        int threadCount = 50;
        end = new CountDownLatch(threadCount);
        int count = 20;//每个线程的操作次数
        AtomicInteger got = new AtomicInteger();//计数器:统计可以拿到连接的线程
        AtomicInteger notGot = new AtomicInteger();//计数器:统计没有拿到连接的线程
        for (int i = 0; i < threadCount; i++) {
            Thread thread = new Thread(new Worker(count, got, notGot), 
            		"worker_"+i);
            thread.start();
        }
        end.await();// main线程在此处等待
        System.out.println("总共尝试了: " + (threadCount * count));
        System.out.println("拿到连接的次数:  " + got);
        System.out.println("没能连接的次数: " + notGot);
    }

    static class Worker implements Runnable {
        int           count;
        AtomicInteger got;
        AtomicInteger notGot;

        public Worker(int count, AtomicInteger got,
                               AtomicInteger notGot) {
            this.count = count;
            this.got = got;
            this.notGot = notGot;
        }

        public void run() {
            while (count > 0) {
                try {
                    // 从线程池中获取连接,如果1000ms内无法获取到,将会返回null
                    // 分别统计连接获取的数量got和未获取到的数量notGot
                    Connection connection = pool.fetchConn(1000);
                    if (connection != null) {
                        try {
                            connection.createStatement();
                            connection.commit();
                        } finally {
                            pool.releaseConn(connection);
                            got.incrementAndGet();
                        }
                    } else {
                        notGot.incrementAndGet();
                        System.out.println(Thread.currentThread().getName()
                        		+"等待超时!");
                    }
                } catch (Exception ex) {
                } finally {
                    count--;
                }
            }
            end.countDown();
        }
    }
}

测试结果:

worker_48等待超时!
worker_42等待超时!
worker_43等待超时!
worker_4等待超时!
worker_4等待超时!
总共尝试了: 1000
拿到连接的次数:  155
没能连接的次数: 845
......


2. join()方法

  • 在很多情况下,主线程创建并启动子线程,如果子线程中要进行大量的耗时运算,主线程将早于子线程结束。这时,如果主线程想等子线程执行完成才结束,比如子线程处理一个数据,主线程想要获得这个数据中的值,就要用到join()方法了。方法join()的作用是等待线程对象销毁。

  • join方法的主要作用就是同步,它可以使得线程之间的并行执行变为串行执行。在A线程中调用了B线程的join()方法时,表示只有当B线程执行完毕时,A线程才能继续执行。


  • join方法中如果传入参数,则表示这样的意思:如果A线程中调用B线程的join(10),则表示A线程会等待B线程执行10毫秒,10毫秒过后,A、B线程并行执行。需要注意的是,jdk规定,join(0)的意思不是A线程等待B线程0秒,而是A线程等待B线程无限时间,直到B线程执行完毕,即join(0)等价于join()。(其实join()中调用的是join(0))
  • join方法必须在线程start方法调用之后调用才有意义。这个也很容易理解:如果一个线程都没有start,那它也就无法同步了。


JumpQueue的构造方法能接收一个线程 该线程作为插队线程。例如:当Thread[0]创建并运行后 传入的主线程作为插队线程来调用Thread[0]的join方法 这样Thread[0]就会放弃执行权 将执行权交给主线程 以此类推创建的10个线程中Thread[9]是最后执行的。

代码演示

public class UseJoin {
	
    //用于插队的线程
    static class JumpQueue implements Runnable {

        private Thread thread;//用来插队的线程

        public JumpQueue(Thread thread){
            this.thread = thread;
        }

        @Override
        public void run() {
            try {
                //thread 代表传入的插队线程 currentThread代表现在正在运行的线程
                System.out.println(thread.getName() + "插入到了线程[" + Thread.currentThread().getName() + "]的前面");
                        + Thread.currentThread().getName());
                //第一次插队线程main调用Thread[0]的join()方法
                thread.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("线程[" + Thread.currentThread().getName() + "]运行完成了");
        }
    }

    public static void main(String[] args) throws Exception {

        Thread previous = Thread.currentThread();//现在是主线程

        for (int i = 0; i < 10; i++) {
            //i=0,previous 是主线程,i=1;previous是i=0这个线程
            Thread thread =
                    new Thread(new JumpQueue(previous), String.valueOf(i));
            System.out.println(previous.getName()+" jump a queue the thread:"
                    +thread.getName());
            thread.start();
            previous = thread;
        }

        SleepTools.second(2);//让主线程休眠2秒
        System.out.println("线程[" + Thread.currentThread().getName() + "]运行完成了");
    }
}

执行结果:

0插入到了线程[1]的前面
main插入到了线程[0]的前面
1插入到了线程[2]的前面
2插入到了线程[3]的前面
3插入到了线程[4]的前面
4插入到了线程[5]的前面
5插入到了线程[6]的前面
6插入到了线程[7]的前面
7插入到了线程[8]的前面
8插入到了线程[9]的前面
线程[main]运行完成了
线程[0]运行完成了
线程[1]运行完成了
线程[2]运行完成了
线程[3]运行完成了
线程[4]运行完成了
线程[5]运行完成了
线程[6]运行完成了
线程[7]运行完成了
线程[8]运行完成了
线程[9]运行完成了

★调用yield() 、sleep()、wait()、notify()等方法对锁有何影响?

线程在执行yield()以后,持有的锁是不释放的 sleep()方法被调用以后,持有的锁是不释放的 调动方法之前,必须要持有锁。调用了wait()方法以后,锁就会被释放(如果不释放锁 那么notify就拿不到锁),当wait方法返回的时候,线程会重新持有锁 调动方法之前,必须要持有锁,调用notify()方法本身不会释放锁的

评论

Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×