java线程之Exchanger

Exchanger的工作原理及实例

1.实现原理        

          Exchanger（交换者）是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。它提供一个同步点，在这个同步点两个线程可以交换彼此的数据。这两个线程通过exchange方法交换数据， 如果第一个线程先执行exchange方法，它会一直等待第二个线程也执行exchange，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。因此使用Exchanger的重点是成对的线程使用exchange()方法，当有一对线程达到了同步点，就会进行交换数据。因此该工具类的线程对象是成对的。

       Exchanger类提供了两个方法，String exchange(V x):用于交换，启动交换并等待另一个线程调用exchange；String exchange(V x,long timeout,TimeUnit unit)：用于交换，启动交换并等待另一个线程调用exchange，并且设置最大等待时间，当等待时间超过timeout便停止等待。

2.实例讲解

        通过以上的原理，可以知道使用Exchanger类的核心便是exchange()方法的使用，接下来通过一个例子来使的该工具类的用途更加清晰。该例子主要讲解的是前段时间NBA交易截止日的交易。

1.  
   package concurrent;
2.  
   import java.util.concurrent.Exchanger;
3.  
   import java.util.concurrent.ExecutorService;
4.  
   import java.util.concurrent.Executors;
5.  
   import java.util.*;
6.  
   public class ExchangerDemo {
7.  
    
8.  
   public static void main(String[] args) {
9.  
   ExecutorService executor = Executors.newCachedThreadPool();
10.  
     
11.  
    final Exchanger exchanger = new Exchanger();
12.  
    executor.execute(new Runnable() {
13.  
    String data1 = "克拉克森，小拉里南斯";
14.  
     
15.  
     
16.  
    @Override
17.  
    public void run() {
18.  
    nbaTrade(data1, exchanger);
19.  
    }
20.  
    });
21.  
     
22.  
     
23.  
    executor.execute(new Runnable() {
24.  
    String data1 = "格里芬";
25.  
     
26.  
    @Override
27.  
    public void run() {
28.  
    nbaTrade(data1, exchanger);
29.  
    }
30.  
    });
31.  
     
32.  
    executor.execute(new Runnable() {
33.  
    String data1 = "哈里斯";
34.  
     
35.  
    @Override
36.  
    public void run() {
37.  
    nbaTrade(data1, exchanger);
38.  
    }
39.  
    });
40.  
     
41.  
    executor.execute(new Runnable() {
42.  
    String data1 = "以赛亚托马斯，弗莱";
43.  
     
44.  
    @Override
45.  
    public void run() {
46.  
    nbaTrade(data1, exchanger);
47.  
    }
48.  
    });
49.  
     
50.  
    executor.shutdown();
51.  
    }
52.  
     
53.  
    private static void nbaTrade(String data1, Exchanger exchanger) {
54.  
    try {
55.  
    System.out.println(Thread.currentThread().getName() + "在交易截止之前把 " + data1 + " 交易出去");
56.  
    Thread.sleep((long) (Math.random() * 1000));
57.  
     
58.  
    String data2 = (String) exchanger.exchange(data1);
59.  
    System.out.println(Thread.currentThread().getName() + "交易得到" + data2);
60.  
    } catch (InterruptedException e) {
61.  
    e.printStackTrace();
62.  
    }
63.  
    }
64.  
    }

运行程序，得到如下结果：

1.  
   pool-1-thread-1在交易截止之前把 克拉克森，小拉里南斯 交易出去
2.  
   pool-1-thread-2在交易截止之前把 格里芬 交易出去
3.  
   pool-1-thread-3在交易截止之前把 哈里斯 交易出去
4.  
   pool-1-thread-4在交易截止之前把 以赛亚托马斯，弗莱 交易出去
5.  
   pool-1-thread-2交易得到哈里斯
6.  
   pool-1-thread-3交易得到格里芬
7.  
   pool-1-thread-4交易得到克拉克森，小拉里南斯
8.  
   pool-1-thread-1交易得到以赛亚托马斯，弗莱

        以上例子可以看出两个都调用exchange()方法的线程会进行交换数据。接下来假设线程数目只有奇数个，观察情况：

如以下代码，将第四个线程注释掉。

1.  
   package concurrent;
2.  
   import java.util.concurrent.Exchanger;
3.  
   import java.util.concurrent.ExecutorService;
4.  
   import java.util.concurrent.Executors;
5.  
   import java.util.*;
6.  
   public class ExchangerDemo {
7.  
    
8.  
   public static void main(String[] args) {
9.  
   ExecutorService executor = Executors.newCachedThreadPool();
10.  
     
11.  
    final Exchanger exchanger = new Exchanger();
12.  
    executor.execute(new Runnable() {
13.  
    String data1 = "克拉克森，小拉里南斯";
14.  
     
15.  
     
16.  
    @Override
17.  
    public void run() {
18.  
    nbaTrade(data1, exchanger);
19.  
    }
20.  
    });
21.  
     
22.  
     
23.  
    executor.execute(new Runnable() {
24.  
    String data1 = "格里芬";
25.  
     
26.  
    @Override
27.  
    public void run() {
28.  
    nbaTrade(data1, exchanger);
29.  
    }
30.  
    });
31.  
     
32.  
    executor.execute(new Runnable() {
33.  
    String data1 = "哈里斯";
34.  
     
35.  
    @Override
36.  
    public void run() {
37.  
    nbaTrade(data1, exchanger);
38.  
    }
39.  
    });
40.  
     
41.  
    // executor.execute(new Runnable() {
42.  
    // String data1 = "以赛亚托马斯，弗莱";
43.  
    //
44.  
    // @Override
45.  
    // public void run() {
46.  
    // nbaTrade(data1, exchanger);
47.  
    // }
48.  
    // });
49.  
     
50.  
    executor.shutdown();
51.  
    }
52.  
     
53.  
    private static void nbaTrade(String data1, Exchanger exchanger) {
54.  
    try {
55.  
    System.out.println(Thread.currentThread().getName() + "在交易截止之前把 " + data1 + " 交易出去");
56.  
    Thread.sleep((long) (Math.random() * 1000));
57.  
     
58.  
    String data2 = (String) exchanger.exchange(data1);
59.  
    System.out.println(Thread.currentThread().getName() + "交易得到" + data2);
60.  
    } catch (InterruptedException e) {
61.  
    e.printStackTrace();
62.  
    }
63.  
    }
64.  
    }

运行程序，得到如下结果：

1.  
   pool-1-thread-1在交易截止之前把 克拉克森，小拉里南斯 交易出去
2.  
   pool-1-thread-2在交易截止之前把 格里芬 交易出去
3.  
   pool-1-thread-3在交易截止之前把 哈里斯 交易出去
4.  
   pool-1-thread-3交易得到格里芬
5.  
   pool-1-thread-2交易得到哈里斯

        由结果可知，线程2和线程3进行了交换数据，而线程1一直等待与它交换数据的线程调用exchange，但是只有3个线程，所以会一直等待。

        因此，当两个线程之间出现数据交换的情况，可以使用Exchanger工具类实现数据交换。注意exchange方法的含义，以及触发数据交换的条件。

 

下面看代码例子：

package thread.exchanger;

import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExchangerTest {
	public static void main(String[] args) {
		ExecutorService pool = Executors.newCachedThreadPool();//线程缓存池
		final Exchanger<String> exchanger = new Exchanger<String>();//用于一对线程的交换
		
		pool.execute(new Runnable() {
			@Override
			public void run() {
				try {
					Thread.sleep((long)(Math.random()*10000));
					String data1 = "白粉";
					System.out.println("线程"+Thread.currentThread().getName()+"已经到达交换地点，准备拿"+data1+"交换！");
					data1 = exchanger.exchange(data1);//进行交换返回交换后的物品
					System.out.println("线程"+Thread.currentThread().getName()+"交换成功，拿到"+data1);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		});
		
		pool.execute(new Runnable() {
			@Override
			public void run() {
				try {
					Thread.sleep((long)(Math.random()*10000));
					String data2 = "钱";
					System.out.println("线程"+Thread.currentThread().getName()+"已经到达交换地点，准备拿"+data2+"交换！");
					data2 = exchanger.exchange(data2);//进行交换返回交换后的物品
					System.out.println("线程"+Thread.currentThread().getName()+"交换成功，拿到"+data2);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		});
		
		pool.shutdown();
	}
}

  

注意：exchanger只能用于一对或者一组线程的交换。交换的线程不能为奇数，否则会出现死锁等待。