Java并发之Thread基础-2

发表于 | 更新于 | 分类于 ， | 评论数： | 阅读次数：
本文字数： 69k | 阅读时长 ≈ 1:02

Java Thread基础(2)

线程管理：线程组

在系统中如果线程数量很多，而且功能分配比较明确，就可以把相同功能的线程放置在一个线程组里。

可以把线程归属到某一个线程组中，线程组中可以有线程对象，也可以有线程组，组中还可以有线程，这样的组织结构有点类似于树的形式。

线程组的作用是：可以批量管理线程或线程组对象，有效地对线程或线程组对象进行组织。

一级关联(ThreadGroup -> Thread)

  • Thread 对象通过构造器初始化时传入ThreadGroup对象，实现与ThreadGroup对象的关联。
1
2
3
4
public Thread(ThreadGroup group, String name){...}
public Thread(ThreadGroup group, Runnable target){...}
public Thread(ThreadGroup group, Runnable target, String name){...}
public Thread(ThreadGroup group, Runnable target, String name,long stackSize){...}
  • Thread 对象通过方法getThreadGroup()可以获得所属线程组。

批量中断组内线程

  • ThreadGroup 对象通过ThreadGroup.interrupt()方法批量中断组内线程。

例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
public class ThreadGroup2ThreadDemo {

    public static class  myThread implements Runnable{

        public void run() {
            try{
                while (!Thread.currentThread().isInterrupted())
                {
                    System.out.println(
                            "Group: " + Thread.currentThread().getThreadGroup().getName()+
                            " >> Thread: " + Thread.currentThread().getName());
                    Thread.sleep(300);
                }
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ThreadGroup threadGroup = new ThreadGroup("group1");
        Thread t1 = new Thread(threadGroup,new myThread());
        Thread t2 = new Thread(threadGroup,new myThread());
        // 线程启动(start)后才会加入线程组
        System.out.println("线程组中活动线程数量"+threadGroup.activeCount());
        t1.start();
        t2.start();
        Thread.sleep(100);
        System.out.println("线程组中活动线程数量"+threadGroup.activeCount());
        Thread.sleep(3000);
        threadGroup.interrupt();
    }
}
  • output
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
线程组中活动线程数量0
Group: group1 >> Thread: Thread-0
Group: group1 >> Thread: Thread-1
线程组中活动线程数量2
Group: group1 >> Thread: Thread-0
Group: group1 >> Thread: Thread-1
Group: group1 >> Thread: Thread-0
Group: group1 >> Thread: Thread-1
Group: group1 >> Thread: Thread-0
Group: group1 >> Thread: Thread-1
Group: group1 >> Thread: Thread-0
Group: group1 >> Thread: Thread-1
Group: group1 >> Thread: Thread-0
Group: group1 >> Thread: Thread-1
Group: group1 >> Thread: Thread-0
Group: group1 >> Thread: Thread-1
Group: group1 >> Thread: Thread-0
Group: group1 >> Thread: Thread-1
Group: group1 >> Thread: Thread-0
Group: group1 >> Thread: Thread-1
Group: group1 >> Thread: Thread-0
Group: group1 >> Thread: Thread-1
Group: group1 >> Thread: Thread-0
Group: group1 >> Thread: Thread-1
java.lang.InterruptedException: sleep interrupted
    at java.lang.Thread.sleep(Native Method)
    at cn.victor.test.ThreadGroup2ThreadDemo$myThread.run(ThreadGroup2ThreadDemo.java:14)
    at java.lang.Thread.run(Thread.java:748)
java.lang.InterruptedException: sleep interrupted
    at java.lang.Thread.sleep(Native Method)
    at cn.victor.test.ThreadGroup2ThreadDemo$myThread.run(ThreadGroup2ThreadDemo.java:14)
    at java.lang.Thread.run(Thread.java:748)

Process finished with exit code 0

多级关联(ThreadGroup -> ThreadGroup)

  • ThreadGroup 对象之间的关联关系也是通过构造器建立。
1
public ThreadGroup(ThreadGroup parent, String name){...}

  • ThreadGroup 对象通过方法getParent()可以获得父线程组。

  • 父级线程组的中断方法会调用子级线程组的中断方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// 摘自 openJDK 源码
public final void interrupt() {
    int ngroupsSnapshot;
    ThreadGroup[] groupsSnapshot;
    synchronized (this) {
        checkAccess();
        for (int i = 0 ; i < nthreads ; i++) {
            threads[i].interrupt();
        }
        ngroupsSnapshot = ngroups;
        if (groups != null) {
            groupsSnapshot = Arrays.copyOf(groups, ngroupsSnapshot);
        } else {
            groupsSnapshot = null;
        }
    }
    for (int i = 0 ; i < ngroupsSnapshot ; i++) {
        groupsSnapshot[i].interrupt();
    }
}

线程组的递归处理

list()

Prints information about this thread group to the standard output. This method is useful only for debugging.

上面是官方文档的说明，list方法用于向标准输出输出线程组的信息，这个方法仅用于debug。从源码中，可以看出list方法会递归输出多级关联的所有信息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public void list() {
    list(System.out, 0);
}
void More list(PrintStream out, int indent) {
    int ngroupsSnapshot;
    ThreadGroup[] groupsSnapshot;
    synchronized (this) {
        for (int j = 0 ; j < indent ; j++) {
            out.print(" ");
        }
        out.println(this);
        indent += 4;
        for (int i = 0 ; i < nthreads ; i++) {
            for (int j = 0 ; j < indent ; j++) {
                out.print(" ");
            }
            out.println(threads[i]);
        }
        ngroupsSnapshot = ngroups;
        if (groups != null) {
            groupsSnapshot = Arrays.copyOf(groups, ngroupsSnapshot);
        } else {
            groupsSnapshot = null;
        }
    }
    // 此处递归
    for (int i = 0 ; i < ngroupsSnapshot ; i++) {
        groupsSnapshot[i].list(out, indent);
    }
}
enumerate()

enumerate方法的作用是枚举线程组的内容，把内容放入对应的容器中。enumerate 既可以枚举当前线程组的线程也可以枚举当前线程组的子线程组。enumerate()的递归与非递归控制是通过boolean值控制的。

1
2
public int enumerate(ThreadGroup list[], boolean recurse) {...}
public int enumerate(Thread list[], boolean recurse){...}

显然从源码中可以看出，当传参为true时，会进行递归操作。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
private int enumerate(Thread list[], int n, boolean recurse) {
    int ngroupsSnapshot = 0;
    ThreadGroup[] groupsSnapshot = null;
    synchronized (this) {
        if (destroyed) {
            return 0;
        }
        int nt = nthreads;
        if (nt > list.length - n) {
            nt = list.length - n;
        }
        for (int i = 0; i < nt; i++) {
            if (threads[i].isAlive()) {
                list[n++] = threads[i];
            }
        }
        if (recurse) {
            ngroupsSnapshot = ngroups;
            if (groups != null) {
                groupsSnapshot = Arrays.copyOf(groups, ngroupsSnapshot);
            } else {
                groupsSnapshot = null;
            }
        }
    }
    if (recurse) {
        for (int i = 0 ; i < ngroupsSnapshot ; i++) {
            n = groupsSnapshot[i].enumerate(list, n, true);
        }
    }
    return n;
}

例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
public class ThreadGroup2GroupDemo {

    public static class  myThread implements Runnable{

        public void run() {
            try{
                while (!Thread.currentThread().isInterrupted())
                {
                    Thread.sleep(300);
                }
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        /** 构建线程组结构
         *   head -> parentGroup -> p-A
         *                       -> p-B
         *                       -> childGroup -> c-a
         *                                     -> c-b
         *                                     -> childChildGroup -> cc-a
         */
        ThreadGroup parentGroup = new ThreadGroup("parentGroup");
        ThreadGroup childGroup = new ThreadGroup(parentGroup,"childGroup");
        ThreadGroup childChildGroup = new ThreadGroup(childGroup,"childChildGroup");

        Thread pThreadA = new Thread(parentGroup,new myThread(),"p-A");
        Thread pThreadB = new Thread(parentGroup,new myThread(),"p-B");
        Thread cThreada = new Thread(childGroup,new myThread(),"c-a");
        Thread cThreadb = new Thread(childGroup,new myThread(),"c-b");
        Thread ccThreada = new Thread(childChildGroup,new myThread(),"cc-a");


        pThreadA.start();
        pThreadB.start();
        cThreada.start();
        cThreadb.start();
        ccThreada.start();

        parentGroup.list();

        ThreadGroup[] list1 = new ThreadGroup[parentGroup.activeGroupCount()];
        System.out.println("递归获取线程组");
        parentGroup.enumerate(list1,true);
        for(int i =0;i<list1.length;i++){
            System.out.println(list1[i].getName());
        }
        ThreadGroup[] list2 = new ThreadGroup[parentGroup.activeGroupCount()];
        System.out.println("非递归获取线程组");
        parentGroup.enumerate(list2,false);
        for(int i =0;i<list2.length;i++){
            if(list2[i]!=null) {
                System.out.println(list2[i].getName());
            }
        }

        Thread[] list3 = new Thread[parentGroup.activeCount()];
        System.out.println("递归获取线程");
        parentGroup.enumerate(list3,true);
        for(int i =0;i<list3.length;i++){
            System.out.println(list3[i].getName());
        }
        Thread[] list4 = new Thread[parentGroup.activeCount()];
        System.out.println("非递归获取线程");
        parentGroup.enumerate(list4,false);
        for(int i =0;i<list4.length;i++){
            if(list4[i]!=null) {
                System.out.println(list4[i].getName());
            }
        }

        Thread.sleep(1000);
        // 中断所有线程
        parentGroup.interrupt();

    }
}
  • output
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
java.lang.ThreadGroup[name=parentGroup,maxpri=10]
    Thread[p-A,5,parentGroup]
    Thread[p-B,5,parentGroup]
    java.lang.ThreadGroup[name=childGroup,maxpri=10]
        Thread[c-a,5,childGroup]
        Thread[c-b,5,childGroup]
        java.lang.ThreadGroup[name=childChildGroup,maxpri=10]
            Thread[cc-a,5,childChildGroup]
递归获取线程组
childGroup
childChildGroup
非递归获取线程组
childGroup
递归获取线程
p-A
p-B
c-a
c-b
cc-a
非递归获取线程
p-A
p-B
java.lang.InterruptedException: sleep interrupted
    at java.lang.Thread.sleep(Native Method)
    at cn.victor.test.ThreadGroup2GroupDemo$myThread.run(ThreadGroup2GroupDemo.java:11)
    at java.lang.Thread.run(Thread.java:748)
java.lang.InterruptedException: sleep interrupted
    at java.lang.Thread.sleep(Native Method)
    at cn.victor.test.ThreadGroup2GroupDemo$myThread.run(ThreadGroup2GroupDemo.java:11)
    at java.lang.Thread.run(Thread.java:748)
java.lang.InterruptedException: sleep interrupted
    at java.lang.Thread.sleep(Native Method)
    at cn.victor.test.ThreadGroup2GroupDemo$myThread.run(ThreadGroup2GroupDemo.java:11)
    at java.lang.Thread.run(Thread.java:748)
java.lang.InterruptedException: sleep interrupted
    at java.lang.Thread.sleep(Native Method)
    at cn.victor.test.ThreadGroup2GroupDemo$myThread.run(ThreadGroup2GroupDemo.java:11)
    at java.lang.Thread.run(Thread.java:748)
java.lang.InterruptedException: sleep interrupted
    at java.lang.Thread.sleep(Native Method)
    at cn.victor.test.ThreadGroup2GroupDemo$myThread.run(ThreadGroup2GroupDemo.java:11)
    at java.lang.Thread.run(Thread.java:748)

Process finished with exit code 0

线程组特性

线程组自动归属特性

  • 自动归属是指对于初始化时没有指定所属线程组的Thread和ThreadGroup对象，他们会被自动归属到当前线程所属的线程组中。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
// openJdk 1.8  源码

//　ThreadGroup 不指定线程组的初始化方法
public ThreadGroup(String name) {
    this(Thread.currentThread().getThreadGroup(), name);
}

// Thread init 初始化方法
private void init(ThreadGroup g, Runnable target, String name,
                       long stackSize, AccessControlContext acc) {
    Thread parent = currentThread();
    SecurityManager security = System.getSecurityManager();
    if (g == null) {
        /* Determine if it's an applet or not */
        /* If there is a security manager, ask the security manager what to do. */
        if (security != null) {
            g = security.getThreadGroup();
            /** 
             * // In Class SecurityManager
             * public ThreadGroup getThreadGroup() {
             * return Thread.currentThread().getThreadGroup();
             * }
             */
        }
 
        /* If the security doesn't have a strong opinion of the matter use the parent thread group. */
        if (g == null) {
            g = parent.getThreadGroup();
        }
    }
    ... ...
}

根线程组

  • 根线程组就是系统线程组system，根线程组上没有父线程组。

查看OpenJdk的ThreadGroup源码，可以发现下面这段代码。这是用于C代码调用，来生成系统线程组的。

1
2
3
4
5
6
7
8
9
/**
 * Creates an empty Thread group that is not in any Thread group.
 * This method is used to create the system Thread group.
 */
private ThreadGroup() {     // called from C code
    this.name = "system";
    this.maxPriority = Thread.MAX_PRIORITY;
    this.parent = null;
}

未捕获异常的统一处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// openJdk 1.8 source code
public void uncaughtException(Thread t, Throwable e) {  
    if (parent != null) {  
        parent.uncaughtException(t, e);//父线程组不为空，设置到父线程组  
    } else {  
        Thread.UncaughtExceptionHandler ueh =   
        Thread.getDefaultUncaughtExceptionHandler();  
        if (ueh != null) { 
            // 默认异常处理器
            ueh.uncaughtException(t, e);  
        } else if (!(e instanceof ThreadDeath)) {  
            System.err.print("Exception in thread \"" + t.getName() + "\" ");  
            e.printStackTrace(System.err);  
        }  
    }  
}
  • 如果当前线程组有父线程组，调用父线程组的uncaughtException方法。
  • 如果父线程组不存在，但是有默认异常处理器，调用异常处理器的uncaughtException方法。
  • 如果没有指定异常处理器，输出异常日志。注意，如果异常是ThreadDeath，忽略该异常。

可以在子类中覆盖ThreadGroup的uncaughtException方法来统一处理线程组异常。

线程优先级

现代操作系统一般是分时操作系统，即一台计算机采用时间片轮转的方式同时为若干用户提供服务。线程分配得到的时间片的多少决定了线程使用处理器资源的多少，也对应了线程优先级这个概念。JAVA线程中使用一个int 值priority来控制优先级，范围为1-10,其中10的优先级最高，1的优先级最低。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
// 最小优先级
public final static int MIN_PRIORITY = 1;
// 默认优先级
public final static int NORM_PRIORITY = 5;
// 最大优先级
public final static int MAX_PRIORITY = 10;

// native method
private native void setPriority0(int newPriority);

// API
public final void setPriority(int newPriority) {
    ThreadGroup g;
    checkAccess();
    if (newPriority > MAX_PRIORITY || newPriority < MIN_PRIORITY) {
        throw new IllegalArgumentException();
    }
    if((g = getThreadGroup()) != null) {
        // 线程有限级不能比线程组优先级大
        if (newPriority > g.getMaxPriority()) {
            newPriority = g.getMaxPriority();
        }
        setPriority0(priority = newPriority);
    }
}

线程的优先级有继承性

在Java中，线程的优先级具有继承性，比如，A线程启动B线程，则B线程的优先级与A是一样。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public class ThreadPriority {

    public static class MyThread extends Thread{

        @Override
        public void run(){
            System.out.println("MyThread priority is :"+ this.getPriority());
        }
    }

    public static void main(String[] args){
        System.out.println("main Thread priority is:"+ Thread.currentThread().getPriority());
        Thread.currentThread().setPriority(8);
        System.out.println("After seted, main Thread priority is:"+ Thread.currentThread().getPriority());
        Thread myThread = new MyThread();
        myThread.start();
    }
}
/*
 * main Thread priority is:5
 * After seted, main Thread priority is:8
 * MyThread priority is :8
 */

优先级规则

  1. CPU尽量把执行资源让给优先级高的线程，即优先级高的线程总是大部分会先执行，但是这不代笔高优先级的线程全部先执行完再执行低优先级的线程。
  2. 优先级具有随机性，优先级高的不一定每次都先执行。

线程优先级不能作为程序正确性的依赖，因为操作系统可以完全不用理会JAVA线程对于优先级的设定。

分析例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
public class TestPriority
{
    private static volatile boolean notStart=true;
    private static volatile boolean notEnd=true;

    public static void main(String[] args) throws Exception
    {
        List<Job> jobs = new ArrayList<>();
        for(int i = 0;i<10;i++)
        {
            int priority = i<5?Thread.MIN_PRIORITY:Thread.MAX_PRIORITY;
            Job job=new Job(priority);
            jobs.add(job);
            Thread thread=new Thread(job, "Thread:"+i);
            thread.setPriority(priority);
            thread.start();

        }//使用这个循环启动了10个线程

        notStart=false;
        TimeUnit.SECONDS.sleep(10);//main线程沉睡10s，使得10个小线程执行结束
        notEnd=false;

        for(Job job:jobs)
        {
            System.out.println("JOB priority:"+job.priority+","+job.jobCount);
        }
    }
    static class Job implements Runnable
    {
        private int priority;
        private long jobCount;
        public Job(int priority)
        {
            this.priority=priority;
        }
        public void run()
        {
            while(notStart)
            {
                Thread.yield();//这里确保main线程将10个小线程启动成功
            }

            while(notEnd)
            {
                Thread.yield();//这里让出CPU资源，使得10个线程自由竞争。
                jobCount++;//记录竞争状态，反映线程的优先级。
            }
        }
    }
}

/** ------ out put ------
 * // OS: Ubuntu
 * JOB priority:1,20002078
 * JOB priority:1,21466360
 * JOB priority:1,20064881
 * JOB priority:1,21806089
 * JOB priority:1,17341995
 * JOB priority:10,21941482
 * JOB priority:10,15592763
 * JOB priority:10,23662996
 * JOB priority:10,17024857
 * JOB priority:10,17770879
 */

显然，根据输出我们可以发现Linux 操作系统并没有理会JAVA设置的优先级。

而在window系统上，我们可能得到这样的结果:

1
2
3
4
5
6
7
8
9
10
JOB priority:1,1099494
JOB priority:1,1097710
JOB priority:1,1099911
JOB priority:1,1100411
JOB priority:1,1099721
JOB priority:10,5208263
JOB priority:10,5198474
JOB priority:10,5213148
JOB priority:10,5184842
JOB priority:10,5172312

这就符合了线程的优先级规则。

守护(daemon)进程

Java的线程有两种类型: 用户进程和守护进程。

什么是守护进程?

守护进程是一种特殊的线程，当进程中不存在非守护线程的时候，则守护进程自动销毁，典型的守护进程是垃圾回收进程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
public class DaemonTest {

    public static class MyThread extends Thread {
        @Override
        public void run() {
            while (true) {
                System.out.println("a thread alive");
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Thread myThread = new MyThread();
        myThread.setDaemon(true);
        myThread.start();
        /**
         * 如果将　myThread.setDaemon(true)　放在start()方法后,
         * myThread 将被当做用户线程,程序会一直执行下去.
         */
        Thread.sleep(500);
    }
}
/** ----- out put -----
 * a thread alive
 * a thread alive
 * a thread alive
 * a thread alive
 * a thread alive
 * Process finished with exit code 0
 */
-------------本文结束感谢您的阅读-------------
坚持分享，您的支持将鼓励我继续创作！
0%